Upload folder using huggingface_hub

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+example/Sprint_Assessment_Workflow.ipynb filter=lfs diff=lfs merge=lfs -text

README.md

@@ -91,6 +91,6 @@ results.save("output_video.mp4")
 
 ## License
 
-```apache 2.0
-This project is licensed under the Apache 2.0 License.
+```mit
+This project is licensed under the MIT License.
 ```

example/Sprint_Assessment_Workflow.ipynb

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:eb6f86383c9517d10213b276f73981f0c5f83634361ba775c826b114ae29673d
+size 22298843

example/annotate_video.py

@@ -0,0 +1,59 @@
+import os
+import argparse
+import pandas as pd
+import cv2
+import numpy as np
+
+# Skeleton definition
+SKELETON = [
+    ("left_ankle", "left_knee"), ("left_knee", "left_hip"), 
+    ("right_ankle", "right_knee"), ("right_knee", "right_hip"),
+    ("left_hip", "right_hip"), ("left_shoulder", "right_shoulder"),
+    ("left_shoulder", "left_hip"), ("right_shoulder", "right_hip"),
+    ("left_shoulder", "left_elbow"), ("left_elbow", "left_wrist"),
+    ("right_shoulder", "right_elbow"), ("right_elbow", "right_wrist"),
+    ("left_ankle", "left_heel"), ("left_ankle", "left_big_toe"),
+    ("right_ankle", "right_heel"), ("right_ankle", "right_big_toe")
+]
+
+def annotate_frames(frames_dir, output_dir, csv_path):
+    """Draw keypoints and skeleton on frames."""
+    df = pd.read_csv(csv_path)
+    os.makedirs(output_dir, exist_ok=True)
+    
+    frame_files = sorted([f for f in os.listdir(frames_dir) if f.endswith('.jpg')])
+    
+    for _, row in df.iterrows():
+        f_idx = int(row['frame'])
+        if f_idx >= len(frame_files): continue
+        
+        frame_name = frame_files[f_idx]
+        img = cv2.imread(os.path.join(frames_dir, frame_name))
+        
+        # Draw Skeleton
+        for start_kp, end_kp in SKELETON:
+            s_x, s_y = row.get(f"{start_kp}_x"), row.get(f"{start_kp}_y")
+            e_x, e_y = row.get(f"{end_kp}_x"), row.get(f"{end_kp}_y")
+            if pd.notna(s_x) and pd.notna(e_x):
+                cv2.line(img, (int(s_x), int(s_y)), (int(e_x), int(e_y)), (0, 255, 0), 2)
+        
+        # Draw Keypoints
+        for col in df.columns:
+            if col.endswith('_x'):
+                kp_name = col[:-2]
+                x, y = row[col], row[f"{kp_name}_y"]
+                if pd.notna(x):
+                    cv2.circle(img, (int(x), int(y)), 4, (0, 0, 255), -1)
+                    
+        cv2.imwrite(os.path.join(output_dir, frame_name), img)
+        if f_idx % 100 == 0:
+            print(f"Annotated {f_idx} frames")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--frames", required=True, help="Input frames directory")
+    parser.add_argument("--csv", required=True, help="Input keypoints CSV")
+    parser.add_argument("--output", required=True, help="Output frames directory")
+    args = parser.parse_args()
+    
+    annotate_frames(args.frames, args.output, args.csv)

example/inference_yolo_pose.py

@@ -0,0 +1,77 @@
+import os
+import argparse
+from ultralytics import YOLO
+import cv2
+from pathlib import Path
+
+# Keypoint schema (17 body + 6 feet = 23 total)
+COCO_BODY_17 = [
+    "nose", "left_eye", "right_eye", "left_ear", "right_ear",
+    "left_shoulder", "right_shoulder", "left_elbow", "right_elbow",
+    "left_wrist", "right_wrist", "left_hip", "right_hip",
+    "left_knee", "right_knee", "left_ankle", "right_ankle"
+]
+
+FEET_6_LABELS = [
+    "left_heel", "left_big_toe", "left_little_toe",
+    "right_heel", "right_big_toe", "right_little_toe"
+]
+
+ALL_KEYPOINTS = COCO_BODY_17 + FEET_6_LABELS
+
+def run_inference(frames_dir, labels_dir, model_path, img_width=1920, img_height=1080):
+    """Run YOLO pose inference on all frames and save labels."""
+    model = YOLO(model_path)
+    os.makedirs(labels_dir, exist_ok=True)
+    
+    frame_files = sorted([f for f in os.listdir(frames_dir) if f.endswith('.jpg')])
+    print(f"Running inference on {len(frame_files)} frames...")
+
+    for idx, frame_file in enumerate(frame_files):
+        frame_path = os.path.join(frames_dir, frame_file)
+        results = model(frame_path, verbose=False)
+
+        label_file = os.path.join(labels_dir, frame_file.replace('.jpg', '.txt'))
+
+        with open(label_file, 'w') as f:
+            for result in results:
+                if result.keypoints is not None:
+                    for kp in result.keypoints.data:
+                        kp_np = kp.cpu().numpy()
+
+                        # Get bounding box (approximate from keypoints)
+                        valid_kp = kp_np[kp_np[:, 2] > 0]
+                        if len(valid_kp) == 0:
+                            continue
+
+                        x_min, y_min = valid_kp[:, 0].min(), valid_kp[:, 1].min()
+                        x_max, y_max = valid_kp[:, 0].max(), valid_kp[:, 1].max()
+                        bbox_x = (x_min + x_max) / 2 / img_width
+                        bbox_y = (y_min + y_max) / 2 / img_height
+                        bbox_w = (x_max - x_min) / img_width
+                        bbox_h = (y_max - y_min) / img_height
+
+                        # Write YOLO format: class_id bbox keypoints
+                        line = f"0 {bbox_x:.6f} {bbox_y:.6f} {bbox_w:.6f} {bbox_h:.6f}"
+                        for kp_point in kp_np:
+                            x_norm = kp_point[0] / img_width
+                            y_norm = kp_point[1] / img_height
+                            conf = kp_point[2]
+                            line += f" {x_norm:.6f} {y_norm:.6f} {conf:.2f}"
+                        f.write(line + "\n")
+
+        if (idx + 1) % 50 == 0:
+            print(f"Processed {idx + 1}/{len(frame_files)} frames")
+
+    print(f"✅ Inference complete. Labels saved to {labels_dir}")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--frames", required=True, help="Directory containing JPEG frames")
+    parser.add_argument("--labels", required=True, help="Output directory for labels")
+    parser.add_argument("--model", required=True, help="Path to YOLO weights (.pt)")
+    parser.add_argument("--width", type=int, default=1920)
+    parser.add_argument("--height", type=int, default=1080)
+    args = parser.parse_args()
+    
+    run_inference(args.frames, args.labels, args.model, args.width, args.height)

example/post_process_keypoints.py

@@ -0,0 +1,78 @@
+import os
+import argparse
+import pandas as pd
+import numpy as np
+from scipy.ndimage import gaussian_filter1d
+from pathlib import Path
+
+# Keypoint schema
+COCO_BODY_17 = [
+    "nose", "left_eye", "right_eye", "left_ear", "right_ear",
+    "left_shoulder", "right_shoulder", "left_elbow", "right_elbow",
+    "left_wrist", "right_wrist", "left_hip", "right_hip",
+    "left_knee", "right_knee", "left_ankle", "right_ankle"
+]
+FEET_6_LABELS = [
+    "left_heel", "left_big_toe", "left_little_toe",
+    "right_heel", "right_big_toe", "right_little_toe"
+]
+ALL_KEYPOINTS = COCO_BODY_17 + FEET_6_LABELS
+
+def parse_labels_to_df(labels_dir, img_width, img_height):
+    records = []
+    label_files = sorted([f for f in os.listdir(labels_dir) if f.endswith('.txt')])
+    for frame_idx, label_file in enumerate(label_files):
+        label_path = os.path.join(labels_dir, label_file)
+        with open(label_path, 'r') as f:
+            lines = f.readlines()
+        for line in lines:
+            parts = line.strip().split()
+            if len(parts) < 5: continue
+            keypoint_data = parts[5:]
+            kp_dict = {}
+            for i in range(0, len(keypoint_data), 3):
+                kp_idx = i // 3
+                if kp_idx >= len(ALL_KEYPOINTS): break
+                x_norm, y_norm, conf = float(keypoint_data[i]), float(keypoint_data[i+1]), float(keypoint_data[i+2])
+                if conf > 0:
+                    kp_dict[ALL_KEYPOINTS[kp_idx]] = {'x': x_norm * img_width, 'y': y_norm * img_height, 'conf': conf}
+            records.append({'frame': frame_idx, 'keypoints': kp_dict})
+    return pd.DataFrame(records)
+
+def process_keypoints(df, sigma=2.0):
+    """Expand, interpolate, and smooth keypoints."""
+    # Expand dictionary into columns
+    rows = []
+    for _, row in df.iterrows():
+        flat = {'frame': row['frame']}
+        for kp, vals in row['keypoints'].items():
+            flat[f"{kp}_x"] = vals['x']
+            flat[f"{kp}_y"] = vals['y']
+            flat[f"{kp}_conf"] = vals['conf']
+        rows.append(flat)
+    
+    expanded = pd.DataFrame(rows)
+    # Temporal interpolation and smoothing
+    for kp in ALL_KEYPOINTS:
+        for suffix in ['_x', '_y']:
+            col = f"{kp}{suffix}"
+            if col in expanded.columns:
+                expanded[col] = expanded[col].interpolate(method='linear').ffill().bfill()
+                expanded[col] = gaussian_filter1d(expanded[col], sigma=sigma)
+    
+    return expanded
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--labels", required=True, help="Input labels directory")
+    parser.add_argument("--output", required=True, help="Output CSV path")
+    parser.add_argument("--width", type=int, default=1920)
+    parser.add_argument("--height", type=int, default=1080)
+    parser.add_argument("--sigma", type=float, default=2.0)
+    args = parser.parse_args()
+    
+    print("Post-processing keypoints...")
+    df_raw = parse_labels_to_df(args.labels, args.width, args.height)
+    df_clean = process_keypoints(df_raw, sigma=args.sigma)
+    df_clean.to_csv(args.output, index=False)
+    print(f"✅ Cleaned keypoints saved to {args.output}")

example/run_full_pipeline.py

@@ -0,0 +1,47 @@
+import os
+import sys
+import argparse
+from video_utils import extract_frames, create_video_from_frames
+
+def run_pipeline(video_path, model_path, output_dir, sigma=2.0):
+    """Orchestrate the full pose estimation pipeline."""
+    # 1. Setup paths
+    frames_dir = os.path.join(output_dir, "frames")
+    labels_dir = os.path.join(output_dir, "labels_raw")
+    annotated_dir = os.path.join(output_dir, "annotated_frames")
+    keypoints_csv = os.path.join(output_dir, "yolo_keypoints.csv")
+    final_video = os.path.join(output_dir, "annotated_output.mp4")
+    
+    os.makedirs(output_dir, exist_ok=True)
+
+    # 2. Extract frames
+    print("--- Step 1: Extracting Frames ---")
+    fps, width, height, num_frames = extract_frames(video_path, frames_dir)
+
+    # 3. Inference
+    print("\n--- Step 2: Running YOLO Inference ---")
+    os.system(f'{sys.executable} inference_yolo_pose.py --frames "{frames_dir}" --labels "{labels_dir}" --model "{model_path}" --width {width} --height {height}')
+
+    # 4. Post-processing
+    print("\n--- Step 3: Post-processing Keypoints ---")
+    os.system(f'{sys.executable} post_process_keypoints.py --labels "{labels_dir}" --output "{keypoints_csv}" --width {width} --height {height} --sigma {sigma}')
+
+    # 5. Annotation
+    print("\n--- Step 4: Annotating Frames ---")
+    os.system(f'{sys.executable} annotate_video.py --frames "{frames_dir}" --csv "{keypoints_csv}" --output "{annotated_dir}"')
+
+    # 6. Final Video
+    print("\n--- Step 5: Creating Final Video ---")
+    create_video_from_frames(annotated_dir, final_video, fps, width, height)
+    
+    print(f"\n✅ Pipeline complete! Output saved to {output_dir}")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--video", required=True, help="Input video path")
+    parser.add_argument("--model", required=True, help="YOLO model weights path")
+    parser.add_argument("--output", default="video_output", help="Output directory")
+    parser.add_argument("--sigma", type=float, default=2.0, help="Smoothing sigma")
+    args = parser.parse_args()
+    
+    run_pipeline(args.video, args.model, args.output, args.sigma)

example/smas_pipeline_v2.py

@@ -0,0 +1,600 @@
+# smas_pipeline.py
+"""S‑MAS Biomechanics Analysis Pipeline
+
+Consumes 2‑D key‑point CSV output from the YOLO pose estimation notebook,
+detects sprint gait events, evaluates the 12 Sprint‑Mechanics Assessment Score (S‑MAS)
+criteria, and produces per‑stride and session‑level JSON results.
+
+The implementation follows the design described in the approved implementation plan.
+"""
+
+import json
+import numpy as np
+import pandas as pd
+from dataclasses import dataclass, field
+from typing import List, Dict, Tuple
+
+VERSION = "1.3.0-signal-based"
+print(f"--- S-MAS Pipeline version {VERSION} loaded ---")
+
+# -----------------------------------------------------------------------------
+# Data structures
+# -----------------------------------------------------------------------------
+@dataclass
+class FrameData:
+    idx: int
+    timestamp: float
+    keypoints: Dict[str, Tuple[float, float, float]]  # (x, y, confidence)
+    # derived quantities (filled later)
+    hip_angle: float = None
+    knee_angle: float = None
+    ankle_angle: float = None
+    trunk_angle: float = None
+    shin_angle: float = None
+    pelvis_x: float = None
+    pelvis_y: float = None
+    left_contact: bool = None
+    right_contact: bool = None
+    foot_contact: bool = None  # legacy/summary flag
+
+@dataclass
+class StrideResult:
+    stride_id: int
+    criterion_scores: Dict[str, int]
+    total_smas_score: int
+
+@dataclass
+class SessionResult:
+    mean_score: float
+    per_criterion_freq: Dict[str, float]
+    stride_variance: float = None
+    strides: List[StrideResult] = field(default_factory=list)
+
+# -----------------------------------------------------------------------------
+# Helper constants (tunable thresholds)
+# -----------------------------------------------------------------------------
+VERTICAL_THRESHOLD_PX = 5          # for foot‑contact detection
+PELVIS_ANKLE_DIST_PX = 5           # for mid‑stance detection
+HORIZONTAL_MOVEMENT_THRESHOLD_PX = 2.0  # for direction detection
+ANGLE_TOLERANCE_DEG = 5           # generic tolerance
+
+# -----------------------------------------------------------------------------
+# 1. Load YOLO key‑point CSV
+# -----------------------------------------------------------------------------
+def load_keypoints(csv_path: str) -> List[FrameData]:
+    df = pd.read_csv(csv_path)
+    frames: Dict[int, FrameData] = {}
+    
+    # Handle wide format: frame, frame_name, nose_x, ...
+    cols = df.columns
+    kp_names = set()
+    for c in cols:
+        if c.endswith('_x'):
+            kp_names.add(c[:-2])
+            
+    for _, row in df.iterrows():
+        f = int(row['frame'])
+        # If we have multiple rows for same frame (multi-person), take the first one
+        if f in frames:
+            continue
+            
+        fr_data = FrameData(
+            idx=f,
+            timestamp=row.get('time', f / 30.0),
+            keypoints={}
+        )
+        
+        for name in kp_names:
+            x = row.get(f"{name}_x")
+            y = row.get(f"{name}_y")
+            conf = row.get(f"{name}_conf", 0.5)
+            if not np.isnan(x) and not np.isnan(y):
+                fr_data.keypoints[name] = (float(x), float(y), float(conf))
+        
+        frames[f] = fr_data
+        
+    return [frames[i] for i in sorted(frames)]
+
+# -----------------------------------------------------------------------------
+# 2. Global Direction Detection
+# -----------------------------------------------------------------------------
+def detect_global_direction(frames: List[FrameData]) -> str:
+    """Detect the general running direction based on frame-to-frame displacement (delta-x)."""
+    if len(frames) < 2:
+        return 'left→right' # Default
+        
+    deltas = []
+    prev_x = None
+    
+    for fr in frames:
+        # We use a stable point like pelvis or nose. 
+        # Here we try to see if pelvis_x is already computed, if not we use nose_x if available.
+        curr_x = None
+        if fr.pelvis_x is not None:
+            curr_x = fr.pelvis_x
+        elif 'nose' in fr.keypoints:
+            curr_x = fr.keypoints['nose'][0]
+            
+        if curr_x is not None:
+            if prev_x is not None:
+                delta = curr_x - prev_x
+                if abs(delta) > HORIZONTAL_MOVEMENT_THRESHOLD_PX:
+                    deltas.append(delta)
+            prev_x = curr_x
+            
+    if not deltas:
+        print("DEBUG: No significant horizontal movement detected. Defaulting to left→right.")
+        return 'left→right'
+        
+    avg_delta = np.mean(deltas)
+    if avg_delta > 0:
+        print(f"DEBUG: Detected direction 'left→right' (avg delta_x: {avg_delta:.2f})")
+        return 'left→right'
+    else:
+        print(f"DEBUG: Detected direction 'right→left' (avg delta_x: {avg_delta:.2f})")
+        return 'right→left'
+
+# -----------------------------------------------------------------------------
+# 2. Compute derived geometric quantities
+# -----------------------------------------------------------------------------
+def _point(name: str, kp: Dict[str, Tuple[float, float, float]]) -> np.ndarray:
+    x, y, _ = kp[name]
+    return np.array([x, y])
+
+def compute_angles(frames: List[FrameData], direction: str = 'left→right') -> None:
+    for fr in frames:
+        kp = fr.keypoints
+        # Core keypoints required for basic pose/angle calculations (standard COCO)
+        core_required = ['left_hip', 'right_hip', 'left_shoulder', 'right_shoulder', 
+                         'left_knee', 'right_knee', 'left_ankle', 'right_ankle']
+        if not all(r in kp for r in core_required):
+            continue
+            
+        # Optional: toe/heel (use ankle if missing)
+        # Note: New CSV uses 'left_big_toe' and 'right_big_toe'
+        l_toe_name = 'left_big_toe' if 'left_big_toe' in kp else ('left_toe' if 'left_toe' in kp else 'left_ankle')
+        r_toe_name = 'right_big_toe' if 'right_big_toe' in kp else ('right_toe' if 'right_toe' in kp else 'right_ankle')
+        
+        l_toe = _point(l_toe_name, kp)
+        r_toe = _point(r_toe_name, kp)
+        l_heel = _point('left_heel', kp) if 'left_heel' in kp else _point('left_ankle', kp)
+        r_heel = _point('right_heel', kp) if 'right_heel' in kp else _point('right_ankle', kp)
+            
+        # Optional nose for proxy head (C7)
+        nose = _point('nose', kp) if 'nose' in kp else None
+        left_sho = _point('left_shoulder', kp)
+        right_sho = _point('right_shoulder', kp)
+        c7 = (left_sho + right_sho) / 2  # Proxy for C7
+
+        # Compute pelvis midpoint, fallback if hips missing
+        if 'left_hip' in kp and 'right_hip' in kp:
+            pelvis = (_point('left_hip', kp) + _point('right_hip', kp)) / 2
+        elif 'left_hip' in kp:
+            pelvis = _point('left_hip', kp)
+        elif 'right_hip' in kp:
+            pelvis = _point('right_hip', kp)
+        else:
+            pelvis = np.array([0.0, 0.0])  # fallback if both hips missing
+
+        fr.pelvis_x, fr.pelvis_y = pelvis
+
+        # Trunk angle (hip midpoint → C7 proxy vs vertical)
+        trunk_vec = c7 - pelvis
+        fr.trunk_angle = np.degrees(np.arctan2(trunk_vec[0], -trunk_vec[1]))
+
+        # Trailing side based on direction
+        trailing = 'right' if direction == 'left→right' else 'left'
+        hip = _point(f'{trailing}_hip', kp)
+        knee = _point(f'{trailing}_knee', kp)
+        ankle = _point(f'{trailing}_ankle', kp)
+        heel = r_heel if trailing == 'right' else l_heel
+        
+        # Inject proxies back into keypoints dict for downstream lookups
+        if f'{trailing}_toe' not in kp: kp[f'{trailing}_toe'] = (float(r_toe[0] if trailing=='right' else l_toe[0]), float(r_toe[1] if trailing=='right' else l_toe[1]), 0.5)
+        if f'{trailing}_heel' not in kp: kp[f'{trailing}_heel'] = (float(heel[0]), float(heel[1]), 0.5)
+        # Also handle the leading side proxies just in case
+        leading = 'left' if trailing == 'right' else 'right'
+        if f'{leading}_toe' not in kp: kp[f'{leading}_toe'] = (float(l_toe[0] if leading=='left' else r_toe[0]), float(l_toe[1] if leading=='left' else r_toe[1]), 0.5)
+        if f'{leading}_heel' not in kp: kp[f'{leading}_heel'] = (float(l_heel[0] if leading=='left' else r_heel[0]), float(l_heel[1] if leading=='left' else r_heel[1]), 0.5)
+        
+        # Proxy for calf (midpoint of knee and ankle)
+        calf = (knee + ankle) / 2
+        
+        # Hip angle (hip‑pelvis‑C7)
+        hip_vec = c7 - hip
+        fr.hip_angle = np.degrees(np.arctan2(hip_vec[0], -hip_vec[1]))
+
+        # Knee angle (hip‑knee‑ankle)
+        thigh = hip - knee
+        shank = ankle - knee
+        cos_angle = np.clip(np.dot(thigh, shank) / (np.linalg.norm(thigh) * np.linalg.norm(shank)), -1, 1)
+        fr.knee_angle = np.degrees(np.arccos(cos_angle))
+
+        # Ankle dorsiflexion/plantarflexion (shank vs vertical)
+        fr.ankle_angle = np.degrees(np.arctan2(ankle[0] - knee[0], -(ankle[1] - knee[1])))
+
+        # Shin angle (knee‑ankle vs vertical)
+        shin_vec = ankle - knee
+        fr.shin_angle = np.degrees(np.arctan2(shin_vec[0], -shin_vec[1]))
+
+        # Simple foot‑contact flag (heel/toe close to estimated ground)
+        # We'll use a slightly more robust contact detection
+        l_heel = _point('left_heel', kp)
+        r_heel = _point('right_heel', kp)
+        l_toe = _point('left_toe', kp)
+        r_toe = _point('right_toe', kp)
+        
+        # Ground estimation: use the lowest points in the current frame as a proxy
+        current_ground = max(l_toe[1], r_toe[1], l_heel[1], r_heel[1], 
+                             _point('left_ankle', kp)[1], _point('right_ankle', kp)[1])
+        
+        # Trailing side heel for contact flag
+        heel_y = l_heel[1] if trailing == 'left' else r_heel[1]
+        
+        fr.left_contact = l_toe[1] > current_ground - VERTICAL_THRESHOLD_PX or l_heel[1] > current_ground - VERTICAL_THRESHOLD_PX
+        fr.right_contact = r_toe[1] > current_ground - VERTICAL_THRESHOLD_PX or r_heel[1] > current_ground - VERTICAL_THRESHOLD_PX
+        fr.foot_contact = fr.left_contact or fr.right_contact
+
+# -----------------------------------------------------------------------------
+# 3. Gait‑event detection
+# -----------------------------------------------------------------------------
+def detect_events(frames: List[FrameData], direction: str = 'left→right') -> Dict[str, List[int]]:
+    events: Dict[str, List[int]] = {
+        'contralateral_toe_off': [],
+        'initial_contact_left': [],
+        'initial_contact_right': [],
+        'midstance_left': [],
+        'midstance_right': [],
+        'MVP': [],
+        'max_knee_ext': [],
+        'MVP_to_late_swing': []
+    }
+
+    # Build contact series per foot
+    left_contacts = [fr.left_contact for fr in frames]
+    right_contacts = [fr.right_contact for fr in frames]
+
+    for i in range(1, len(frames) - 1):
+        # Only detect events on frames that have derived quantities
+        if frames[i].pelvis_x is None or frames[i].knee_angle is None:
+            continue
+            
+        # Contralateral toe‑off (trailing foot leaves ground)
+        if direction == 'left→right':
+            # Trailing is Right, so we look for Right toe-off
+            if right_contacts[i - 1] and not right_contacts[i]:
+                events['contralateral_toe_off'].append(frames[i - 1].idx)
+        else:
+            # Trailing is Left
+            if left_contacts[i - 1] and not left_contacts[i]:
+                events['contralateral_toe_off'].append(frames[i - 1].idx)
+
+        # Initial contact per foot
+        if not left_contacts[i - 1] and left_contacts[i]:
+            events['initial_contact_left'].append(frames[i].idx)
+        if not right_contacts[i - 1] and right_contacts[i]:
+            events['initial_contact_right'].append(frames[i].idx)
+
+    # Mid‑stance (pelvis over supporting ankle)
+    for fr in frames:
+        if fr.pelvis_x is None:
+            continue
+
+        if direction == 'left→right':
+            if 'left_ankle' not in fr.keypoints:
+                continue
+            ankle_x = fr.keypoints['left_ankle'][0]
+        else:
+            if 'right_ankle' not in fr.keypoints:
+                continue
+            ankle_x = fr.keypoints['right_ankle'][0]
+
+        if ankle_x is None:
+            continue
+
+        if abs(fr.pelvis_x - ankle_x) < PELVIS_ANKLE_DIST_PX:
+            key = 'midstance_left' if direction == 'left→right' else 'midstance_right'
+            events[key].append(fr.idx)
+
+    # MVP and Max Knee Extension (detecting local ones between toe-off and IC)
+    # We find intervals of "Swing" (No contact on trailing foot)
+    trailing_contacts = right_contacts if direction == 'left→right' else left_contacts
+    
+    swing_intervals = []
+    start_idx = None
+    for i, contact in enumerate(trailing_contacts):
+        if not contact and start_idx is None:
+            start_idx = i
+        elif contact and start_idx is not None:
+            if i - start_idx > 3: # Ignore very short glitches
+                swing_intervals.append((start_idx, i))
+            start_idx = None
+    if start_idx is not None:
+        swing_intervals.append((start_idx, len(frames)-1))
+
+    for start, end in swing_intervals:
+        window = frames[start:end+1]
+        
+        # 1. MVP (Highest pelvis height = MINIMUM Y in screen coords)
+        p_ys = [fr.pelvis_y for fr in window if fr.pelvis_y is not None]
+        if p_ys:
+            min_y = min(p_ys)
+            for fr in window:
+                if fr.pelvis_y == min_y:
+                    events['MVP'].append(fr.idx)
+                    break
+        
+        # 2. Max Knee Extension (Minimum knee angle)
+        k_angs = [fr.knee_angle for fr in window if fr.knee_angle is not None]
+        if k_angs:
+            min_k = min(k_angs)
+            for fr in window:
+                if fr.knee_angle == min_k:
+                    events['max_knee_ext'].append(fr.idx)
+                    break
+                    
+    # Ensure MVP_to_late_swing is populated for each segment (handled in score_stride windowing later)
+    # But for backward compatibility we store the FIRST one found if any
+    if events['MVP'] and events['max_knee_ext']:
+        s = events['MVP'][0]
+        e = events['max_knee_ext'][0]
+        if s > e: s, e = e, s
+        events['MVP_to_late_swing'] = list(range(s, e + 1))
+
+    return events
+
+# -----------------------------------------------------------------------------
+# 4. Signal Processing & Stride Segmentation
+# -----------------------------------------------------------------------------
+def smooth_signal(data: np.ndarray, window: int = 5) -> np.ndarray:
+    """Simple moving average smoothing."""
+    if len(data) < window: return data
+    return np.convolve(data, np.ones(window)/window, mode='same')
+
+def segment_strides(frames: List[FrameData], direction: str = 'left→right') -> List[Dict[str, int]]:
+    """Segment strides based on pelvis vertical oscillation (COM signal)."""
+    y_vals = np.array([f.pelvis_y if f.pelvis_y is not None else 0 for f in frames])
+    if len(y_vals) < 10:
+        return []
+
+    # Smooth the signal to remove jitter
+    y_smooth = smooth_signal(y_vals, window=7)
+    
+    # Locate "Troughs" (Local Maxima in screen-Y = Lowest position in physical space = IC area)
+    # We find peaks in the Y signal (max-Y)
+    strides = []
+    
+    # Simple peak detection (find local maxima of screen-Y)
+    # A peak must be higher than its neighbors by some prominence
+    peaks = []
+    prominence = 5.0 # px
+    for i in range(5, len(y_smooth) - 5):
+        val = y_smooth[i]
+        # Local max check
+        if val == max(y_smooth[i-5:i+6]):
+            # Prominence check (simple version: compare to local window min)
+            local_min = min(y_smooth[max(0, i-15):min(len(y_smooth), i+16)])
+            if val - local_min > prominence:
+                peaks.append(i)
+                
+    # One stride = interval between consecutive peaks/troughs
+    # We'll use these peaks as "Stride Boundaries" (approximating Initial Contact)
+    for j in range(len(peaks) - 1):
+        start_idx = peaks[j]
+        end_idx = peaks[j+1]
+        
+        # We need to find MVP and Max Knee Ext WITHIN this window
+        window_frames = frames[start_idx:end_idx+1]
+        
+        # MVP is the Peak (Minimum Y in screen space)
+        win_ys = [fr.pelvis_y for fr in window_frames if fr.pelvis_y is not None]
+        if not win_ys: continue
+        mvp_y = min(win_ys)
+        mvp_idx = next(fr.idx for fr in window_frames if fr.pelvis_y == mvp_y)
+        
+        # Max Knee Extension (Minimum knee angle during swing)
+        # Swing usually happens in the second half of this peak-to-peak interval
+        k_angs = [fr.knee_angle for fr in window_frames if fr.knee_angle is not None]
+        if not k_angs: continue
+        mke_angle = min(k_angs)
+        mke_idx = next(fr.idx for fr in window_frames if fr.knee_angle == mke_angle)
+        
+        # Toe-off is usually slightly before MVP
+        # For simplicity, we define the "stride" as the full window.
+        strides.append({
+            'contralateral_toe_off': frames[start_idx].idx, # Proxy: start of recovery
+            'MVP': mvp_idx,
+            'max_knee_ext': mke_idx,
+            'initial_contact': frames[end_idx].idx
+        })
+        
+    print(f"DEBUG: Signal-based segmenter found {len(strides)} strides.")
+    return strides
+
+# -----------------------------------------------------------------------------
+# 5. Scoring functions (each returns 0 or 1)
+# -----------------------------------------------------------------------------
+def trailing_limb_extension(trailing_hip_angle: float, trailing_knee_angle: float) -> int:
+    return int(trailing_hip_angle >= 45 and trailing_knee_angle <= 5)
+
+def back_kick(heel_y: float, calf_y: float, shin_angle: float) -> int:
+    heel_above_calf = heel_y < calf_y  # smaller y = higher in image coordinates
+    shin_not_parallel = shin_angle > 0
+    return int(heel_above_calf and shin_not_parallel)
+
+def trunk_rotation(event_frames: List[int], frames: List[FrameData]) -> int:
+    # Simple proxy: large arm swing crossing the midline
+    crossed = False
+    for idx in event_frames:
+        fr = next(f for f in frames if f.idx == idx)
+        left_shoulder_x = fr.keypoints['left_shoulder'][0]
+        right_shoulder_x = fr.keypoints['right_shoulder'][0]
+        if left_shoulder_x > right_shoulder_x:
+            crossed = True
+            break
+    return int(crossed)
+
+def thigh_separation(trailing_knee_x: float, gluteal_line_x: float) -> int:
+    return int(trailing_knee_x < gluteal_line_x)
+
+def lumbar_extension(fr: FrameData) -> int:
+    # proxy: trunk angle more negative than -10° (extension arch)
+    return int(fr.trunk_angle < -10)
+
+def forward_lean(fr: FrameData) -> int:
+    return int(abs(fr.trunk_angle) > 15)
+
+def foot_contact_com_distance(fr: FrameData, foot_x: float) -> int:
+    # distance between pelvis (CoM) and foot contact x
+    return int(abs(fr.pelvis_x - foot_x) > 0.5 * (abs(fr.keypoints['left_ankle'][0] - fr.keypoints['right_ankle'][0])))
+
+def shin_angle_criterion(fr: FrameData) -> int:
+    # ankle joint center in front of knee → shin angle < 0°
+    return int(fr.shin_angle < 0)
+
+def foot_inclination(fr: FrameData) -> int:
+    # gap between fore‑foot or heel and ground > threshold
+    ground_y = max(fr.keypoints['left_toe'][1], fr.keypoints['right_toe'][1])
+    fore_gap = fr.keypoints['left_toe'][1] - ground_y
+    heel_gap = fr.keypoints['left_heel'][1] - ground_y
+    return int(fore_gap > VERTICAL_THRESHOLD_PX or heel_gap > VERTICAL_THRESHOLD_PX)
+
+# -----------------------------------------------------------------------------
+# 6. Scoring a single stride
+# -----------------------------------------------------------------------------
+def score_stride(frames: List[FrameData], stride_events: Dict[str, int], direction: str = 'left→right') -> Tuple[Dict[str, int], int]:
+    # Helper to fetch a frame by idx
+    def get_frame(idx: int) -> FrameData:
+        return next((f for f in frames if f.idx == idx), frames[0])
+
+    # Map event names to frames
+    # Rename 'initial_contact' back to specific name for compatibility with existing scoring logic
+    ic_key = 'initial_contact_left' if direction == 'left→right' else 'initial_contact_right'
+    ev = {
+        'contralateral_toe_off': get_frame(stride_events['contralateral_toe_off']),
+        'MVP': get_frame(stride_events['MVP']),
+        'max_knee_ext': get_frame(stride_events['max_knee_ext']),
+        ic_key: get_frame(stride_events['initial_contact']),
+        'MVP_to_late_swing': list(range(stride_events['MVP'], stride_events['max_knee_ext'] + 1))
+    }
+
+    REQUIRED_EVENTS = [
+        'contralateral_toe_off',
+        'MVP',
+        'max_knee_ext'
+    ]
+
+    for e in REQUIRED_EVENTS:
+        if e not in ev:
+            raise RuntimeError(f"Missing required gait event: {e}")
+
+    # Trailing side based on direction
+    trailing = 'right' if direction == 'left→right' else 'left'
+    leading = 'left' if trailing == 'right' else 'right'
+
+    # Build criterion dictionary
+    scores = {
+        'trailing_limb_extension': trailing_limb_extension(
+            ev['contralateral_toe_off'].hip_angle,
+            ev['contralateral_toe_off'].knee_angle),
+        'back_kick': back_kick(
+            ev['contralateral_toe_off'].keypoints[f'{trailing}_heel'][1],
+            # Use proxy calf (midpoint of knee and ankle)
+            (ev['contralateral_toe_off'].keypoints[f'{trailing}_knee'][1] + ev['contralateral_toe_off'].keypoints[f'{trailing}_ankle'][1]) / 2,
+            ev['contralateral_toe_off'].shin_angle),
+        'trunk_rotation': trunk_rotation(ev['MVP_to_late_swing'], frames),
+        'thigh_separation_swing': thigh_separation(
+            ev['max_knee_ext'].keypoints[f'{trailing}_knee'][0],
+            # Use hip as gluteal line proxy
+            ev['max_knee_ext'].keypoints[f'{trailing}_hip'][0]),
+        'lumbar_extension_MVP': lumbar_extension(ev['MVP']),
+        'forward_lean_IC': forward_lean(ev['initial_contact_left'] if direction == 'left→right' else ev['initial_contact_right']),
+        'lumbar_extension_IC': lumbar_extension(ev['initial_contact_left'] if direction == 'left→right' else ev['initial_contact_right']),
+        'thigh_separation_IC': thigh_separation(
+            ev['initial_contact_left'].keypoints[f'{trailing}_knee'][0] if direction == 'left→right' else ev['initial_contact_right'].keypoints[f'{trailing}_knee'][0],
+            ev['initial_contact_left'].keypoints[f'{trailing}_hip'][0] if direction == 'left→right' else ev['initial_contact_right'].keypoints[f'{trailing}_hip'][0]),
+        'foot_contact_com_distance': foot_contact_com_distance(
+            ev['initial_contact_left'] if direction == 'left→right' else ev['initial_contact_right'],
+            ev['initial_contact_left'].keypoints[f'{trailing}_ankle'][0] if direction == 'left→right' else ev['initial_contact_right'].keypoints[f'{trailing}_ankle'][0]),
+        'shin_angle_criterion': shin_angle_criterion(ev['max_knee_ext']),
+        'foot_inclination': foot_inclination(ev['initial_contact_left'] if direction == 'left→right' else ev['initial_contact_right'])
+    }
+
+    total = sum(scores.values())
+    return scores, total
+
+# -----------------------------------------------------------------------------
+# 7. Driver function
+# -----------------------------------------------------------------------------
+def run_smas_pipeline(csv_path: str, direction: str = None) -> SessionResult:
+    print(f"Running S-MAS pipeline v{VERSION} on {csv_path}...")
+    frames = load_keypoints(csv_path)
+    
+    # First compute basics so we have pelvis_x for direction detection
+    # We do a partial compute_angles just for pelvis if needed, but easier to just use nose if pelvis not there yet.
+    # Actually, let's call a preliminary compute_angles.
+    compute_angles(frames, 'left→right') # Direction doesn't impact pelvis_x calculation
+    
+    # Auto-detect direction if not provided
+    if direction is None:
+        direction = detect_global_direction(frames)
+    
+    # Full angle computation with correct direction
+    compute_angles(frames, direction)
+    
+    # Segment events into individual strides using robust signal processing
+    stride_segments = segment_strides(frames, direction)
+    
+    # Identify localized events for each stride
+    events = detect_events(frames, direction)
+    
+    # Fallback if no full strides detected
+    if not stride_segments:
+        print("WARNING: No complete strides detected. Falling back to global analysis.")
+        # Create a single global segment using best-effort events
+        global_seg = {
+            'contralateral_toe_off': events['contralateral_toe_off'][0] if events['contralateral_toe_off'] else frames[0].idx,
+            'MVP': events['MVP'][0] if events['MVP'] else frames[len(frames)//4].idx,
+            'max_knee_ext': events['max_knee_ext'][0] if events['max_knee_ext'] else frames[len(frames)//2].idx,
+            'initial_contact': (events['initial_contact_left'] if direction == 'left→right' else events['initial_contact_right'])[0] if (events['initial_contact_left'] if direction == 'left→right' else events['initial_contact_right']) else frames[-1].idx
+        }
+        stride_segments = [global_seg]
+
+    strides_results = []
+    for i, seg in enumerate(stride_segments):
+        crit, tot = score_stride(frames, seg, direction)
+        strides_results.append(StrideResult(stride_id=i+1, criterion_scores=crit, total_smas_score=tot))
+
+    # Aggregation
+    scores = [s.total_smas_score for s in strides_results]
+    mean_score = np.mean(scores)
+    
+    # Per-criterion frequency calculation
+    criterion_counts = {}
+    for s in strides_results:
+        for k, v in s.criterion_scores.items():
+            criterion_counts[k] = criterion_counts.get(k, 0) + v
+    
+    per_criterion_freq = {k: v / len(strides_results) for k, v in criterion_counts.items()}
+    
+    sess = SessionResult(
+        mean_score=float(mean_score),
+        per_criterion_freq=per_criterion_freq,
+        stride_variance=float(np.var(scores)) if len(scores) > 1 else 0.0,
+        strides=strides_results
+    )
+    print(f"Pipeline complete. Analyzed {len(strides_results)} strides. Mean S-MAS: {mean_score:.2f}")
+    return sess
+
+# -----------------------------------------------------------------------------
+# 7. CLI entry point (optional)
+# -----------------------------------------------------------------------------
+if __name__ == "__main__":
+    import argparse, sys
+    parser = argparse.ArgumentParser(description="Run S‑MAS pipeline on YOLO key‑point CSV.")
+    parser.add_argument('csv_path', help='Path to YOLO key‑point CSV file')
+    parser.add_argument('--direction', default='left→right', help='Running direction (left→right or right→left)')
+    args = parser.parse_args()
+    result = run_smas_pipeline(args.csv_path, args.direction)
+    json.dump(result.__dict__, sys.stdout, indent=2, default=lambda o: o.__dict__)
+    print()

example/train_yolo_pose.py

@@ -0,0 +1,27 @@
+import argparse
+from ultralytics import YOLO
+
+def train_model(data_path, epochs=100, imgsz=640, batch=16, model_type='yolo11n-pose.pt', name='sprint_pose'):
+    """Train a YOLO pose model."""
+    model = YOLO(model_type)
+    
+    results = model.train(
+        data=data_path,
+        epochs=epochs,
+        imgsz=imgsz,
+        batch=batch,
+        name=name
+    )
+    print(f"✅ Training complete. Results saved in runs/pose/{name}")
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--data", required=True, help="Path to data.yaml")
+    parser.add_argument("--epochs", type=int, default=100)
+    parser.add_argument("--imgsz", type=int, default=640)
+    parser.add_argument("--batch", type=int, default=16)
+    parser.add_argument("--model", default='yolo11n-pose.pt')
+    parser.add_argument("--name", default='sprint_pose')
+    args = parser.parse_args()
+    
+    train_model(args.data, args.epochs, args.imgsz, args.batch, args.model, args.name)

example/video_utils.py

@@ -0,0 +1,48 @@
+import cv2
+import os
+from pathlib import Path
+
+def extract_frames(video_path, output_dir):
+    """Extract all frames from video."""
+    cap = cv2.VideoCapture(video_path)
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+
+    print(f"Video info: {width}x{height}, {fps:.2f} FPS, {total_frames} frames")
+
+    os.makedirs(output_dir, exist_ok=True)
+    frame_idx = 0
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        frame_filename = os.path.join(output_dir, f"frame_{frame_idx:06d}.jpg")
+        cv2.imwrite(frame_filename, frame)
+        frame_idx += 1
+
+        if frame_idx % 100 == 0:
+            print(f"Extracted {frame_idx}/{total_frames} frames")
+
+    cap.release()
+    print(f"✅ Extracted {frame_idx} frames to {output_dir}")
+    return fps, width, height, frame_idx
+
+def create_video_from_frames(frames_dir, output_path, fps, width, height):
+    """Compile frames into an MP4 video."""
+    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
+    
+    frame_files = sorted([f for f in os.listdir(frames_dir) if f.endswith('.jpg')])
+    print(f"Compiling {len(frame_files)} frames into {output_path}...")
+    
+    for i, frame_file in enumerate(frame_files):
+        frame = cv2.imread(os.path.join(frames_dir, frame_file))
+        out.write(frame)
+        if (i + 1) % 100 == 0:
+            print(f"Written {i+1}/{len(frame_files)} frames")
+            
+    out.release()
+    print(f"✅ Video saved to {output_path}")