""" BiomechanicsAgent — extracts named, documented, unit-bearing measurements from pose data. Input: Pose2DResult (or Body3DResult if used), MovementResult Output: BiomechFeatures(test_name, view, angles, alignments, ...) Failure: returns BiomechFeatures with confidence=0.0 and notes. Params: 0 (pure computation — no model). License: n/a. Gated: no. This module is MEASUREMENT ONLY — no scoring happens here. Scoring is done by the rubric functions in formscout/rubric/. """ from __future__ import annotations import math from typing import Any from formscout.types import ( Pose2DResult, Body3DResult, MovementResult, BiomechFeatures, ) from formscout import config def _angle_between_points(a: tuple, b: tuple, c: tuple) -> float: """ Compute angle at point b formed by segments ba and bc. Returns degrees. Returns NaN if any point is missing. """ try: ba = (a[0] - b[0], a[1] - b[1]) bc = (c[0] - b[0], c[1] - b[1]) dot = ba[0] * bc[0] + ba[1] * bc[1] mag_ba = math.sqrt(ba[0] ** 2 + ba[1] ** 2) mag_bc = math.sqrt(bc[0] ** 2 + bc[1] ** 2) if mag_ba == 0 or mag_bc == 0: return float("nan") cos_angle = max(-1.0, min(1.0, dot / (mag_ba * mag_bc))) return math.degrees(math.acos(cos_angle)) except (TypeError, IndexError, ZeroDivisionError): return float("nan") def _get_joint(keypoints: dict, joint_id: int) -> tuple | None: """Extract (x, y) for a joint, or None if missing/low-confidence.""" j = keypoints.get(joint_id) if j is None: return None if j.get("conf", 0) < config.POSE_CONF_THRESHOLD: return None return (j["x"], j["y"]) # COCO joint indices NOSE, L_EYE, R_EYE, L_EAR, R_EAR = 0, 1, 2, 3, 4 L_SHOULDER, R_SHOULDER = 5, 6 L_ELBOW, R_ELBOW = 7, 8 L_WRIST, R_WRIST = 9, 10 L_HIP, R_HIP = 11, 12 L_KNEE, R_KNEE = 13, 14 L_ANKLE, R_ANKLE = 15, 16 class BiomechanicsAgent: """Pure-function biomechanics measurement — no model calls.""" def run( self, pose2d: Pose2DResult, body3d: Body3DResult, movement: MovementResult, ) -> BiomechFeatures: if not pose2d.keypoints: return BiomechFeatures( test_name=movement.test_name, view="2d", side=movement.side, angles={}, alignments={}, symmetry_delta=None, timing={}, confidence=0.0, notes="no keypoints available", ) view = "3d" if body3d.used else "2d" dispatch = { "deep_squat": self._deep_squat, "hurdle_step": self._hurdle_step, "inline_lunge": self._inline_lunge, "shoulder_mobility": self._shoulder_mobility, "active_slr": self._active_slr, "trunk_stability_pushup": self._trunk_stability_pushup, "rotary_stability": self._rotary_stability, } fn = dispatch.get(movement.test_name) if fn is None: return BiomechFeatures( test_name=movement.test_name, view=view, side=movement.side, angles={}, alignments={}, symmetry_delta=None, timing={}, confidence=0.0, notes=f"unknown test: {movement.test_name}", ) return fn(pose2d, view, movement.side) def _deep_squat(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """Extract deep squat biomechanics from the deepest frame.""" # Find the frame with lowest hip Y (deepest squat position) best_frame_idx = 0 lowest_hip_y = -1.0 for i, kps in enumerate(pose2d.keypoints): l_hip = _get_joint(kps, L_HIP) r_hip = _get_joint(kps, R_HIP) if l_hip and r_hip: mid_hip_y = (l_hip[1] + r_hip[1]) / 2 if mid_hip_y > lowest_hip_y: # higher Y = lower in image lowest_hip_y = mid_hip_y best_frame_idx = i kps = pose2d.keypoints[best_frame_idx] notes_parts: list[str] = [] # Extract joints l_hip = _get_joint(kps, L_HIP) r_hip = _get_joint(kps, R_HIP) l_knee = _get_joint(kps, L_KNEE) r_knee = _get_joint(kps, R_KNEE) l_ankle = _get_joint(kps, L_ANKLE) r_ankle = _get_joint(kps, R_ANKLE) l_shoulder = _get_joint(kps, L_SHOULDER) r_shoulder = _get_joint(kps, R_SHOULDER) # Compute angles angles: dict[str, float] = {} # Hip-knee-ankle angle (knee flexion) — average of both sides l_knee_angle = _angle_between_points(l_hip, l_knee, l_ankle) if all([l_hip, l_knee, l_ankle]) else float("nan") r_knee_angle = _angle_between_points(r_hip, r_knee, r_ankle) if all([r_hip, r_knee, r_ankle]) else float("nan") if not math.isnan(l_knee_angle): angles["left_knee_flexion_deg"] = l_knee_angle else: notes_parts.append("left knee angle unavailable") if not math.isnan(r_knee_angle): angles["right_knee_flexion_deg"] = r_knee_angle else: notes_parts.append("right knee angle unavailable") # Femur angle from horizontal # Femur = hip to knee. Angle from horizontal = atan2(dy, dx) if l_hip and l_knee: dy = l_knee[1] - l_hip[1] dx = l_knee[0] - l_hip[0] angles["left_femur_from_horizontal_deg"] = abs(math.degrees(math.atan2(dy, dx))) if r_hip and r_knee: dy = r_knee[1] - r_hip[1] dx = r_knee[0] - r_hip[0] angles["right_femur_from_horizontal_deg"] = abs(math.degrees(math.atan2(dy, dx))) # Torso-tibia angle (torso parallel to tibia = score 3 criterion) if l_shoulder and l_hip and l_knee and l_ankle: torso_angle = math.degrees(math.atan2(l_hip[1] - l_shoulder[1], l_hip[0] - l_shoulder[0])) tibia_angle = math.degrees(math.atan2(l_ankle[1] - l_knee[1], l_ankle[0] - l_knee[0])) angles["torso_tibia_angle_deg"] = abs(torso_angle - tibia_angle) # Alignments alignments: dict[str, Any] = {} # Knee valgus check: are knees inside the ankle line? if l_knee and r_knee and l_ankle and r_ankle: knee_width = abs(l_knee[0] - r_knee[0]) ankle_width = abs(l_ankle[0] - r_ankle[0]) alignments["knees_tracking_over_feet"] = knee_width >= (ankle_width - config.DEEP_SQUAT_KNEE_TRACKING_MARGIN_PX) alignments["knee_valgus_deg"] = 0.0 # placeholder for actual valgus angle # Heels elevated detection (approximation: ankle Y relative to frame bottom) # This is a rough heuristic — proper detection needs foot keypoints or depth alignments["heels_elevated"] = False # default; refine with better detection # Dowel position (need wrist positions relative to feet) if l_wrist := _get_joint(kps, L_WRIST): if r_wrist := _get_joint(kps, R_WRIST): if l_ankle and r_ankle: mid_wrist_x = (l_wrist[0] + r_wrist[0]) / 2 mid_ankle_x = (l_ankle[0] + r_ankle[0]) / 2 alignments["dowel_over_feet"] = abs(mid_wrist_x - mid_ankle_x) < 50 alignments["dowel_feet_offset_px"] = mid_wrist_x - mid_ankle_x # Confidence based on how many measurements we got n_expected = 6 # main measurements n_got = len(angles) + len([v for v in alignments.values() if v is not None]) confidence = min(1.0, n_got / n_expected) * pose2d.confidence return BiomechFeatures( test_name="deep_squat", view=view, side="na", angles=angles, alignments=alignments, symmetry_delta=None, timing={"deepest_frame": best_frame_idx}, confidence=confidence, notes="; ".join(notes_parts) if notes_parts else "", ) # ─── Helper: find peak frame by joint Y ───────────────────────────────── def _find_peak_frame(self, pose2d: Pose2DResult, joint_id: int, maximize: bool = True) -> int: """Find frame where a joint reaches its extreme Y position.""" best_idx, best_val = 0, -1.0 if maximize else float("inf") for i, kps in enumerate(pose2d.keypoints): j = _get_joint(kps, joint_id) if j: if (maximize and j[1] > best_val) or (not maximize and j[1] < best_val): best_val = j[1] best_idx = i return best_idx def _bilateral_features( self, pose2d: Pose2DResult, view: str, side: str, test_name: str, extractor, ) -> BiomechFeatures: """Run a bilateral test: compute both sides, report the specified side + symmetry_delta.""" left = extractor(pose2d, "left") right = extractor(pose2d, "right") # Pick the requested side as primary primary = left if side == "left" else right if side == "right" else left other = right if side == "left" else left if side == "right" else right # Merge angles with side prefix for the primary angles = primary.get("angles", {}) alignments = primary.get("alignments", {}) timing = primary.get("timing", {}) # Compute symmetry delta from the main measurement main_key = primary.get("main_measure_key") sym_delta = None if main_key and main_key in left.get("angles", {}) and main_key in right.get("angles", {}): sym_delta = abs(left["angles"][main_key] - right["angles"][main_key]) n_got = len(angles) + len([v for v in alignments.values() if v is not None]) confidence = min(1.0, n_got / max(primary.get("expected", 3), 1)) * pose2d.confidence return BiomechFeatures( test_name=test_name, view=view, side=side, angles=angles, alignments=alignments, symmetry_delta=sym_delta, timing=timing, confidence=confidence, notes=primary.get("notes", ""), ) # ─── Hurdle Step ───────────────────────────────────────────────────────── def _hurdle_step(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """Hurdle Step: hip/knee flexion of stepping leg, stance stability.""" def extract(p2d: Pose2DResult, s: str) -> dict: hip_id = L_HIP if s == "left" else R_HIP knee_id = L_KNEE if s == "left" else R_KNEE ankle_id = L_ANKLE if s == "left" else R_ANKLE # Stance side is opposite stance_hip = R_HIP if s == "left" else L_HIP stance_knee = R_KNEE if s == "left" else L_KNEE stance_ankle = R_ANKLE if s == "left" else L_ANKLE # Peak = frame where stepping knee is highest (lowest Y in image) peak_idx = self._find_peak_frame(p2d, knee_id, maximize=False) kps = p2d.keypoints[peak_idx] hip = _get_joint(kps, hip_id) knee = _get_joint(kps, knee_id) ankle = _get_joint(kps, ankle_id) s_hip = _get_joint(kps, stance_hip) s_knee = _get_joint(kps, stance_knee) s_ankle = _get_joint(kps, stance_ankle) angles = {} alignments = {} notes_parts = [] # Hip flexion of stepping leg if all([hip, knee, ankle]): angles["step_knee_flexion_deg"] = _angle_between_points(hip, knee, ankle) # Hip angle (torso-femur) shoulder_id = L_SHOULDER if s == "left" else R_SHOULDER shoulder = _get_joint(kps, shoulder_id) if all([shoulder, hip, knee]): angles["step_hip_flexion_deg"] = _angle_between_points(shoulder, hip, knee) # Stance knee should stay extended if all([s_hip, s_knee, s_ankle]): angles["stance_knee_angle_deg"] = _angle_between_points(s_hip, s_knee, s_ankle) alignments["stance_knee_extended"] = angles["stance_knee_angle_deg"] > 160 # Lateral trunk lean: shoulders should be level l_sh = _get_joint(kps, L_SHOULDER) r_sh = _get_joint(kps, R_SHOULDER) if l_sh and r_sh: angles["shoulder_tilt_deg"] = abs(math.degrees( math.atan2(r_sh[1] - l_sh[1], r_sh[0] - l_sh[0]) )) alignments["trunk_stable"] = angles["shoulder_tilt_deg"] < 10 return { "angles": angles, "alignments": alignments, "timing": {"peak_step_frame": peak_idx}, "main_measure_key": "step_hip_flexion_deg", "expected": 4, "notes": "; ".join(notes_parts), } return self._bilateral_features(pose2d, view, side, "hurdle_step", extract) # ─── In-Line Lunge ─────────────────────────────────────────────────────── def _inline_lunge(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """In-Line Lunge: knee flexion depth, trunk upright, balance.""" def extract(p2d: Pose2DResult, s: str) -> dict: # Front leg is the assessed side hip_id = L_HIP if s == "left" else R_HIP knee_id = L_KNEE if s == "left" else R_KNEE ankle_id = L_ANKLE if s == "left" else R_ANKLE rear_knee_id = R_KNEE if s == "left" else L_KNEE # Deepest lunge = front knee lowest peak_idx = self._find_peak_frame(p2d, knee_id, maximize=True) kps = p2d.keypoints[peak_idx] hip = _get_joint(kps, hip_id) knee = _get_joint(kps, knee_id) ankle = _get_joint(kps, ankle_id) l_sh = _get_joint(kps, L_SHOULDER) r_sh = _get_joint(kps, R_SHOULDER) l_hip = _get_joint(kps, L_HIP) r_hip = _get_joint(kps, R_HIP) angles = {} alignments = {} # Front knee flexion if all([hip, knee, ankle]): angles["front_knee_flexion_deg"] = _angle_between_points(hip, knee, ankle) # Trunk upright: midline shoulder-to-hip angle from vertical if l_sh and r_sh and l_hip and r_hip: mid_sh = ((l_sh[0] + r_sh[0]) / 2, (l_sh[1] + r_sh[1]) / 2) mid_hip = ((l_hip[0] + r_hip[0]) / 2, (l_hip[1] + r_hip[1]) / 2) trunk_from_vert = abs(math.degrees( math.atan2(mid_hip[0] - mid_sh[0], mid_sh[1] - mid_hip[1]) )) angles["trunk_lean_from_vertical_deg"] = trunk_from_vert alignments["trunk_upright"] = trunk_from_vert < 15 # Knee over ankle alignment if knee and ankle: alignments["knee_over_ankle"] = abs(knee[0] - ankle[0]) < 40 return { "angles": angles, "alignments": alignments, "timing": {"deepest_lunge_frame": peak_idx}, "main_measure_key": "front_knee_flexion_deg", "expected": 3, "notes": "", } return self._bilateral_features(pose2d, view, side, "inline_lunge", extract) # ─── Shoulder Mobility ─────────────────────────────────────────────────── def _shoulder_mobility(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """Shoulder Mobility: inter-fist distance normalized to hand length.""" def extract(p2d: Pose2DResult, s: str) -> dict: # "side" = the hand reaching over (top hand) top_wrist = L_WRIST if s == "left" else R_WRIST bot_wrist = R_WRIST if s == "left" else L_WRIST # Use mid-sequence frame (static hold) mid_idx = len(p2d.keypoints) // 2 kps = p2d.keypoints[mid_idx] top_w = _get_joint(kps, top_wrist) bot_w = _get_joint(kps, bot_wrist) angles = {} alignments = {} if top_w and bot_w: # Vertical distance between fists (normalized by torso length) fist_dist_px = math.sqrt((top_w[0] - bot_w[0])**2 + (top_w[1] - bot_w[1])**2) angles["inter_fist_distance_px"] = fist_dist_px # Normalize by torso length (shoulder to hip) sh_id = L_SHOULDER if s == "left" else R_SHOULDER hip_id = L_HIP if s == "left" else R_HIP sh = _get_joint(kps, sh_id) hip = _get_joint(kps, hip_id) if sh and hip: torso_len = math.sqrt((sh[0] - hip[0])**2 + (sh[1] - hip[1])**2) if torso_len > 0: norm_dist = fist_dist_px / torso_len angles["inter_fist_normalized"] = norm_dist # Score 3: fists within 1 hand-length (~0.3 torso) # Score 2: within 1.5 hand-lengths alignments["fists_within_one_hand"] = norm_dist < 0.35 alignments["fists_within_1_5_hand"] = norm_dist < 0.55 return { "angles": angles, "alignments": alignments, "timing": {"measure_frame": mid_idx}, "main_measure_key": "inter_fist_normalized", "expected": 2, "notes": "", } return self._bilateral_features(pose2d, view, side, "shoulder_mobility", extract) # ─── Active Straight-Leg Raise ─────────────────────────────────────────── def _active_slr(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """ASLR: hip flexion angle of raised leg; down-leg stays flat.""" def extract(p2d: Pose2DResult, s: str) -> dict: hip_id = L_HIP if s == "left" else R_HIP knee_id = L_KNEE if s == "left" else R_KNEE ankle_id = L_ANKLE if s == "left" else R_ANKLE # Down leg d_hip_id = R_HIP if s == "left" else L_HIP d_knee_id = R_KNEE if s == "left" else L_KNEE d_ankle_id = R_ANKLE if s == "left" else L_ANKLE # Peak = raised ankle at highest point (lowest Y) peak_idx = self._find_peak_frame(p2d, ankle_id, maximize=False) kps = p2d.keypoints[peak_idx] hip = _get_joint(kps, hip_id) knee = _get_joint(kps, knee_id) ankle = _get_joint(kps, ankle_id) d_hip = _get_joint(kps, d_hip_id) d_knee = _get_joint(kps, d_knee_id) d_ankle = _get_joint(kps, d_ankle_id) angles = {} alignments = {} # Raised leg hip flexion: angle of femur from horizontal if hip and ankle: dy = hip[1] - ankle[1] # positive = ankle above hip dx = ankle[0] - hip[0] hip_flex = math.degrees(math.atan2(dy, abs(dx) if abs(dx) > 1 else 1)) angles["raised_leg_angle_deg"] = max(0, hip_flex) # Score 3: malleolus past contralateral knee (>70°) # Score 2: between contralateral knee and mid-thigh (45-70°) alignments["past_contralateral_knee"] = hip_flex > 70 alignments["past_mid_thigh"] = hip_flex > 45 # Down leg: should stay flat (knee angle ~180) if all([d_hip, d_knee, d_ankle]): down_knee_angle = _angle_between_points(d_hip, d_knee, d_ankle) angles["down_leg_knee_angle_deg"] = down_knee_angle alignments["down_leg_flat"] = down_knee_angle > 160 return { "angles": angles, "alignments": alignments, "timing": {"peak_raise_frame": peak_idx}, "main_measure_key": "raised_leg_angle_deg", "expected": 3, "notes": "", } return self._bilateral_features(pose2d, view, side, "active_slr", extract) # ─── Trunk Stability Push-Up ───────────────────────────────────────────── def _trunk_stability_pushup(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """Trunk Stability Push-Up: body rigidity through the press.""" angles = {} alignments = {} notes_parts = [] # Analyze multiple frames to detect sag/lag trunk_sags: list[tuple[int, float]] = [] # (frame_idx, sag_px) for i, kps in enumerate(pose2d.keypoints): l_sh = _get_joint(kps, L_SHOULDER) r_sh = _get_joint(kps, R_SHOULDER) l_hip = _get_joint(kps, L_HIP) r_hip = _get_joint(kps, R_HIP) l_ankle = _get_joint(kps, L_ANKLE) r_ankle = _get_joint(kps, R_ANKLE) if l_sh and r_sh and l_hip and r_hip and l_ankle and r_ankle: # Sag = hip drops below shoulder-ankle line sh_y = (l_sh[1] + r_sh[1]) / 2 hip_y = (l_hip[1] + r_hip[1]) / 2 ankle_y = (l_ankle[1] + r_ankle[1]) / 2 # In image coords: sag = hip_y > midpoint of shoulder-ankle Y expected_hip_y = (sh_y + ankle_y) / 2 sag_px = hip_y - expected_hip_y trunk_sags.append((i, sag_px)) max_sag_frame = 0 if trunk_sags: sags = [s for _, s in trunk_sags] max_sag_frame = max(trunk_sags, key=lambda t: t[1])[0] mean = sum(sags) / len(sags) variance = (sum((x - mean) ** 2 for x in sags) / len(sags)) ** 0.5 max_sag = max(sags) angles["max_sag_px"] = max_sag angles["trunk_variance_px"] = variance alignments["body_rigid"] = max_sag < 30 and variance < 15 alignments["no_sag"] = max_sag < 30 else: notes_parts.append("insufficient landmarks for trunk analysis") # Hand position (near head = harder = score 3 position) if pose2d.keypoints: mid_kps = pose2d.keypoints[0] nose = _get_joint(mid_kps, NOSE) l_w = _get_joint(mid_kps, L_WRIST) r_w = _get_joint(mid_kps, R_WRIST) if nose and l_w and r_w: avg_wrist_y = (l_w[1] + r_w[1]) / 2 # Hands near head = wrist Y close to nose Y alignments["hands_at_forehead"] = abs(avg_wrist_y - nose[1]) < 50 n_got = len(angles) + len([v for v in alignments.values() if v is not None]) confidence = min(1.0, n_got / 3) * pose2d.confidence return BiomechFeatures( test_name="trunk_stability_pushup", view=view, side="na", angles=angles, alignments=alignments, symmetry_delta=None, timing={"n_frames_analyzed": len(trunk_sags), "max_sag_frame": max_sag_frame}, confidence=confidence, notes="; ".join(notes_parts) if notes_parts else "", ) # ─── Rotary Stability ──────────────────────────────────────────────────── def _rotary_stability(self, pose2d: Pose2DResult, view: str, side: str) -> BiomechFeatures: """Rotary Stability: coordination of ipsilateral arm/leg extension.""" angles = {} alignments = {} notes_parts = [] # Look for the frame with max arm+leg extension # Quadruped: hands + knees on ground, extending one arm + one leg best_ext_frame = 0 best_ext_val = 0 for i, kps in enumerate(pose2d.keypoints): l_w = _get_joint(kps, L_WRIST) r_w = _get_joint(kps, R_WRIST) l_a = _get_joint(kps, L_ANKLE) r_a = _get_joint(kps, R_ANKLE) l_sh = _get_joint(kps, L_SHOULDER) r_sh = _get_joint(kps, R_SHOULDER) # Extension = distance of wrist from shoulder + ankle from hip ext_val = 0 if l_w and l_sh: ext_val += abs(l_w[0] - l_sh[0]) if r_w and r_sh: ext_val += abs(r_w[0] - r_sh[0]) if ext_val > best_ext_val: best_ext_val = ext_val best_ext_frame = i kps = pose2d.keypoints[best_ext_frame] if pose2d.keypoints else {} # Trunk stability: shoulders level, hips level l_sh = _get_joint(kps, L_SHOULDER) r_sh = _get_joint(kps, R_SHOULDER) l_hip = _get_joint(kps, L_HIP) r_hip = _get_joint(kps, R_HIP) if l_sh and r_sh: sh_tilt = abs(l_sh[1] - r_sh[1]) angles["shoulder_level_diff_px"] = sh_tilt alignments["shoulders_level"] = sh_tilt < 20 if l_hip and r_hip: hip_tilt = abs(l_hip[1] - r_hip[1]) angles["hip_level_diff_px"] = hip_tilt alignments["hips_level"] = hip_tilt < 20 # Check for trunk sag across frames (similar to pushup) trunk_variance = [] for kps_frame in pose2d.keypoints: ls = _get_joint(kps_frame, L_SHOULDER) rs = _get_joint(kps_frame, R_SHOULDER) lh = _get_joint(kps_frame, L_HIP) rh = _get_joint(kps_frame, R_HIP) if ls and rs and lh and rh: mid_sh_y = (ls[1] + rs[1]) / 2 mid_hip_y = (lh[1] + rh[1]) / 2 trunk_variance.append(mid_hip_y - mid_sh_y) if trunk_variance: std = (sum((x - sum(trunk_variance) / len(trunk_variance))**2 for x in trunk_variance) / len(trunk_variance)) ** 0.5 angles["trunk_stability_std_px"] = std alignments["trunk_stable"] = std < 15 n_got = len(angles) + len([v for v in alignments.values() if v is not None]) confidence = min(1.0, n_got / 3) * pose2d.confidence return BiomechFeatures( test_name="rotary_stability", view=view, side="na", angles=angles, alignments=alignments, symmetry_delta=None, timing={"peak_extension_frame": best_ext_frame}, confidence=confidence, notes="; ".join(notes_parts) if notes_parts else "", )