PROCITY/pipeline/clips_verify.py

#!/usr/bin/env python3
"""PROCITY R41 §41.1 — gate: motion_manifest.json <-> web/models/clips/*.glb integrity.

    python3 pipeline/clips_verify.py                    # 0 errors = green
    python3 pipeline/clips_verify.py --stage <dir>      # + quantisation fidelity vs the float source

Checks, all measured from the bytes (Lane F: this is the "every referenced clip resolves" gate):
  1. every group file named in the manifest exists, and its byte size matches the recorded one
  2. every clip id resolves to a NAMED animation inside its group GLB, and nothing extra rides along
  3. every group GLB is ZERO-DRAW — 0 meshes, 0 materials, 0 images, no Draco
  4. every group carries the same 66-node mixamorig skeleton (65 joints + Armature)
  5. every animation channel targets a node that exists, on a path the manifest declares
  6. recorded duration matches the max keyframe time in the file
  7. `_rotOnly` survivability: every clip keeps >=1 non-Hips quaternion track, i.e. the clip still
     says something after web/js/citizens/rigs.js filters it
  8. with --stage: max angular error of the int16 rotation quantisation vs the float32 source
"""
import argparse
import json
import math
import os
import struct
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from clips_pack import acc_floats, canon_name, glb_read      # noqa: E402


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--manifest", default="web/assets/motion_manifest.json")
    ap.add_argument("--clips", default="web/models/clips")
    ap.add_argument("--stage", help="staging dir of per-clip float32 GLBs (fidelity check)")
    a = ap.parse_args()
    man = json.load(open(a.manifest))
    err, warn = [], []
    total = 0
    skel = None

    for gfile, g in man["groups"].items():
        p = os.path.join(a.clips, gfile)
        if not os.path.exists(p):
            err.append("%s: MISSING" % gfile)
            continue
        size = os.path.getsize(p)
        total += size
        if size != g["bytes"]:
            err.append("%s: size %d != manifest %d" % (gfile, size, g["bytes"]))
        js, bin_ = glb_read(p)
        if js.get("meshes") or js.get("materials") or js.get("images"):
            err.append("%s: NOT zero-draw (meshes=%d materials=%d images=%d)"
                       % (gfile, len(js.get("meshes", [])), len(js.get("materials", [])),
                          len(js.get("images", []))))
        if any("draco" in e.lower() for e in js.get("extensionsUsed", [])):
            err.append("%s: Draco present" % gfile)
        names = [canon_name(n.get("name")) for n in js["nodes"]]
        if skel is None:
            skel = names
        elif names != skel:
            err.append("%s: skeleton differs from %s" % (gfile, list(man["groups"])[0]))
        joints = [n for n in names if n.startswith("mixamorig:")]
        if len(names) != 66 or len(joints) != 65:
            err.append("%s: expected 66 nodes / 65 mixamorig joints, got %d / %d"
                       % (gfile, len(names), len(joints)))
        anims = {an.get("name"): an for an in js.get("animations", [])}
        if sorted(anims) != sorted(g["clips"]):
            err.append("%s: animations %s != manifest clips %s"
                       % (gfile, sorted(anims), sorted(g["clips"])))
        for cid in g["clips"]:
            if cid not in anims:
                continue
            m = man["clips"][cid]
            if m["group"] != gfile:
                err.append("%s: clips[%s].group=%s" % (gfile, cid, m["group"]))
            an = anims[cid]
            dur, keeps = 0.0, 0
            for ch in an["channels"]:
                ni = ch["target"]["node"]
                if not (0 <= ni < len(names)):
                    err.append("%s/%s: channel targets node %d out of range" % (gfile, cid, ni))
                    continue
                if ch["target"]["path"] != "rotation" and man["_tracks"] == "rotation-only":
                    err.append("%s/%s: %s channel in a rotation-only build"
                               % (gfile, cid, ch["target"]["path"]))
                if ch["target"]["path"] == "rotation" and not names[ni].endswith(":Hips"):
                    keeps += 1
                t = acc_floats(js, bin_, an["samplers"][ch["sampler"]]["input"])
                if t:
                    dur = max(dur, t[-1][0])
            if abs(dur - m["duration"]) > 1e-3:
                err.append("%s/%s: duration %.4f != manifest %.4f" % (gfile, cid, dur, m["duration"]))
            if keeps == 0:
                err.append("%s/%s: nothing survives rigs.js `_rotOnly`" % (gfile, cid))
            if len(an["channels"]) != m["channels"]:
                err.append("%s/%s: %d channels != manifest %d"
                           % (gfile, cid, len(an["channels"]), m["channels"]))

    if total != man["totalBytes"]:
        err.append("totalBytes %d != measured %d" % (man["totalBytes"], total))
    if len(man["groups"]) > 6:
        err.append("ruling 6: %d group files > 6" % len(man["groups"]))
    if man["clipCount"] != len(man["clips"]):
        err.append("clipCount mismatch")

    # 8. quantisation fidelity against the float32 staging GLBs
    if a.stage:
        worst, worst_id = 0.0, None
        for cid, m in man["clips"].items():
            sp = os.path.join(a.stage, "%s.glb" % cid)
            if not os.path.exists(sp):
                warn.append("no staged source for %s" % cid)
                continue
            sj, sb = glb_read(sp)
            gj_, gb = glb_read(os.path.join(a.clips, m["group"]))
            snames = [canon_name(n.get("name")) for n in sj["nodes"]]
            gnames = [canon_name(n.get("name")) for n in gj_["nodes"]]
            ga = next(x for x in gj_["animations"] if x.get("name") == cid)
            gmap = {gnames[c["target"]["node"]]: c for c in ga["channels"]
                    if c["target"]["path"] == "rotation"}
            for ch in sj["animations"][0]["channels"]:
                if ch["target"]["path"] != "rotation":
                    continue
                bone = snames[ch["target"]["node"]]
                if bone not in gmap:
                    continue
                q0 = acc_floats(sj, sb, sj["animations"][0]["samplers"][ch["sampler"]]["output"])
                q1 = acc_floats(gj_, gb, ga["samplers"][gmap[bone]["sampler"]]["output"])
                if len(q0) != len(q1):
                    err.append("%s/%s: keyframe count %d != %d" % (cid, bone, len(q0), len(q1)))
                    continue
                for x, y in zip(q0, q1):
                    if man["_encoding"].startswith("rotation outputs are int16"):
                        y = tuple(max(v / 32767.0, -1.0) for v in y)
                    d = abs(sum(p * q for p, q in zip(x, y)))
                    d /= (math.sqrt(sum(p * p for p in x)) * math.sqrt(sum(q * q for q in y)) or 1)
                    ang = math.degrees(2 * math.acos(max(-1.0, min(1.0, d))))
                    if ang > worst:
                        worst, worst_id = ang, "%s/%s" % (cid, bone)
        print("quantisation: worst angular error %.6f deg (%s)" % (worst, worst_id))
        if worst > 0.05:
            err.append("quantisation error %.4f deg exceeds 0.05" % worst)

    print("%d clips / %d groups / %d bytes (%.2f MB)"
          % (man["clipCount"], len(man["groups"]), total, total / 1048576.0))
    for w in warn:
        print("WARN  %s" % w)
    for e in err:
        print("ERROR %s" % e)
    print("clips_verify: %s (%d errors, %d warnings)"
          % ("GREEN" if not err else "RED", len(err), len(warn)))
    return 1 if err else 0


if __name__ == "__main__":
    sys.exit(main())