#!/usr/bin/env python3 """Build RC50 visual review/correction manifest. This is intentionally review-only: it does not change runtime sequences or atlas packs. It creates an index for the browser lab so reviewers can flag movement/anchor/image issues. """ from __future__ import annotations import json, hashlib from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / 'assets' / 'data' / 'image-admin' def load(name): with open(DATA / name, 'r', encoding='utf-8') as f: return json.load(f) def dump(name, data): p = DATA / name with open(p, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write('\n') return p def sha_short(obj): b = json.dumps(obj, sort_keys=True, ensure_ascii=False).encode('utf-8') return hashlib.sha256(b).hexdigest()[:12] seq = load('pose-atlas-exercise-sequences.json') packs = load('pose-atlas-packs.json') trace = load('rc49_motion_trace_verification.json') rc49 = load('rc49_special_prop_binding.json') physio = load('rc45_ai_physio_first_pass.json') if (DATA/'rc45_ai_physio_first_pass.json').exists() else {} poses_by_key = {p['poseKey']: p for p in packs.get('poses', [])} rc49_gate_by_ex = {c['exerciseKey']: c.get('runtimeGate', '') for c in rc49.get('changes', [])} rc49_action_by_ex = {c['exerciseKey']: c for c in rc49.get('changes', [])} trace_by_ex = trace.get('traces', {}) review_hold = set(trace.get('summary', {}).get('special', {}).get('reviewHoldExercises', [])) manual_prop = set(trace.get('summary', {}).get('special', {}).get('manualPropLockedExercises', [])) # Extract rc45 physio entries if present, but tolerate many shapes. physio_by_ex = {} for key in ('exercises', 'items', 'review', 'firstPass'): val = physio.get(key) if isinstance(physio, dict) else None if isinstance(val, dict): physio_by_ex.update(val) elif isinstance(val, list): for item in val: if isinstance(item, dict) and item.get('exerciseKey'): physio_by_ex[item['exerciseKey']] = item priority_order = { 'review-hold-needs-true-pose-set': 0, 'manual-knee-pivot-locked': 1, 'manual-prop-locked': 2, 'manual-review-retained': 3, 'special-prop-active': 4, 'green-active': 5, 'default': 6, } def source_state(frame, pose_key): pose = poses_by_key.get(pose_key, {}) active = candidate = alpha = 0 min_active_conf = 1.0 min_candidate_conf = 1.0 athlete_summary = [] for aid, ad in (pose.get('athletes') or {}).items(): sk = ((ad.get('metrics') or {}).get('skeleton') or {}) src = sk.get('runtimeLandmarkSource') or 'unknown' rl = sk.get('runtimeLandmarks') or {} cand = sk.get('runtimeLandmarksMediapipeCandidate') or {} if isinstance(src, str) and src.startswith('mediapipe'): active += 1 elif cand: candidate += 1 else: alpha += 1 anchors = [frame.get('anchorLandmark'), 'footCenter', 'handCenter', 'kneeCenter', 'hipCenter'] confs = [] cand_confs = [] for a in anchors: if a and isinstance(rl.get(a), dict): confs.append(rl[a].get('confidence', rl[a].get('visibility', None))) if a and isinstance(cand.get(a), dict): cand_confs.append(cand[a].get('confidence', cand[a].get('visibility', None))) confs = [float(x) for x in confs if isinstance(x, (int, float))] cand_confs = [float(x) for x in cand_confs if isinstance(x, (int, float))] if confs: min_active_conf = min(min_active_conf, min(confs)) if cand_confs: min_candidate_conf = min(min_candidate_conf, min(cand_confs)) athlete_summary.append({ 'athleteId': aid, 'src': ad.get('src'), 'runtimeLandmarkSource': src, 'anchor': frame.get('anchorLandmark'), 'anchorConfidence': min(confs) if confs else None, 'candidateConfidence': min(cand_confs) if cand_confs else None, }) return { 'mediapipeActiveAthletes': active, 'mediapipeCandidateAthletes': candidate, 'alphaFallbackAthletes': alpha, 'minRuntimeConfidence': None if min_active_conf == 1.0 else round(min_active_conf, 3), 'minCandidateConfidence': None if min_candidate_conf == 1.0 else round(min_candidate_conf, 3), 'athletes': athlete_summary, } exercises = [] for ex_key, ex in seq.get('sequences', {}).items(): gate = rc49_gate_by_ex.get(ex_key) if not gate: if ex_key in review_hold: gate = 'review-hold-needs-true-pose-set' elif ex_key in manual_prop: gate = 'manual-prop-locked' else: gate = 'green-active' if ex.get('poses') else 'default' frames = [] missing = [] active_frames = 0 candidate_frames = 0 for i, fr in enumerate(ex.get('poses', []), start=1): pk = fr.get('poseKey') if pk not in poses_by_key: missing.append(pk) state = source_state(fr, pk) if state['mediapipeActiveAthletes']: active_frames += 1 if state['mediapipeCandidateAthletes']: candidate_frames += 1 frames.append({ 'index': i, 'poseKey': pk, 'phase': fr.get('phase'), 'pathT': fr.get('pathT'), 'contact': fr.get('contact'), 'anchorLandmark': fr.get('anchorLandmark'), 'limbIntent': fr.get('limbIntent'), 'runtimeGate': gate, 'reviewDefault': 'hold' if 'review-hold' in gate else 'unchecked', 'candidateState': state, }) phys = physio_by_ex.get(ex_key, {}) if isinstance(physio_by_ex.get(ex_key, {}), dict) else {} exercises.append({ 'exerciseKey': ex_key, 'label': ex.get('label') or ex_key, 'runtimeGate': gate, 'reviewPriority': priority_order.get(gate, priority_order['default']), 'reviewRecommended': bool(ex_key in review_hold or ex_key in manual_prop or 'hold' in gate or 'manual' in gate), 'frameCount': len(frames), 'mediapipeActiveFrames': active_frames, 'mediapipeCandidateFrames': candidate_frames, 'missingPoseKeys': missing, 'rc49Action': rc49_action_by_ex.get(ex_key), 'physioFirstPass': {k: phys.get(k) for k in ('appDecision','priority','purpose','mustShow','avoid','coachCues') if k in phys}, 'frames': frames, }) exercises.sort(key=lambda x: (x['reviewPriority'], x['exerciseKey'])) summary = { 'version': 'v97-rc50-visual-review-correction-mode', 'createdAt': datetime.now(timezone.utc).isoformat(timespec='seconds'), 'sourceBase': (seq.get('rc50VisualReviewCorrectionMode') or {}).get('base') or seq.get('version'), 'runtimeChanged': False, 'userConfirmedBase': 'RC49 visual smoke test was reported OK in chat before RC50 build.', 'purpose': 'Reviewer flags movement/anchor/image-candidate issues against physio movement truth without modifying runtime.', 'counts': { 'exercises': len(exercises), 'frames': sum(e['frameCount'] for e in exercises), 'reviewRecommendedExercises': sum(1 for e in exercises if e['reviewRecommended']), 'reviewHoldExercises': len(review_hold), 'manualPropLockedExercises': len(manual_prop), 'mediapipeActiveFrameEntries': sum(e['mediapipeActiveFrames'] for e in exercises), 'mediapipeCandidateFrameEntries': sum(e['mediapipeCandidateFrames'] for e in exercises), 'missingPoseKeys': sum(len(e['missingPoseKeys']) for e in exercises), }, 'patchContract': { 'exportFile': 'rc50_visual_review_patch.json', 'reviewStatuses': ['ok','wrong-foot-hand','wrong-direction','wrong-anchor','wrong-image-candidate','needs-new-pose','hold','replace-exercise'], 'applyPolicy': 'review-only by default; apply script creates proposal/report and does not overwrite runtime unless explicit future step is built.' }, 'hashes': { 'sequenceHash': sha_short(seq.get('sequences', {})), 'packHash': sha_short(packs.get('poses', [])), 'traceHash': sha_short(trace), }, } manifest = {'summary': summary, 'exercises': exercises} path = dump('rc50_visual_review_manifest.json', manifest) print(path) print(json.dumps(summary, indent=2, ensure_ascii=False))