#!/usr/bin/env python3 """NEXT Athlete OS ยท RC48 motion trace verifier. Non-visual, deterministic QA for the pose-atlas runtime layer. This tool samples the current RC48 movement recipes after green expansion and checks whether anchors, contacts, path timing and review gates are internally consistent. """ from __future__ import annotations import json import math import pathlib import datetime as _dt from typing import Any, Dict, List, Tuple ROOT = pathlib.Path(__file__).resolve().parents[1] DATA = ROOT / "assets" / "data" / "image-admin" OUT = DATA / "rc48_motion_trace_verification.json" QA_OUT = ROOT / "qa_screenshots" / "rc48_motion_trace_verification_report.json" SEQUENCES_PATH = DATA / "pose-atlas-exercise-sequences.json" PACKS_PATH = DATA / "pose-atlas-packs.json" RC46_PATH = DATA / "rc46_selective_movement_rebuild.json" RC45_PATH = DATA / "rc45_ai_physio_first_pass.json" RC48_PATH = DATA / "rc48_green_motion_expansion.json" def load_json(path: pathlib.Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def num(value: Any, fallback: float = 0.0) -> float: try: v = float(value) return v if math.isfinite(v) else fallback except Exception: return fallback def norm_key(s: str) -> str: import re return re.sub(r"[^a-z0-9]+", "-", str(s or "").strip().lower()).strip("-") def ease_in_out(t: float) -> float: t = max(0.0, min(1.0, num(t, 0))) return 2 * t * t if t < 0.5 else 1 - ((-2 * t + 2) ** 2) / 2 def motion_phase_frames(profile: Dict[str, Any], count: int) -> List[float]: n = max(1, count) family = str(profile.get("family", "")) if n == 1: return [0.0] if "jump" in family: presets = { 4: [0, .22, .58, 1], 5: [0, .18, .50, .78, 1], 6: [0, .12, .28, .52, .78, 1], 7: [0, .10, .22, .40, .58, .78, 1], } if n in presets: return presets[n] if family == "fixed_bar_sequence": presets = {4: [0, .26, .58, 1], 5: [0, .18, .44, .68, 1]} if n in presets: return presets[n] if family == "fixed_knee_pivot_sequence": presets = {3: [0, .52, 1], 4: [0, .34, .68, 1]} if n in presets: return presets[n] if family == "hinge_slam_lift": presets = { 4: [0, .28, .70, 1], 5: [0, .20, .46, .76, 1], 6: [0, .16, .34, .58, .82, 1], 7: [0, .12, .25, .42, .62, .82, 1], } if n in presets: return presets[n] return [i / (n - 1) for i in range(n)] def path_frame_times(profile: Dict[str, Any], count: int, explicit_frames: List[float | None]) -> List[float]: n = max(1, count) if len(explicit_frames) == n and all(v is not None and math.isfinite(float(v)) for v in explicit_frames): prev = 0.0 out: List[float] = [] for i, v in enumerate(explicit_frames): raw = max(0.0, min(1.0, num(v, i / max(1, n - 1)))) cur = raw if i == 0 else max(prev, raw) out.append(cur) prev = cur return out return motion_phase_frames(profile, n) def path_point(profile: Dict[str, Any], t: float, floor_y: float, contact: str) -> Dict[str, float]: floor = max(58.0, min(94.0, num(floor_y, 84))) family = str(profile.get("family", "stationary_vertical")) e = ease_in_out(t) x, y, scale, rotation = 50.0, floor, 1.0, 0.0 x0 = max(4, min(96, num(profile.get("x0"), 26 if family == "linear_ground" else 50))) x1 = max(4, min(96, num(profile.get("x1"), 74 if family == "linear_ground" else 50))) arc_height = max(0, min(34, num(profile.get("arcHeight"), 18 if "jump" in family else 0))) settle = num(profile.get("settle"), 3 if "jump" in family else 0) if family == "linear_ground": x = x0 + (x1 - x0) * e y = floor elif family in ("jump_arc_forward", "jump_arc_lateral"): x = x0 + (x1 - x0) * e y = floor - arc_height * math.sin(math.pi * t) if t < .16: y += settle * (1 - t / .16) if t > .82: y += settle * ((t - .82) / .18) scale = .985 if t > .82 else 1 elif family == "jump_arc_vertical": x = x0 y = floor - arc_height * math.sin(math.pi * t) if t < .16: y += settle * (1 - t / .16) if t > .82: y += settle * ((t - .82) / .18) elif family in ("ground_extend_return", "hinge_slam_lift", "fixed_knee_pivot_sequence", "fixed_anchor_pivot"): x = x0 y = floor if family == "hinge_slam_lift" and .44 < t < .72: y += 2.2 * math.sin(math.pi * ((t - .44) / .28)) elif family in ("fixed_bar_sequence", "fixed_bar_vertical_pull"): x = x0 y = max(2, min(9, num(profile.get("barY"), 5))) c = str(contact or "").lower() if "jump" in family and c in ("grounded", "ground/load", "land/reset", "takeoff"): if c in ("grounded", "ground/load", "land/reset"): y = floor if c == "takeoff": y = min(floor - 1, max(floor - 4, y)) return {"x": round(x, 3), "y": round(y, 3), "scale": round(scale, 4), "rotation": round(rotation, 3)} def motion_contact(profile: Dict[str, Any], index: int, count: int, entry: Dict[str, Any]) -> str: if entry.get("contact"): return str(entry["contact"]) family = str(profile.get("family", "")) if "jump" in family: t = motion_phase_frames(profile, count)[index] if count > 1 else 0 if .30 < t < .74: return "airborne" if t <= .22: return "ground/load" return "land/reset" if family in ("fixed_bar_sequence", "fixed_bar_vertical_pull"): return "bar-anchor" if family in ("fixed_knee_pivot_sequence", "fixed_anchor_pivot"): return "fixed-pivot" if family == "ground_extend_return": return "hands/feet-ground" return "grounded" def make_pose_index(packs: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: return {p["poseKey"]: p for p in packs.get("poses", [])} def skeleton_for(pose: Dict[str, Any], athlete_id: str) -> Dict[str, Any]: return pose.get("athletes", {}).get(athlete_id, {}).get("metrics", {}).get("skeleton", {}) def runtime_lm(pose: Dict[str, Any], athlete_id: str, name: str) -> Dict[str, Any] | None: sk = skeleton_for(pose, athlete_id) if name == "barTop": return {"x": 50, "y": 0, "confidence": 1, "source": "synthetic-bar-top"} return sk.get("runtimeLandmarks", {}).get(name) or sk.get("landmarks", {}).get(name) def normalize_anchor_name(name: str) -> str: aliases = { "primary": "primaryContactFoot", "primaryFoot": "primaryContactFoot", "dominant": "dominantContactFoot", "dominantFoot": "dominantContactFoot", "dominantContact": "dominantContactFoot", "bothFeet": "footCenter", "feet": "footCenter", "handsFeet": "supportCenter", "hands": "handCenter", "bar": "barTop", "knee": "kneeCenter", } raw = str(name or "").strip() return aliases.get(raw, raw) def intended_anchor(entry: Dict[str, Any], profile: Dict[str, Any], contact: str) -> str: explicit = normalize_anchor_name(entry.get("anchorLandmark") or entry.get("anchorPoint") or profile.get("anchorLandmark") or "") if explicit: return explicit intent = entry.get("limbIntent") or {} intended = normalize_anchor_name(intent.get("primaryAnchor") or intent.get("anchorLandmark") or intent.get("anchor") or "") if intended: return intended family = str(profile.get("family", "")) anchor = str(profile.get("anchor", "")) phase = str(entry.get("phase", "")).lower() c = str(contact or entry.get("contact", "")).lower() if family in ("fixed_bar_sequence", "fixed_bar_vertical_pull"): return "barTop" if family in ("fixed_knee_pivot_sequence", "fixed_anchor_pivot"): return "kneeCenter" if family == "ground_extend_return" or anchor == "rear-foot-lock": return "rearFoot" if anchor == "hands-feet" or "hands" in c or "feet-ground" in c: return "supportCenter" if "takeoff" in c or "land" in c or "land" in phase or "takeoff" in phase: return "dominantContactFoot" if anchor == "feet" and family == "linear_ground": return "dominantContactFoot" return "footCenter" 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 verify() -> Dict[str, Any]: seq_data = load_json(SEQUENCES_PATH) packs = load_json(PACKS_PATH) rc46 = load_json(RC46_PATH) if RC46_PATH.exists() else {} rc45 = load_json(RC45_PATH) if RC45_PATH.exists() else {} sequences = seq_data.get("sequences", {}) pose_index = make_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]] = [] traces: Dict[str, Any] = {} counts = { "sequences": len(sequences), "frames": 0, "athletes": len(athletes), "missingPoseKeys": 0, "missingAthleteSprites": 0, "missingAnchors": 0, "pathTimingErrors": 0, "jumpContactErrors": 0, "reviewHoldExercises": 0, "activeMediapipeAnchorFrames": 0, "lowConfidenceActiveAnchors": 0, } critical = [ "lateral-line-hops", "pull-up", "chin-up", "buddy-hamstring-curls", "box-jump", "seated-box-jump", "chest-pass", "rotational-throw", ] for exercise_key, seq in sequences.items(): poses = [e if isinstance(e, dict) else {"poseKey": e} for e in seq.get("poses", [])] counts["frames"] += len(poses) profile = seq.get("motionProfile") or {} family = str(profile.get("family", "")) layout = {**seq_data.get("defaultLayout", {}), **seq.get("layout", {})} floor_y = num(layout.get("floorY"), 84) explicit = [num(e.get("pathT"), math.nan) if "pathT" in e else None for e in poses] frames = path_frame_times(profile, len(poses), explicit) trace_frames = [] # raw monotonic check before runtime correction finite_explicit = [v for v in explicit if v is not None and math.isfinite(v)] if len(finite_explicit) == len(poses): for i in range(1, len(finite_explicit)): if finite_explicit[i] < finite_explicit[i-1]: counts["pathTimingErrors"] += 1 issues.append(issue("error", "PATH_T_NOT_MONOTONIC", "Explicit pathT goes backwards before runtime correction.", exercise_key, poses[i].get("poseKey", ""), i + 1)) if seq.get("rc46ReviewHold") or seq.get("rc46MovementRebuild", {}).get("runtimeGate") == "review-hold": counts["reviewHoldExercises"] += 1 issues.append(issue("info", "REVIEW_HOLD", "Exercise is intentionally held because current image sequence is not the movement truth.", exercise_key)) for i, entry in enumerate(poses): pose_key = str(entry.get("poseKey", "")) pose = pose_index.get(pose_key) if not pose: counts["missingPoseKeys"] += 1 issues.append(issue("error", "MISSING_POSE_KEY", "Sequence references a poseKey that is not in pose-atlas-packs.json.", exercise_key, pose_key, i + 1)) continue contact = motion_contact(profile, i, len(poses), entry) anchor = intended_anchor(entry, profile, contact) point = path_point(profile, frames[i] if i < len(frames) else 0, floor_y, contact) athlete_rows = [] for aid in athletes: athlete = pose.get("athletes", {}).get(aid) if not athlete or not athlete.get("src"): counts["missingAthleteSprites"] += 1 issues.append(issue("error", "MISSING_ATHLETE_SPRITE", f"Missing sprite for {aid}.", exercise_key, pose_key, i + 1)) continue src = ROOT / athlete["src"] if not src.exists(): counts["missingAthleteSprites"] += 1 issues.append(issue("error", "MISSING_SPRITE_FILE", f"Sprite file not found for {aid}: {athlete['src']}", exercise_key, pose_key, i + 1)) lm = runtime_lm(pose, aid, anchor) if not lm and anchor == "dominantContactFoot": lm = runtime_lm(pose, aid, "primaryContactFoot") or runtime_lm(pose, aid, "footCenter") if not lm: counts["missingAnchors"] += 1 issues.append(issue("error", "MISSING_ANCHOR_LANDMARK", f"Missing anchor {anchor} for {aid}.", exercise_key, pose_key, i + 1)) athlete_rows.append({"athleteId": aid, "anchor": anchor, "ok": False}) continue source = str(lm.get("source", "")) conf = num(lm.get("confidence", lm.get("visibility", 1)), 1) mp_active = source.startswith("mediapipe") if mp_active: counts["activeMediapipeAnchorFrames"] += 1 if mp_active and conf < .55: counts["lowConfidenceActiveAnchors"] += 1 issues.append(issue("warning", "LOW_CONFIDENCE_MEDIAPIPE_ANCHOR", f"Active MediaPipe anchor {anchor} for {aid} has confidence {conf:.3f}.", exercise_key, pose_key, i + 1)) athlete_rows.append({"athleteId": aid, "anchor": anchor, "ok": True, "x": round(num(lm.get("x"), 0), 2), "y": round(num(lm.get("y"), 0), 2), "confidence": round(conf, 3), "source": source}) if "jump" in family: if contact in ("grounded", "ground/load", "land/reset") and abs(point["y"] - floor_y) > .01: counts["jumpContactErrors"] += 1 issues.append(issue("error", "JUMP_GROUNDED_NOT_ON_FLOOR", f"Ground contact frame y={point['y']} but floor={floor_y}.", exercise_key, pose_key, i + 1)) if contact == "airborne" and point["y"] >= floor_y - 2: counts["jumpContactErrors"] += 1 issues.append(issue("error", "JUMP_AIRBORNE_TOO_LOW", f"Airborne frame not visibly above floor: y={point['y']} floor={floor_y}.", exercise_key, pose_key, i + 1)) trace_frames.append({ "frame": i + 1, "poseKey": pose_key, "phase": entry.get("phase", ""), "pathT": round(frames[i] if i < len(frames) else 0, 3), "contact": contact, "anchor": anchor, "pathPoint": point, "athletes": athlete_rows, }) gate = seq.get("rc46MovementRebuild", {}).get("runtimeGate", "") status = seq.get("rc46MovementRebuild", {}).get("status", "") traces[exercise_key] = { "label": seq.get("label", exercise_key), "family": family, "anchor": profile.get("anchor", ""), "runtimeGate": gate, "status": status, "reviewHold": bool(seq.get("rc46ReviewHold") or gate == "review-hold"), "frames": trace_frames, } # policy assertions for critical movement rebuild lh = traces.get("lateral-line-hops", {}) if lh: anchors = [f.get("anchor") for f in lh.get("frames", [])] if not anchors or any(a != "footCenter" for a in anchors): issues.append(issue("error", "LATERAL_HOPS_NOT_FOOT_PAIR", "Lateral line hops must use footCenter for every keyframe." , "lateral-line-hops")) if lh.get("anchor") != "foot-pair-lock": issues.append(issue("error", "LATERAL_HOPS_WRONG_PROFILE_ANCHOR", "Lateral line hops motionProfile.anchor must be foot-pair-lock.", "lateral-line-hops")) for key in ("pull-up", "chin-up"): tr = traces.get(key, {}) if tr and any(f.get("anchor") != "barTop" for f in tr.get("frames", [])): issues.append(issue("error", "BAR_SEQUENCE_NOT_BAR_LOCKED", "Bar sequence must use barTop for every keyframe.", key)) tr = traces.get("buddy-hamstring-curls", {}) if tr and any(f.get("anchor") != "kneeCenter" for f in tr.get("frames", [])): issues.append(issue("error", "HAMSTRING_CURL_NOT_KNEE_LOCKED", "Buddy hamstring curl must use kneeCenter for every keyframe.", "buddy-hamstring-curls")) for key in ("chest-pass", "rotational-throw"): tr = traces.get(key, {}) if tr and not tr.get("reviewHold"): issues.append(issue("error", "THROW_SEQUENCE_NOT_HELD", "Throw sequence should remain review-hold until true throw images exist.", key)) level_counts = {"error": 0, "warning": 0, "info": 0} for it in issues: level_counts[it["level"]] = level_counts.get(it["level"], 0) + 1 safe_to_continue = level_counts["error"] == 0 report = { "version": "v97-rc48-motion-trace-verifier", "createdAt": now, "status": "non-visual-runtime-trace-qa", "runtimeChangedInRc48": True, "visualTestPerformed": True, "visualTestRequiredBeforeAcceptance": False, "safeToContinueWithoutVisualTest": safe_to_continue, "principle": "RC48 green expansion after user visual smoke-test: deterministic trace QA confirms anchors/timing after broader low-risk activation.", "sourceVersions": { "poseAtlasSequences": seq_data.get("version", ""), "poseAtlasPacks": packs.get("version", ""), "rc46Rebuild": rc46.get("version", ""), "rc45PhysioFirstPass": rc45.get("version", ""), "rc48Expansion": load_json(RC48_PATH).get("version", "") if RC48_PATH.exists() else "", }, "summary": {**counts, "issues": level_counts}, "criticalExerciseSummary": {k: traces.get(k) for k in critical if k in traces}, "issues": issues, "traces": traces, "nextGate": "Deployable as RC48. Next step should focus on held special movements: throws, box props, partner/knee and low-confidence floor patterns.", } OUT.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") QA_OUT.parent.mkdir(parents=True, exist_ok=True) QA_OUT.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") return report if __name__ == "__main__": r = verify() print(json.dumps({"version": r["version"], "safe": r["safeToContinueWithoutVisualTest"], "summary": r["summary"], "output": str(OUT)}, ensure_ascii=False, indent=2))