#!/usr/bin/env python3 """ Run MediaPipe Pose on collected keypose still images and emit NEXT11 candidate data. This is NOT an IK approval tool. Output is candidate-only and must be reviewed against existing IK/Exercise Base sacred points before any Nex runtime use. """ from __future__ import annotations import argparse import json import math import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple LANDMARK_NAMES = [ 'nose','left_eye_inner','left_eye','left_eye_outer','right_eye_inner','right_eye','right_eye_outer', 'left_ear','right_ear','mouth_left','mouth_right','left_shoulder','right_shoulder','left_elbow', 'right_elbow','left_wrist','right_wrist','left_pinky','right_pinky','left_index','right_index', 'left_thumb','right_thumb','left_hip','right_hip','left_knee','right_knee','left_ankle','right_ankle', 'left_heel','right_heel','left_foot_index','right_foot_index' ] POSE_CONNECTIONS = [ (0,1),(1,2),(2,3),(3,7),(0,4),(4,5),(5,6),(6,8),(9,10), (11,12),(11,13),(13,15),(15,17),(15,19),(15,21),(17,19), (12,14),(14,16),(16,18),(16,20),(16,22),(18,20), (11,23),(12,24),(23,24),(23,25),(24,26),(25,27),(26,28), (27,29),(28,30),(29,31),(30,32),(27,31),(28,32) ] IMAGE_EXTS = {'.png', '.jpg', '.jpeg', '.webp', '.bmp'} DEFAULT_KEYPOSE_ORDER = ['start_or_setup','load_or_descent','deep_or_bottom','return_or_ascent','finish'] def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z') def norm(s: str) -> str: return re.sub(r"[^a-z0-9]+", "_", s.lower()).strip("_") def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description='Run MediaPipe Pose on NEXT11 keypose still images and emit IK candidate JSON.') p.add_argument('--image-root', required=True, help='Directory containing images, usually one subfolder per exercise') p.add_argument('--exercise-index', help='Output from discover_next11_ik_data.py') p.add_argument('--source-plan', help='Output from build_keypose_image_plan.py; gives preferred keypose order') p.add_argument('--exercise', help='Filter/single exercise key/display contains this string') p.add_argument('--out', required=True, help='Output directory') p.add_argument('--pose-model', required=True, help='Path to pose_landmarker_full.task/lite.task/heavy.task') p.add_argument('--no-annotated', action='store_true', help='Skip annotated images') return p.parse_args() def collect_images_for_exercise(root: Path, exercise_key: str) -> List[Path]: candidates: List[Path] = [] aliases = {exercise_key, norm(exercise_key), norm(exercise_key).replace('_','-')} # Folder names in prior packages may use either hyphen or underscore. Accept both, # but return each file only once. This is candidate-pipeline only; no runtime impact. for alias in sorted(aliases): folder = root / alias if folder.exists(): candidates.extend(p for p in sorted(folder.rglob('*')) if p.suffix.lower() in IMAGE_EXTS) # Also allow loose files prefixed with exercise key or alias. for p in sorted(root.rglob('*')): if p.is_file() and p.suffix.lower() in IMAGE_EXTS: stem = norm(p.name) if any(stem.startswith(norm(a)) for a in aliases): candidates.append(p) # De-duplicate while preserving order. seen=set(); out=[] for p in candidates: rp=str(p.resolve()) if rp not in seen: seen.add(rp); out.append(p) return out def load_json(path: Optional[str]) -> Optional[Dict[str, Any]]: if not path: return None p = Path(path) if not p.exists(): return None return json.loads(p.read_text(encoding='utf-8')) def select_exercises(index: Optional[Dict[str, Any]], plan: Optional[Dict[str, Any]], image_root: Path, filter_text: Optional[str]) -> List[Dict[str, Any]]: by_key: Dict[str, Dict[str, Any]] = {} if index: for ex in index.get('exercises', []): key = str(ex.get('exerciseKey') or norm(ex.get('displayName', 'exercise'))) by_key[key] = dict(ex) if plan: for ex in plan.get('exercises', []): key = str(ex.get('exerciseKey') or norm(ex.get('displayName', 'exercise'))) base = by_key.get(key, {'exerciseKey': key, 'displayName': ex.get('displayName') or key}) base['keyposes'] = ex.get('keyposes', []) base['sourceFolder'] = ex.get('sourceFolder') by_key[key] = base # Image folders are valid even without index. for d in sorted(image_root.iterdir()) if image_root.exists() else []: if d.is_dir(): key = norm(d.name) by_key.setdefault(key, {'exerciseKey': key, 'displayName': d.name, 'keyposes': []}) exercises = list(by_key.values()) if filter_text: needle = norm(filter_text) exercises = [e for e in exercises if needle in norm(str(e.get('exerciseKey','') + ' ' + e.get('displayName','')))] return sorted(exercises, key=lambda e: str(e.get('exerciseKey',''))) def infer_keypose_id(path: Path, idx: int, keyposes: List[Dict[str, Any]]) -> str: low = norm(path.stem) for kp in keyposes: phase = norm(str(kp.get('phaseId') or '')) if phase and phase in low: return phase aliases = [ ('bottom', 'bottom'), ('deep', 'bottom'), ('down', 'descent_mid'), ('descent', 'descent_mid'), ('load', 'load_or_descent'), ('takeoff', 'takeoff'), ('apex', 'flight_apex'), ('flight', 'flight_apex'), ('landing', 'landing_contact'), ('press', 'press_mid'), ('return', 'return_or_ascent'), ('finish', 'finish'), ('start', 'start_or_setup'), ('top', 'top_or_start'), ('setup', 'start_or_setup'), ] for needle, phase in aliases: if needle in low: return phase if idx < len(keyposes): return norm(str(keyposes[idx].get('phaseId') or DEFAULT_KEYPOSE_ORDER[min(idx, len(DEFAULT_KEYPOSE_ORDER)-1)])) if idx < len(DEFAULT_KEYPOSE_ORDER): return DEFAULT_KEYPOSE_ORDER[idx] return f'extra_{idx+1:02d}' def lm_to_dict(lm: Any, name: str) -> Dict[str, float | str]: return { 'name': name, 'x': float(lm.x), 'y': float(lm.y), 'z': float(lm.z), 'visibility': float(getattr(lm, 'visibility', 0.0)), 'presence': float(getattr(lm, 'presence', 0.0)), } def midpoint(a: Dict[str, Any], b: Dict[str, Any], name: str) -> Dict[str, float | str]: return { 'name': name, 'x': (float(a['x']) + float(b['x'])) / 2, 'y': (float(a['y']) + float(b['y'])) / 2, 'z': (float(a['z']) + float(b['z'])) / 2, 'visibility': min(float(a.get('visibility', 0)), float(b.get('visibility', 0))), } def angle_deg(a: Dict[str, Any], b: Dict[str, Any], c: Dict[str, Any]) -> Optional[float]: ax, ay = float(a['x']), float(a['y']) bx, by = float(b['x']), float(b['y']) cx, cy = float(c['x']), float(c['y']) v1, v2 = (ax-bx, ay-by), (cx-bx, cy-by) n1, n2 = math.hypot(*v1), math.hypot(*v2) if n1 < 1e-9 or n2 < 1e-9: return None dot = max(-1.0, min(1.0, (v1[0]*v2[0] + v1[1]*v2[1])/(n1*n2))) return math.degrees(math.acos(dot)) def point_line_distance(p: Dict[str, Any], a: Dict[str, Any], b: Dict[str, Any]) -> Optional[float]: px, py = float(p['x']), float(p['y']) ax, ay = float(a['x']), float(a['y']) bx, by = float(b['x']), float(b['y']) denom = math.hypot(bx-ax, by-ay) if denom < 1e-9: return None return abs((by-ay)*px - (bx-ax)*py + bx*ay - by*ax) / denom def derived_candidates(lms: List[Dict[str, Any]]) -> Dict[str, Any]: by = {str(p['name']): p for p in lms} out: Dict[str, Any] = {} def has(*names: str) -> bool: return all(n in by for n in names) pairs = [ ('left_hip','right_hip','root_hip_midpoint'), ('left_hip','right_hip','pelvis_sacrum_candidate'), ('left_shoulder','right_shoulder','thorax_shoulder_midpoint'), ('left_index','left_pinky','left_palm_candidate'), ('right_index','right_pinky','right_palm_candidate'), ('left_heel','left_foot_index','left_foot_contact_candidate'), ('right_heel','right_foot_index','right_foot_contact_candidate'), ] for a,b,name in pairs: if has(a,b): out[name] = midpoint(by[a], by[b], name) # Pass through primary landmarks under NEXT11-ish names as candidates. for name in ['left_shoulder','right_shoulder','left_elbow','right_elbow','left_wrist','right_wrist','left_hip','right_hip','left_knee','right_knee','left_ankle','right_ankle','left_heel','right_heel','left_foot_index','right_foot_index','nose']: if name in by: out[f'{name}_candidate'] = by[name] return out def generic_metrics(lms: List[Dict[str, Any]], exercise_key: str) -> Dict[str, Any]: by = {str(p['name']): p for p in lms} m: Dict[str, Any] = {} vis = [float(p.get('visibility', 0)) for p in lms] m['mean_pose_visibility'] = sum(vis)/len(vis) if vis else 0 m['min_pose_visibility'] = min(vis) if vis else 0 for side in ['left','right']: s,e,w = by.get(f'{side}_shoulder'), by.get(f'{side}_elbow'), by.get(f'{side}_wrist') h,k,a = by.get(f'{side}_hip'), by.get(f'{side}_knee'), by.get(f'{side}_ankle') if s and e and w: m[f'{side}_elbow_angle_deg'] = angle_deg(s,e,w) if h and k and a: m[f'{side}_knee_angle_deg'] = angle_deg(h,k,a) if k and a and by.get(f'{side}_foot_index'): m[f'{side}_ankle_angle_deg'] = angle_deg(k,a,by[f'{side}_foot_index']) if all(k in by for k in ['left_shoulder','right_shoulder','left_hip','right_hip','left_ankle','right_ankle','left_knee','right_knee']): shoulder_mid = midpoint(by['left_shoulder'], by['right_shoulder'], 'shoulder_mid') hip_mid = midpoint(by['left_hip'], by['right_hip'], 'hip_mid') ankle_mid = midpoint(by['left_ankle'], by['right_ankle'], 'ankle_mid') knee_mid = midpoint(by['left_knee'], by['right_knee'], 'knee_mid') devs = [d for d in [point_line_distance(hip_mid, shoulder_mid, ankle_mid), point_line_distance(knee_mid, shoulder_mid, ankle_mid)] if d is not None] if devs: m['body_line_deviation_2d_norm'] = max(devs) m['root_y_norm'] = float(hip_mid['y']) m['thorax_y_norm'] = float(shoulder_mid['y']) # Pushup phase hint only when relevant. if 'push' in norm(exercise_key): elbows = [float(v) for k,v in m.items() if k.endswith('_elbow_angle_deg') and v is not None] if elbows: avg = sum(elbows)/len(elbows) m['avg_elbow_angle_deg'] = avg m['phase_guess_from_elbow'] = 'top_or_near_top' if avg > 145 else ('bottom_or_near_bottom' if avg < 115 else 'mid_range') return m def draw_annotation(image_bgr: Any, lms: List[Dict[str, Any]], out_path: Path) -> None: import cv2 h, w = image_bgr.shape[:2] annotated = image_bgr.copy() for a,b in POSE_CONNECTIONS: if a < len(lms) and b < len(lms): la, lb = lms[a], lms[b] if float(la.get('visibility',0)) < 0.15 or float(lb.get('visibility',0)) < 0.15: continue pa = (int(float(la['x'])*w), int(float(la['y'])*h)) pb = (int(float(lb['x'])*w), int(float(lb['y'])*h)) cv2.line(annotated, pa, pb, (255,255,255), 2) for lm in lms: if float(lm.get('visibility',0)) < 0.15: continue p = (int(float(lm['x'])*w), int(float(lm['y'])*h)) cv2.circle(annotated, p, 3, (0,255,0), -1) out_path.parent.mkdir(parents=True, exist_ok=True) cv2.imwrite(str(out_path), annotated) def process() -> int: args = parse_args() image_root = Path(args.image_root) out_root = Path(args.out) model_path = Path(args.pose_model) if not model_path.exists(): print(f'Pose model not found: {model_path}', file=sys.stderr) print('Pass --pose-model /path/to/pose_landmarker_full.task. Keep model files local; do not commit large binaries unless project policy allows.', file=sys.stderr) return 2 index = load_json(args.exercise_index) plan = load_json(args.source_plan) exercises = select_exercises(index, plan, image_root, args.exercise) if not exercises: print('No exercises selected. Provide --exercise-index/--source-plan or folders under --image-root.', file=sys.stderr) return 2 try: import cv2 import mediapipe as mp except Exception: print('Missing dependency. Install: pip install mediapipe opencv-python', file=sys.stderr) raise BaseOptions = mp.tasks.BaseOptions PoseLandmarker = mp.tasks.vision.PoseLandmarker PoseLandmarkerOptions = mp.tasks.vision.PoseLandmarkerOptions VisionRunningMode = mp.tasks.vision.RunningMode opts = PoseLandmarkerOptions( base_options=BaseOptions( model_asset_path=str(model_path), delegate=BaseOptions.Delegate.CPU, ), running_mode=VisionRunningMode.IMAGE, num_poses=1, ) summary = {'exercisesSelected': len(exercises), 'exercisesWithImages': 0, 'imagesProcessed': 0, 'outputs': []} out_root.mkdir(parents=True, exist_ok=True) with PoseLandmarker.create_from_options(opts) as detector: for ex in exercises: key = str(ex.get('exerciseKey') or norm(ex.get('displayName','exercise'))) keyposes = ex.get('keyposes') or [] images = collect_images_for_exercise(image_root, key) if not images: continue summary['exercisesWithImages'] += 1 records = [] for idx, img_path in enumerate(images): image_bgr = cv2.imread(str(img_path)) if image_bgr is None: records.append({'sourceImage': str(img_path), 'error': 'cv2 could not read image'}) continue mp_image = mp.Image.create_from_file(str(img_path)) result = detector.detect(mp_image) lms: List[Dict[str, Any]] = [] world_lms: List[Dict[str, Any]] = [] if result.pose_landmarks: lms = [lm_to_dict(lm, LANDMARK_NAMES[i] if i < len(LANDMARK_NAMES) else f'landmark_{i}') for i,lm in enumerate(result.pose_landmarks[0])] if result.pose_world_landmarks: world_lms = [lm_to_dict(lm, LANDMARK_NAMES[i] if i < len(LANDMARK_NAMES) else f'landmark_{i}') for i,lm in enumerate(result.pose_world_landmarks[0])] keypose_id = infer_keypose_id(img_path, idx, keyposes) rec = { 'keyposeId': keypose_id, 'sourceImage': str(img_path), 'imageWidth': int(image_bgr.shape[1]), 'imageHeight': int(image_bgr.shape[0]), 'hasPose': bool(lms), 'landmarks': lms, 'worldLandmarks': world_lms, 'derivedNEXT11Candidates': derived_candidates(lms), 'candidateMetrics': generic_metrics(lms, key), 'quality': { 'meanVisibility': generic_metrics(lms, key).get('mean_pose_visibility', 0) if lms else 0, 'candidateOnly': True, 'notApprovedIK': True, } } if not args.no_annotated and lms: ann_path = out_root / key / 'annotated' / f'{idx+1:02d}_{keypose_id}_{img_path.stem}.annotated.jpg' draw_annotation(image_bgr, lms, ann_path) rec['annotatedImage'] = str(ann_path) records.append(rec) summary['imagesProcessed'] += 1 package = { 'schema': 'next11.mediapipe_keypose_image_candidates.v1', 'generatedAt': now_iso(), 'exerciseKey': key, 'displayName': ex.get('displayName') or key, 'approvalStatus': 'candidate_only_not_approved_ik', 'sourceMode': 'online_or_manual_keypose_still_images', 'ikProvenance': ex.get('sources', []), 'expectedKeyposesFromPlan': keyposes, 'records': records, 'nextStep': 'Physio/IK QA -> Exercise Base sacred points -> Nex equality verifier', } out_json = out_root / key / f'{key}.mediapipe_keypose_candidates.json' out_json.parent.mkdir(parents=True, exist_ok=True) out_json.write_text(json.dumps(package, indent=2, ensure_ascii=False), encoding='utf-8') summary['outputs'].append(str(out_json)) summary_path = out_root / 'mediapipe_keypose_candidate_summary.json' summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding='utf-8') print(json.dumps(summary, indent=2, ensure_ascii=False)) return 0 if __name__ == '__main__': raise SystemExit(process())