#!/usr/bin/env python3 """ NEXT Athlete OS ยท RC40 MediaPipe patch merge tool. Purpose ------- Merge JSON patches exported by tools/mediapipe_pose_extract_browser.html into assets/data/image-admin/pose-atlas-packs.json without blindly overwriting the hand-authored / alpha-derived runtime bindings from RC38/RC39. Default behaviour is dry-run + candidate-only. To write into the atlas, pass --apply. To make MediaPipe landmarks active for runtime, also pass --activate-runtime. Otherwise they are stored as runtimeLandmarksMediapipeCandidate and the current runtimeLandmarks stay untouched. Key safety rules ---------------- - Raw MediaPipe anatomical left/right is stored separately under skeleton.mediaPipe. - Runtime-facing leftFoot/rightFoot and leftHand/rightHand remain visual-space names. The merge maps MediaPipe candidates by x-order so image and skeleton keep the same visible limb identity after runtime flip. - Existing alpha runtimeLandmarks are copied to runtimeLandmarksAlphaFallback before any active overwrite. - Existing limbBinding and sequence limbIntent are preserved. """ from __future__ import annotations import argparse import copy import datetime as dt import json import math import shutil import sys from pathlib import Path from typing import Any, Dict, List, Tuple ROOT = Path(__file__).resolve().parents[1] DEFAULT_PACKS = ROOT / "assets/data/image-admin/pose-atlas-packs.json" DEFAULT_REPORT = ROOT / "assets/data/image-admin/rc40_mediapipe_patch_merge_report.json" DEFAULT_BACKUP = ROOT / "assets/data/image-admin/pose-atlas-packs.rc39-before-rc40-mediapipe-merge.backup.json" RUNTIME_KEYS = ( "headTop", "headCenter", "shoulderCenter", "leftHand", "rightHand", "handCenter", "hipCenter", "kneeCenter", "leftFoot", "rightFoot", "footCenter", "dominantContactFoot", "primaryContactFoot", "rearFoot", "frontFoot", "supportCenter", "lowerCenter" ) CORE_KEYS = ("leftFoot", "rightFoot", "leftHand", "rightHand", "shoulderCenter", "hipCenter", "kneeCenter") def now_iso() -> str: return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat() def read_json(path: Path) -> Any: return json.loads(path.read_text(encoding="utf-8")) def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def pct(n: Any) -> float | None: try: f = float(n) except (TypeError, ValueError): return None if not math.isfinite(f): return None return max(0.0, min(100.0, round(f, 2))) def conf_from(point: Dict[str, Any], fallback: float = 0.5) -> float: for key in ("confidence", "visibility", "presence"): if key in point: try: v = float(point.get(key)) if math.isfinite(v): return round(max(0.0, min(1.0, v)), 3) except (TypeError, ValueError): pass return fallback def clean_point(point: Any, source: str = "mediapipe-rc40") -> Dict[str, Any] | None: if not isinstance(point, dict): return None x, y = pct(point.get("x")), pct(point.get("y")) if x is None or y is None: return None out = {"x": x, "y": y, "confidence": conf_from(point), "source": source} if "visibility" in point: try: out["visibility"] = round(max(0.0, min(1.0, float(point.get("visibility")))), 3) except (TypeError, ValueError): pass return out def midpoint(a: Dict[str, Any] | None, b: Dict[str, Any] | None, source: str = "mediapipe-rc40") -> Dict[str, Any] | None: if not a and not b: return None if a and not b: return copy.deepcopy(a) if b and not a: return copy.deepcopy(b) assert a and b return { "x": round((float(a["x"]) + float(b["x"])) / 2.0, 2), "y": round((float(a["y"]) + float(b["y"])) / 2.0, 2), "confidence": round(min(float(a.get("confidence", 0.5)), float(b.get("confidence", 0.5))), 3), "source": source, } def normalized_derived(item: Dict[str, Any]) -> Dict[str, Any]: mp = item.get("mediaPipe") if isinstance(item.get("mediaPipe"), dict) else {} # Support both RC39 and future patch shapes. derived = ( mp.get("derivedRuntimeLandmarks") or mp.get("runtimeLandmarks") or item.get("derivedRuntimeLandmarks") or item.get("runtimeLandmarks") or {} ) if not isinstance(derived, dict): return {} return {k: clean_point(v) for k, v in derived.items() if clean_point(v)} def visual_lr(candidate: Dict[str, Any], left_key: str, right_key: str) -> Tuple[Dict[str, Any] | None, Dict[str, Any] | None, bool]: a = candidate.get(left_key) b = candidate.get(right_key) if not a or not b: return a, b, False # MediaPipe names are anatomical. Runtime names are visual-space. Sort by x. if float(a["x"]) <= float(b["x"]): left, right = copy.deepcopy(a), copy.deepcopy(b) swapped = False else: left, right = copy.deepcopy(b), copy.deepcopy(a) swapped = True left["visualBinding"] = "visual-left" right["visualBinding"] = "visual-right" return left, right, swapped def closest_named_foot(old: Dict[str, Any], name: str, left: Dict[str, Any] | None, right: Dict[str, Any] | None, center: Dict[str, Any] | None) -> Dict[str, Any] | None: oldp = old.get(name) if not isinstance(oldp, dict) or pct(oldp.get("x")) is None: return copy.deepcopy(center or left or right) options = [("left", left), ("right", right), ("center", center)] options = [(label, p) for label, p in options if p] if not options: return None ox = float(oldp.get("x")) _, best = min(options, key=lambda pair: abs(float(pair[1].get("x", 50)) - ox)) return copy.deepcopy(best) def build_visual_runtime(candidate: Dict[str, Any], old_runtime: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any]]: out: Dict[str, Any] = {} notes: Dict[str, Any] = {"footLabelsSwappedToVisual": False, "handLabelsSwappedToVisual": False} left_foot, right_foot, swapped_feet = visual_lr(candidate, "leftFoot", "rightFoot") left_hand, right_hand, swapped_hands = visual_lr(candidate, "leftHand", "rightHand") notes["footLabelsSwappedToVisual"] = swapped_feet notes["handLabelsSwappedToVisual"] = swapped_hands for key in ("headTop", "headCenter", "shoulderCenter", "hipCenter", "kneeCenter"): if candidate.get(key): out[key] = copy.deepcopy(candidate[key]) if left_hand: out["leftHand"] = left_hand if right_hand: out["rightHand"] = right_hand out["handCenter"] = midpoint(left_hand, right_hand) or copy.deepcopy(candidate.get("handCenter") or old_runtime.get("handCenter")) if left_foot: out["leftFoot"] = left_foot if right_foot: out["rightFoot"] = right_foot foot_center = midpoint(left_foot, right_foot) or copy.deepcopy(candidate.get("footCenter") or old_runtime.get("footCenter")) if foot_center: foot_center["source"] = "mediapipe-rc40-visual" out["footCenter"] = foot_center # Preserve the old contact semantics by remapping old rear/front/dominant names to the closest # visual MediaPipe foot. This keeps sequence limbIntent stable. for name in ("dominantContactFoot", "primaryContactFoot", "rearFoot", "frontFoot"): mapped = closest_named_foot(old_runtime, name, left_foot, right_foot, foot_center) if mapped: mapped["source"] = "mediapipe-rc40-visual-contact-preserved" out[name] = mapped support = copy.deepcopy(candidate.get("supportCenter")) if not support: support = midpoint(out.get("handCenter"), out.get("footCenter"), "mediapipe-rc40-visual") if support: out["supportCenter"] = support lower = copy.deepcopy(candidate.get("lowerCenter")) if not lower: lower = midpoint(out.get("kneeCenter"), out.get("footCenter"), "mediapipe-rc40-visual") if lower: out["lowerCenter"] = lower # Fill any missing runtime keys from alpha fallback so the runtime never loses a point. for key in RUNTIME_KEYS: if key not in out and isinstance(old_runtime.get(key), dict): fallback = copy.deepcopy(old_runtime[key]) fallback.setdefault("fallbackFrom", "alpha-runtime") out[key] = fallback return out, notes def raw_landmarks(item: Dict[str, Any]) -> List[Dict[str, Any]]: mp = item.get("mediaPipe") if isinstance(item.get("mediaPipe"), dict) else {} lms = mp.get("landmarks") or item.get("landmarks") or [] return lms if isinstance(lms, list) else [] def mean_visibility(lms: List[Dict[str, Any]]) -> float: vals = [] for p in lms: if isinstance(p, dict): try: v = float(p.get("visibility", p.get("confidence", 0))) if math.isfinite(v): vals.append(max(0.0, min(1.0, v))) except (TypeError, ValueError): pass return round(sum(vals) / len(vals), 3) if vals else 0.0 def validate_item(item: Dict[str, Any], athlete: Dict[str, Any], pose_key: str, athlete_id: str) -> Tuple[bool, List[str], List[str], Dict[str, Any]]: errors: List[str] = [] warnings: List[str] = [] lms = raw_landmarks(item) candidate = normalized_derived(item) old_runtime = athlete.get("metrics", {}).get("skeleton", {}).get("runtimeLandmarks", {}) or {} status = str(item.get("status") or item.get("mediaPipe", {}).get("status") or "").lower() if status and status not in ("detected", "ok", "success", "merged", "candidate"): errors.append(f"status={status}") if len(lms) and len(lms) != 33: warnings.append(f"raw landmark count is {len(lms)}, expected 33") if not len(lms): warnings.append("no raw landmarks included; merging derived candidate only") if mean_visibility(lms) and mean_visibility(lms) < 0.35: warnings.append(f"low mean visibility {mean_visibility(lms)}") src = str(item.get("src") or item.get("spriteSrc") or "") expected_src = str(athlete.get("src") or "") if src and expected_src and src.split("/")[-1] != expected_src.split("/")[-1]: errors.append(f"sprite filename mismatch patch={src.split('/')[-1]} atlas={expected_src.split('/')[-1]}") present = [k for k in CORE_KEYS if k in candidate] if len(present) < 4: errors.append(f"too few core derived runtime landmarks: {present}") # Compare to alpha fallback, but don't hard-fail unless wildly off. MediaPipe can improve # hands/knees a lot, but feet jumping half a sprite usually means wrong crop/person. displacements: Dict[str, float] = {} for key in ("leftFoot", "rightFoot", "leftHand", "rightHand", "hipCenter", "shoulderCenter"): a, b = old_runtime.get(key), candidate.get(key) if isinstance(a, dict) and isinstance(b, dict): try: d = math.dist((float(a.get("x", 50)), float(a.get("y", 50))), (float(b.get("x", 50)), float(b.get("y", 50)))) displacements[key] = round(d, 2) if key.endswith("Foot") and d > 34: warnings.append(f"large {key} displacement {d:.1f}pct vs alpha fallback") if key.endswith("Hand") and d > 42: warnings.append(f"large {key} displacement {d:.1f}pct vs alpha fallback") except Exception: pass meta = {"rawLandmarkCount": len(lms), "meanVisibility": mean_visibility(lms), "candidateKeys": sorted(candidate.keys()), "displacementsPct": displacements} return not errors, errors, warnings, meta def index_poses(packs: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: return {str(p.get("poseKey")): p for p in packs.get("poses", []) if isinstance(p, dict) and p.get("poseKey")} def merge_patch(packs: Dict[str, Any], patch: Dict[str, Any], *, activate_runtime: bool, candidate_only: bool) -> Dict[str, Any]: started = now_iso() pose_index = index_poses(packs) items = patch.get("items") if isinstance(patch.get("items"), list) else [] report = { "version": "v97-rc40-mediapipe-patch-merge", "startedAt": started, "finishedAt": None, "patchSchema": patch.get("schema"), "patchCreatedAt": patch.get("createdAt"), "applyPolicy": { "candidateOnly": bool(candidate_only), "activateRuntime": bool(activate_runtime), "leftRightRuntimePolicy": "visual-x-order; raw MediaPipe anatomical labels preserved under mediaPipe.rawLandmarks", "preserveAlphaFallback": True, "preserveSequenceLimbIntent": True, }, "counts": {"patchItems": len(items), "accepted": 0, "skipped": 0, "runtimeActivated": 0, "candidateStored": 0, "missingPose": 0, "missingAthlete": 0}, "items": [], } for item in items: pose_key = str(item.get("poseKey") or "") athlete_id = str(item.get("athleteId") or "") detail = {"poseKey": pose_key, "athleteId": athlete_id, "accepted": False, "warnings": [], "errors": [], "mergeNotes": {}} pose = pose_index.get(pose_key) if not pose: report["counts"]["missingPose"] += 1 report["counts"]["skipped"] += 1 detail["errors"].append("poseKey not found in atlas") report["items"].append(detail) continue athlete = pose.get("athletes", {}).get(athlete_id) if not athlete: report["counts"]["missingAthlete"] += 1 report["counts"]["skipped"] += 1 detail["errors"].append("athleteId not found for pose") report["items"].append(detail) continue ok, errors, warnings, meta = validate_item(item, athlete, pose_key, athlete_id) detail["warnings"].extend(warnings) detail["errors"].extend(errors) detail["validation"] = meta if not ok: report["counts"]["skipped"] += 1 report["items"].append(detail) continue sk = athlete.setdefault("metrics", {}).setdefault("skeleton", {}) old_runtime = copy.deepcopy(sk.get("runtimeLandmarks", {}) or {}) if old_runtime and "runtimeLandmarksAlphaFallback" not in sk: sk["runtimeLandmarksAlphaFallback"] = old_runtime raw_candidate = normalized_derived(item) visual_candidate, notes = build_visual_runtime(raw_candidate, old_runtime) detail["mergeNotes"] = notes sk["mediaPipe"] = { "schema": "mediapipe-pose-33-v1", "status": "runtime-active" if activate_runtime and not candidate_only else "merged-candidate", "mergedAt": started, "patchSchema": patch.get("schema"), "sourceTool": patch.get("sourceTool") or "tools/mediapipe_pose_extract_browser.html", "coordinateSpace": "sprite-percent-unflipped", "rawLeftRightPolicy": "MediaPipe anatomical names are preserved here only; runtime landmarks use visual x-order names.", "rawLandmarks": raw_landmarks(item), "derivedRuntimeLandmarksRaw": raw_candidate, "visualRuntimeLandmarks": visual_candidate, "validation": meta, "mergeNotes": notes, } sk["runtimeLandmarksMediapipeCandidate"] = visual_candidate report["counts"]["candidateStored"] += 1 if activate_runtime and not candidate_only: sk["runtimeLandmarks"] = visual_candidate sk["runtimeLandmarkSource"] = "mediapipe-rc40-active-with-alpha-fallback" sk["quality"] = "mediapipe-merged-runtime-active-alpha-fallback" report["counts"]["runtimeActivated"] += 1 else: sk["runtimeLandmarkSource"] = sk.get("runtimeLandmarkSource") or "alpha-rc39-active-mediapipe-candidate-stored" detail["accepted"] = True report["counts"]["accepted"] += 1 report["items"].append(detail) packs["version"] = "v97-rc40-mediapipe-patch-merge-ready" packs["mediaPipeBridge"] = { "schema": "mediapipe-pose-33-v1", "status": "patch-merge-ready", "browserTool": "tools/mediapipe_pose_extract_browser.html", "mergeTool": "tools/pose_atlas_rc40_merge_mediapipe_patch.py", "safeMergePolicy": "candidate-only by default; activate runtime only with --activate-runtime", "runtimeLeftRightPolicy": "visual-x-order; image and skeleton are transformed together by runtime flip", "rawMediaPipePolicy": "anatomical MediaPipe left/right is preserved under metrics.skeleton.mediaPipe.rawLandmarks", "lastMergeReport": "assets/data/image-admin/rc40_mediapipe_patch_merge_report.json", } report["finishedAt"] = now_iso() return report def make_selftest_patch(packs: Dict[str, Any], limit: int = 8) -> Dict[str, Any]: items = [] for pose in packs.get("poses", []): for athlete_id, athlete in (pose.get("athletes") or {}).items(): sk = athlete.get("metrics", {}).get("skeleton", {}) rt = sk.get("runtimeLandmarks", {}) if not rt: continue # Build a tiny deterministic fake MediaPipe patch from existing runtime landmarks. # This validates merge plumbing and visual left/right mapping without pretending # that real MediaPipe has run. derived = {} for k, v in rt.items(): if isinstance(v, dict) and "x" in v and "y" in v: derived[k] = {"x": v.get("x"), "y": v.get("y"), "visibility": v.get("confidence", 0.66)} raw = [] for i in range(33): raw.append({"name": f"selfTest{i:02d}", "x": 50, "y": 50, "z": 0, "visibility": 0.75}) items.append({ "poseKey": pose.get("poseKey"), "athleteId": athlete_id, "src": athlete.get("src"), "status": "detected", "mediaPipe": { "schema": "mediapipe-pose-33-v1-self-test", "landmarks": raw, "derivedRuntimeLandmarks": derived, }, }) if len(items) >= limit: return { "schema": "rc40-mediapipe-pose-33-patch-self-test-v1", "createdAt": now_iso(), "sourceTool": "self-test inside pose_atlas_rc40_merge_mediapipe_patch.py", "items": items, } return {"schema": "rc40-mediapipe-pose-33-patch-self-test-v1", "createdAt": now_iso(), "items": items} def main(argv: List[str] | None = None) -> int: ap = argparse.ArgumentParser(description="Merge MediaPipe pose patches into NEXT pose atlas safely.") ap.add_argument("--patch", type=Path, help="Patch JSON exported from tools/mediapipe_pose_extract_browser.html") ap.add_argument("--packs", type=Path, default=DEFAULT_PACKS, help="pose-atlas-packs.json path") ap.add_argument("--out", type=Path, help="Output atlas path. Defaults to --packs when --apply, otherwise no atlas write.") ap.add_argument("--report", type=Path, default=DEFAULT_REPORT, help="Merge report JSON path") ap.add_argument("--apply", action="store_true", help="Write merged atlas. Without this, only report is written.") ap.add_argument("--activate-runtime", action="store_true", help="Make MediaPipe visual landmarks active as runtimeLandmarks.") ap.add_argument("--self-test", action="store_true", help="Use a deterministic fake patch from current alpha landmarks to test merge plumbing.") ap.add_argument("--limit", type=int, default=8, help="Self-test item limit") args = ap.parse_args(argv) packs = read_json(args.packs) if args.self_test: patch = make_selftest_patch(packs, args.limit) else: if not args.patch: ap.error("--patch is required unless --self-test is used") patch = read_json(args.patch) # candidate_only means: do not overwrite runtimeLandmarks. candidate_only = not args.activate_runtime work = copy.deepcopy(packs) report = merge_patch(work, patch, activate_runtime=args.activate_runtime, candidate_only=candidate_only) report["dryRun"] = not args.apply report["packsPath"] = str(args.packs) report["outputPath"] = str(args.out or args.packs if args.apply else "") report["backupPath"] = str(DEFAULT_BACKUP if args.apply and (args.out is None or args.out == args.packs) else "") if args.apply: out = args.out or args.packs if out == args.packs: shutil.copy2(args.packs, DEFAULT_BACKUP) write_json(out, work) write_json(args.report, report) print(json.dumps({"report": str(args.report), "counts": report["counts"], "dryRun": report["dryRun"]}, indent=2)) if report["counts"].get("skipped"): return 2 if args.apply else 0 return 0 if __name__ == "__main__": raise SystemExit(main())