#!/usr/bin/env python3 """NEXT Athlete OS ยท RC49 special prop trace verifier. Deterministic QA for RC49. Checks ordinary pose/anchor integrity plus new special prop/partner/throw gates. """ from __future__ import annotations import json, math, pathlib, datetime as dt from typing import Any, Dict, List ROOT = pathlib.Path(__file__).resolve().parents[1] DATA = ROOT / "assets" / "data" / "image-admin" QA = ROOT / "qa_screenshots" SEQ = DATA / "pose-atlas-exercise-sequences.json" PACKS = DATA / "pose-atlas-packs.json" RC49 = DATA / "rc49_special_prop_binding.json" OUT = DATA / "rc49_motion_trace_verification.json" QA_OUT = QA / "rc49_motion_trace_verification_report.json" def load(p: pathlib.Path) -> Any: return json.loads(p.read_text(encoding="utf-8")) def save(p: pathlib.Path, data: Any) -> None: p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def num(v: Any, fallback: float = 0) -> float: try: f = float(v) return f if math.isfinite(f) else fallback except Exception: return fallback def issue(level: str, code: str, message: str, exercise: str = "", pose: str = "", frame: int | None = None) -> Dict[str, Any]: out = {"level": level, "code": code, "message": message} if exercise: out["exerciseKey"] = exercise if pose: out["poseKey"] = pose if frame is not None: out["frame"] = frame return out def pose_index(packs: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: return {p.get("poseKey"): p for p in packs.get("poses", []) if p.get("poseKey")} def skel(pose: Dict[str, Any], aid: str) -> Dict[str, Any]: return (((pose.get("athletes") or {}).get(aid) or {}).get("metrics") or {}).get("skeleton") or {} def lm(pose: Dict[str, Any], aid: str, name: str) -> Dict[str, Any] | None: if name == "barTop": return {"x": 50, "y": 0, "confidence": 1, "source": "synthetic"} s = skel(pose, aid) return (s.get("runtimeLandmarks") or {}).get(name) or (s.get("landmarks") or {}).get(name) def normalize_anchor(name: str) -> str: aliases = {"feet": "footCenter", "bothFeet": "footCenter", "hands": "handCenter", "bar": "barTop", "knee": "kneeCenter", "dominant": "dominantContactFoot", "dominantFoot": "dominantContactFoot"} raw = str(name or "").strip() return aliases.get(raw, raw) def verify() -> Dict[str, Any]: seq_data = load(SEQ) packs = load(PACKS) rc49 = load(RC49) if RC49.exists() else {} sequences = seq_data.get("sequences", {}) pidx = pose_index(packs) athletes = [a.get("id") for a in packs.get("athletes", []) if a.get("id")] now = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") issues: List[Dict[str, Any]] = [] special = {"propBoundFrames": 0, "medballFrames": 0, "boxFrames": 0, "partnerFrames": 0, "throwPlaceholderFrames": 0, "manualPropLockedExercises": [], "reviewHoldExercises": []} counts = {"sequences": len(sequences), "frames": 0, "athletes": len(athletes), "missingPoseKeys": 0, "missingAthleteSprites": 0, "missingAnchors": 0, "activeMediapipeAnchorFrames": 0, "lowConfidenceActiveAnchors": 0, "pathTimingErrors": 0, "specialGateErrors": 0} traces: Dict[str, Any] = {} for key, seq in sequences.items(): poses = seq.get("poses", []) counts["frames"] += len(poses) prev_t = -1.0 gate = seq.get("latestMovementGate") or (seq.get("rc49SpecialPropBinding") or {}).get("runtimeGate") or "" if gate in ("manual-prop-locked", "manual-knee-pivot-locked", "manual-review-retained"): special["manualPropLockedExercises"].append(key) if gate == "review-hold-needs-true-pose-set": special["reviewHoldExercises"].append(key) frames = [] for i, entry in enumerate(poses): pk = entry.get("poseKey") if isinstance(entry, dict) else str(entry) pose = pidx.get(pk) anchor = normalize_anchor(entry.get("anchorLandmark") or (entry.get("limbIntent") or {}).get("primaryAnchor") or "footCenter") if isinstance(entry, dict) else "footCenter" t = num(entry.get("pathT"), i / max(1, len(poses) - 1)) if isinstance(entry, dict) else i / max(1, len(poses) - 1) if t + 1e-9 < prev_t: counts["pathTimingErrors"] += 1 issues.append(issue("error", "path_t_not_monotonic", f"pathT moved backwards {prev_t}->{t}", key, pk, i)) prev_t = max(prev_t, t) prop = entry.get("propBinding") if isinstance(entry, dict) else None if prop: special["propBoundFrames"] += 1 kind = prop.get("kind", "") if "medball" in kind: special["medballFrames"] += 1 if kind == "box": special["boxFrames"] += 1 if kind == "partner-ankle-lock": special["partnerFrames"] += 1 if "placeholder" in str(entry.get("phase", "")) or prop.get("runtimeGate") == "review-hold-needs-true-pose-set": special["throwPlaceholderFrames"] += 1 if not pose: counts["missingPoseKeys"] += 1 issues.append(issue("error", "missing_pose_key", "Frame references missing atlas pose", key, pk, i)) continue for aid in athletes: a = (pose.get("athletes") or {}).get(aid) if not a or not a.get("src"): counts["missingAthleteSprites"] += 1 issues.append(issue("error", "missing_athlete_sprite", f"Missing sprite for athlete {aid}", key, pk, i)) continue s = skel(pose, aid) active = s.get("mediaPipeRuntimeActive") is True if active: counts["activeMediapipeAnchorFrames"] += 1 anchor_lm = lm(pose, aid, anchor) if anchor_lm is None: counts["missingAnchors"] += 1 issues.append(issue("error", "missing_anchor", f"Missing {anchor} for {aid}", key, pk, i)) elif active: conf = num(anchor_lm.get("confidence"), num(anchor_lm.get("visibility"), 0)) if conf < 0.70 and anchor != "barTop": counts["lowConfidenceActiveAnchors"] += 1 issues.append(issue("error", "low_confidence_active_anchor", f"Active MediaPipe anchor {anchor} confidence {conf:.3f}", key, pk, i)) frames.append({"poseKey": pk, "phase": entry.get("phase") if isinstance(entry, dict) else "", "pathT": t, "anchor": anchor, "prop": bool(prop)}) traces[key] = {"gate": gate, "frames": frames} # Gate-specific assertions. for required in ["med-ball-slam", "box-jump", "seated-box-jump", "buddy-hamstring-curls", "chest-pass", "rotational-throw"]: if required not in sequences: counts["specialGateErrors"] += 1 issues.append(issue("error", "missing_special_sequence", "Required special sequence missing", required)) if sequences.get("chest-pass", {}).get("latestMovementGate") != "review-hold-needs-true-pose-set": counts["specialGateErrors"] += 1 issues.append(issue("error", "chest_pass_not_held", "Chest pass must remain review-held until true pose candidates exist", "chest-pass")) if sequences.get("rotational-throw", {}).get("latestMovementGate") != "review-hold-needs-true-pose-set": counts["specialGateErrors"] += 1 issues.append(issue("error", "rotational_throw_not_held", "Rotational throw must remain review-held until true pose candidates exist", "rotational-throw")) if sequences.get("med-ball-slam", {}).get("latestMovementGate") != "special-prop-active": counts["specialGateErrors"] += 1 issues.append(issue("error", "medball_not_prop_active", "Med-ball slam should be special-prop-active", "med-ball-slam")) summary = { "version": "v97-rc49-special-prop-binding", "createdAt": now, "sourceVersion": seq_data.get("version"), "counts": counts, "special": special, "rc49SequenceChanges": rc49.get("sequenceChanges"), "medballActivation": rc49.get("medballMediaPipeActivation", {}), "issues": { "error": sum(1 for x in issues if x["level"] == "error"), "warn": sum(1 for x in issues if x["level"] == "warn"), "info": sum(1 for x in issues if x["level"] == "info"), }, "pass": sum(1 for x in issues if x["level"] == "error") == 0, } out = {"summary": summary, "issues": issues, "traces": traces} return out def main() -> None: QA.mkdir(exist_ok=True) report = verify() save(OUT, report) save(QA_OUT, report) print(json.dumps(report["summary"], ensure_ascii=False, indent=2)) if __name__ == "__main__": main()