#!/usr/bin/env python3 """NEXT Athlete OS ยท RC48 green motion expansion. Builds on RC46/RC47 after user visual smoke test was OK. Principle: expand only to low-risk/high-confidence motion families and keep prop/partner/floor/throw edge cases gated. Exercise movement truth remains the source of authority; pose/sheet names are not authoritative. """ from __future__ import annotations import json from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Set ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "assets" / "data" / "image-admin" QA = ROOT / "qa_screenshots" SEQ_PATH = DATA / "pose-atlas-exercise-sequences.json" PACKS_PATH = DATA / "pose-atlas-packs.json" RC46_PATH = DATA / "rc46_selective_movement_rebuild.json" RC47_PATH = DATA / "rc47_motion_trace_verification.json" OUT_DATA = DATA / "rc48_green_motion_expansion.json" OUT_REPORT = QA / "rc48_green_motion_expansion_report.json" BACKUP_SEQ = DATA / "pose-atlas-exercise-sequences.rc47-before-rc48.backup.json" BACKUP_PACKS = DATA / "pose-atlas-packs.rc47-before-rc48.backup.json" RC48_VERSION = "v97-rc48-green-motion-expansion" NOW = datetime.now(timezone.utc).isoformat(timespec="seconds") def load_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def save_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def limb(anchor: str, *, role: str, confidence: float = 0.99, source: str = "rc48-green-motion-expansion") -> dict: return { "schema": "rc48-movement-intent-v1", "primaryAnchor": anchor, "anchorSource": source, "movementRole": role, "confidence": confidence, "linkedToSpriteLandmarks": True, "coordinateSpace": "sprite-percent-unflipped", "runtimeTransformPolicy": "skeleton and sprite share the same anchor + flip transform", "note": "Exercise movement truth controls anchor/fas. Pose/sheet names are not authoritative.", } def tag_seq(seq: Dict[str, Any], status: str, decision: str, *, risk: str = "low-moderate", gate: str = "green-active") -> None: seq["rc48GreenExpansion"] = { "version": RC48_VERSION, "status": status, "decision": decision, "riskTier": risk, "runtimeGate": gate, "source": "RC47 visual smoke-test OK + RC46 selective rebuild + RC47 trace verifier", "poseNameAuthority": False, "updatedAt": NOW, } # Keep RC46 metadata if present, but expose a common latest gate for tools. seq["latestMovementGate"] = gate def rewrite_entry(entry: Dict[str, Any], *, phase: str | None = None, path_t: float | None = None, contact: str | None = None, anchor: str | None = None, role: str | None = None) -> None: if phase is not None: entry["phase"] = phase if path_t is not None: entry["pathT"] = round(float(path_t), 3) if contact is not None: entry["contact"] = contact if anchor is not None: entry["anchorLandmark"] = anchor anchor_name = entry.get("anchorLandmark") or anchor or "footCenter" entry["limbIntent"] = limb(anchor_name, role=role or entry.get("phase") or "keyframe") entry["rc48Frame"] = { "phase": entry.get("phase"), "pathT": entry.get("pathT"), "contact": entry.get("contact"), "anchorLandmark": anchor_name, "movementRole": role or entry.get("phase") or "keyframe", "reviewNote": "Green expansion after visual smoke-test; low-risk/high-confidence movement family only.", } def spread(n: int, preset: List[float] | None = None) -> List[float]: if preset and len(preset) == n: return preset if n <= 1: return [0.0] return [i / (n - 1) for i in range(n)] def tighten_sequences(sequences: Dict[str, Any]) -> List[Dict[str, Any]]: changes: List[Dict[str, Any]] = [] def add_change(key: str, action: str, impact: str = "active sequence metadata/runtime landmarks expanded") -> None: changes.append({"exerciseKey": key, "action": action, "runtimeImpact": impact}) # Vertical/elastic jump family: same foot-pair identity all the way through. vertical_specs = { "linear-pogos": {"loopMs": 1320, "arcHeight": 9, "phases": ["ready", "ankle-load", "quick-takeoff", "small-air", "elastic-land", "reset"], "status": "green-active-tightened"}, "squat-jump": {"loopMs": 1650, "arcHeight": 15, "phases": ["ready", "load-squat", "takeoff", "airborne", "land-soft", "reset"], "status": "green-active-tightened"}, } for key, cfg in vertical_specs.items(): seq = sequences.get(key) if not seq: continue poses = seq.get("poses", []) ts = spread(len(poses), [0, .12, .28, .52, .78, 1] if len(poses) == 6 else None) for i, entry in enumerate(poses): ph = cfg["phases"][i] if i < len(cfg["phases"]) else f"vertical-{i+1}" contact = "airborne" if "airborne" in ph else ("takeoff" if "takeoff" in ph else ("land/reset" if "land" in ph else "grounded")) rewrite_entry(entry, phase=ph, path_t=ts[i], contact=contact, anchor="footCenter", role="bilateral foot-pair vertical jump / no foot identity swap") seq.setdefault("motionProfile", {}).update({"family": "jump_arc_vertical", "anchor": "foot-pair-lock", "x0": 50, "x1": 50, "arcHeight": cfg["arcHeight"], "settle": 1.4, "loopMs": cfg["loopMs"], "ghostCount": 0, "ghostMode": "none", "debug": True}) seq.setdefault("layout", {}).update({"floorY": 84}) tag_seq(seq, cfg["status"], "Foot-pair lock + MediaPipe activation expanded after RC47 visual smoke-test.") add_change(key, "foot-pair vertical jump/pogo tightened") # Lateral pogo shares same atlas family as lateral-line-hops, but should be a bilateral side bounce, not random foot picking. if "lateral-pogo-jumps" in sequences: seq = sequences["lateral-pogo-jumps"] poses = seq.get("poses", []) ts = spread(len(poses), [0, .24, .58, 1] if len(poses) == 4 else None) phases = ["ready-left-foot-pair", "takeoff-foot-pair", "airborne-foot-pair", "land-right-foot-pair"] for i, entry in enumerate(poses): ph = phases[i] if i < len(phases) else f"lateral-pogo-{i+1}" contact = "airborne" if "airborne" in ph else ("takeoff" if "takeoff" in ph else ("land/reset" if "land" in ph else "grounded")) rewrite_entry(entry, phase=ph, path_t=ts[i], contact=contact, anchor="footCenter", role="bilateral lateral pogo / feet remain a visual pair") seq.setdefault("motionProfile", {}).update({"family": "jump_arc_lateral", "anchor": "foot-pair-lock", "x0": 42, "x1": 58, "arcHeight": 9, "settle": 1.4, "loopMs": 1450, "direction": "right", "ghostCount": 0, "ghostMode": "none", "debug": True}) seq.setdefault("layout", {}).update({"floorY": 84}) tag_seq(seq, "green-active-tightened", "Lateral pogo now follows the same bilateral foot-pair principle as lateral-line-hops.") add_change("lateral-pogo-jumps", "bilateral lateral pogo foot-pair logic added") # Squat family: bilateral feet, hip/knee truth; no fake travel. for key, phases in { "bodyweight-squat": ["stand", "load", "bottom-control", "return-stand"], "goblet-squat": ["goblet-stand", "goblet-load", "goblet-bottom", "goblet-drive", "goblet-reset"], }.items(): seq = sequences.get(key) if not seq: continue poses = seq.get("poses", []) ts = spread(len(poses)) for i, entry in enumerate(poses): ph = phases[i] if i < len(phases) else f"squat-{i+1}" rewrite_entry(entry, phase=ph, path_t=ts[i], contact="grounded", anchor="footCenter", role="bilateral squat: feet planted, knees/hips define depth") seq.setdefault("motionProfile", {}).update({"family": "stationary_vertical", "anchor": "feet", "x0": 50, "x1": 50, "loopMs": 1900 if key == "bodyweight-squat" else 2100, "ghostCount": 0, "ghostMode": "none", "debug": True}) seq.setdefault("layout", {}).update({"floorY": 84}) tag_seq(seq, "green-active-tightened", "Squat pattern uses MediaPipe geometry for feet/hip/knee alignment; object load stays visual/manual for goblet.") add_change(key, "squat hip/knee/foot pattern tightened") # Footwork family that passed smoke-test: carioca remains crossover, shuffle remains non-crossover. footwork = { "carioca-shuffle": {"anchor": "dominantContactFoot", "role": "crossover gait contact; each step has intentional foot identity", "phases": ["ready", "cross-front", "open", "cross-behind", "reset"], "profile_anchor": "crossover-foot-contact"}, "lateral-power-shuffle": {"anchor": "footCenter", "role": "non-crossover lateral base-of-support", "phases": ["athletic-ready", "push", "slide", "catch", "push-return", "reset"], "profile_anchor": "foot-pair-base"}, } for key, cfg in footwork.items(): seq = sequences.get(key) if not seq: continue poses = seq.get("poses", []) ts = spread(len(poses)) for i, entry in enumerate(poses): ph = cfg["phases"][i] if i < len(cfg["phases"]) else f"{key}-{i+1}" rewrite_entry(entry, phase=ph, path_t=ts[i], contact="grounded", anchor=cfg["anchor"], role=cfg["role"]) seq.setdefault("motionProfile", {}).update({"family": "linear_ground", "anchor": cfg["profile_anchor"], "x0": 34, "x1": 66, "loopMs": 1800, "ghostCount": 0, "ghostMode": "none", "debug": True}) tag_seq(seq, "green-active-footwork", "Footwork keeps explicit crossover/non-crossover semantics. MediaPipe geometry helps contact points but does not choose exercise intent.") add_change(key, "footwork semantics promoted from review to green-active") # Farmer carry: safe low-risk linear carry. Hands are measured, but load object is not inferred. if "db-farmer-carry" in sequences: seq = sequences["db-farmer-carry"] poses = seq.get("poses", []) ts = spread(len(poses)) phases = ["carry-step-a", "carry-step-b"] for i, entry in enumerate(poses): rewrite_entry(entry, phase=phases[i] if i < len(phases) else f"carry-{i+1}", path_t=ts[i], contact="grounded", anchor="footCenter", role="upright carry: foot path + hand/dumbbell visual relationship") seq.setdefault("motionProfile", {}).update({"family": "linear_ground", "anchor": "feet-with-hand-carry", "x0": 36, "x1": 64, "loopMs": 1700, "ghostCount": 0, "ghostMode": "none", "debug": True}) tag_seq(seq, "green-active-carry", "Farmer carry promoted; dumbbells remain visual props but hands/feet are MediaPipe-verified where confidence is adequate.") add_change("db-farmer-carry", "farmer carry promoted to green-active") # Keep broad jump as review-tightened only due low-confidence landing frame. if "broad-jump" in sequences: seq = sequences["broad-jump"] for i, entry in enumerate(seq.get("poses", [])): rewrite_entry(entry, anchor="footCenter", role="broad jump stays manual until low-confidence landing frame is reviewed") seq.setdefault("motionProfile", {}).update({"family": "jump_arc_forward", "anchor": "manual-foot-pair-review", "x0": 28, "x1": 68, "arcHeight": 14, "loopMs": 1750, "ghostCount": 0, "ghostMode": "none", "debug": True}) tag_seq(seq, "manual-review-tightened", "Broad jump is not MediaPipe-runtime-expanded because one candidate frame is below green threshold.", risk="moderate", gate="review-tightened") add_change("broad-jump", "kept manual due low-confidence frame", "metadata only") # Explicitly carry forward holds so tools show why not activated. for key, reason in { "chest-pass": "Wrong family placeholder; needs true chest-pass image/pose sequence.", "rotational-throw": "Wrong family placeholder; needs true rotational throw image/pose sequence.", "buddy-hamstring-curls": "Partner/knee-pivot special: remains manual even if visual smoke test OK.", "box-jump": "Box prop relationship needs human prop QA before more activation.", "seated-box-jump": "Box prop relationship needs human prop QA before more activation.", }.items(): seq = sequences.get(key) if seq: tag_seq(seq, "held-after-rc48-green-expansion", reason, risk="moderate-high", gate="held-special-or-prop") return changes def min_conf(candidate: Dict[str, Any], names: List[str]) -> float: vals: List[float] = [] for name in names: v = candidate.get(name) if not isinstance(v, dict): return 0.0 vals.append(float(v.get("confidence", v.get("visibility", 0)) or 0)) return min(vals) if vals else 0.0 def activate_mediapipe(packs: Dict[str, Any], sequences: Dict[str, Any]) -> Dict[str, Any]: activation_rules = { # Existing RC46 safe set retained. "lateral-line-hops": {"threshold": .68, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "bodyweight-squat": {"threshold": .82, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "hamstring-bridge": {"threshold": .66, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "shoulderCenter"], "mode": "candidate-retain-only"}, "pull-up": {"threshold": .64, "required": ["leftHand", "rightHand", "handCenter", "shoulderCenter", "hipCenter"]}, "chin-up": {"threshold": .64, "required": ["leftHand", "rightHand", "handCenter", "shoulderCenter", "hipCenter"]}, # RC48 new green expansion. "linear-pogos": {"threshold": .82, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "squat-jump": {"threshold": .82, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "lateral-pogo-jumps": {"threshold": .68, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "goblet-squat": {"threshold": .82, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "carioca-shuffle": {"threshold": .82, "required": ["leftFoot", "rightFoot", "dominantContactFoot", "hipCenter", "kneeCenter", "shoulderCenter"]}, "lateral-power-shuffle": {"threshold": .74, "required": ["leftFoot", "rightFoot", "footCenter", "hipCenter", "kneeCenter", "shoulderCenter"]}, "db-farmer-carry": {"threshold": .60, "required": ["leftFoot", "rightFoot", "footCenter", "leftHand", "rightHand", "hipCenter", "shoulderCenter"]}, } selected: Dict[str, Set[str]] = {} for ex, rule in activation_rules.items(): if rule.get("mode") == "candidate-retain-only": continue seq = sequences.get(ex) if not seq: continue # Do not activate anything explicitly held. if seq.get("rc48GreenExpansion", {}).get("runtimeGate", "") in {"held-special-or-prop", "review-tightened"}: continue for p in seq.get("poses", []): pk = p.get("poseKey") if pk: selected.setdefault(pk, set()).add(ex) activated, retained, skipped = [], [], [] for pose in packs.get("poses", []): pk = pose.get("poseKey") exs = sorted(selected.get(pk, [])) if not exs: continue thresholds = [activation_rules[e]["threshold"] for e in exs] required = sorted(set(sum((activation_rules[e]["required"] for e in exs), []))) threshold = min(thresholds) for aid, athlete in (pose.get("athletes") or {}).items(): sk = (((athlete or {}).get("metrics") or {}).get("skeleton") or {}) cand = sk.get("runtimeLandmarksMediapipeCandidate") if not cand: skipped.append({"poseKey": pk, "athleteId": aid, "exercises": exs, "reason": "no-candidate"}) continue score = min_conf(cand, required) already_mp = str(sk.get("runtimeLandmarkSource", "")).startswith("mediapipe") if score >= threshold: if "runtimeLandmarksAlphaFallback" not in sk: sk["runtimeLandmarksAlphaFallback"] = deepcopy(sk.get("runtimeLandmarks", {})) old_lm = sk.get("runtimeLandmarks", {}) or {} next_lm = deepcopy(cand) # Preserve synthetic/manual anchors that MediaPipe cannot see. for extra in ("barTop",): if extra in old_lm and extra not in next_lm: next_lm[extra] = deepcopy(old_lm[extra]) for lm in next_lm.values(): if isinstance(lm, dict): lm["activatedBy"] = RC48_VERSION sk["runtimeLandmarks"] = next_lm sk["runtimeLandmarkSource"] = "mediapipe-rc48-green-active" sk["quality"] = "mediapipe-green-active-with-alpha-fallback" sk.setdefault("mediaPipe", {})["runtimeActivatedAt"] = NOW sk.setdefault("mediaPipe", {})["runtimeActivationPolicy"] = "RC48 green expansion: low-risk/high-confidence exercises only; prop/partner/throw/floor edge cases remain gated." row = {"poseKey": pk, "athleteId": aid, "exercises": exs, "minRequiredConfidence": round(score, 3), "threshold": threshold} if already_mp: retained.append(row) else: activated.append(row) else: skipped.append({"poseKey": pk, "athleteId": aid, "exercises": exs, "minRequiredConfidence": round(score, 3), "threshold": threshold, "reason": "below-threshold"}) packs["version"] = RC48_VERSION packs["rc48Note"] = "Green motion expansion after RC47 visual smoke test. Runtime landmarks activated only for low-risk/high-confidence motion families." packs["mediaPipeBridge"] = packs.get("mediaPipeBridge", {}) | { "rc48GreenRuntimeExpansion": { "status": "green-partial-active", "newActivationCount": len(activated), "retainedOrReactivatedCount": len(retained), "skippedCount": len(skipped), "updatedAt": NOW, "policy": "Expand standing/jump/footwork/carry only. Hold throw, partner, prop and low-confidence floor patterns." } } return {"activated": activated, "retained": retained, "skipped": skipped, "selectedPoseKeys": sorted(selected.keys())} def main() -> None: seq = load_json(SEQ_PATH) packs = load_json(PACKS_PATH) rc46 = load_json(RC46_PATH) if RC46_PATH.exists() else {} rc47 = load_json(RC47_PATH) if RC47_PATH.exists() else {} if not BACKUP_SEQ.exists(): save_json(BACKUP_SEQ, seq) if not BACKUP_PACKS.exists(): save_json(BACKUP_PACKS, packs) changes = tighten_sequences(seq.get("sequences", {})) activation = activate_mediapipe(packs, seq.get("sequences", {})) seq["version"] = RC48_VERSION seq["limbIntentSchema"] = "rc48-movement-intent-v1" seq["rc48GreenMotionExpansion"] = { "version": RC48_VERSION, "createdAt": NOW, "sourceRc46": rc46.get("version"), "sourceRc47": rc47.get("version"), "visualSmokeTest": "User reported RC47 works; no sandbox browser animation test available.", "runtimePolicy": { "poseNamesAuthoritative": False, "expandedExercises": sorted({c["exerciseKey"] for c in changes if c["runtimeImpact"] != "metadata only"}), "heldExercises": ["chest-pass", "rotational-throw", "buddy-hamstring-curls", "box-jump", "seated-box-jump", "broad-jump"], "selectiveMediaPipeRuntimeActivation": True, "doNotInferObjectsFromMediaPipe": True, }, } report = { "version": RC48_VERSION, "createdAt": NOW, "status": "green-motion-expansion-after-visual-smoke-test", "summary": { "sequenceChanges": len(changes), "newMediaPipeActivations": len(activation["activated"]), "retainedOrReactivatedMediaPipe": len(activation["retained"]), "skippedMediaPipeCandidates": len(activation["skipped"]), "selectedPoseKeys": len(activation["selectedPoseKeys"]), "runtimeChanged": True, "imagesGenerated": False, "principle": "movement truth over pose/sheet names; green expansion only where visual smoke-test + confidence allow it", }, "changes": changes, "activatedMediapipeRuntimeLandmarks": activation["activated"], "retainedOrReactivatedMediapipeRuntimeLandmarks": activation["retained"], "skippedMediapipeRuntimeLandmarks": activation["skipped"], "heldPolicy": seq["rc48GreenMotionExpansion"]["runtimePolicy"]["heldExercises"], "notes": [ "RC48 does not generate or replace images.", "RC48 expands to standing/jump/footwork/carry families after RC47 worked visually for the user.", "Throw/medball wrong-family placeholders, partner/knee, box-prop and low-confidence broad-jump remain held or review-tightened.", "MediaPipe still measures geometry only; it does not define exercise correctness.", ], } movement_data = { "version": RC48_VERSION, "createdAt": NOW, "status": "green movement expansion", "report": report["summary"], "exerciseDecisions": {c["exerciseKey"]: c for c in changes}, "activationPolicy": seq["rc48GreenMotionExpansion"]["runtimePolicy"], } save_json(SEQ_PATH, seq) save_json(PACKS_PATH, packs) save_json(OUT_DATA, movement_data) save_json(OUT_REPORT, report) print(json.dumps(report["summary"], ensure_ascii=False, indent=2)) if __name__ == "__main__": main()