feat(renderer): generate annotated pose videos

app.py

@@ -92,6 +92,8 @@ def analyze_video(
 - **Exercise:** {report["exercise"]["exercise"]} ({report["exercise"]["confidence"]:.0%} confidence)
 - **Variation:** {report["variation"]["detected_variation"]} ({report["variation"]["variation_confidence"]:.0%} confidence)
 - **Reps:** {len(report["reps"]["reps"])}
+- **Analysis mode:** {report["artifacts"].get("analysis_mode", "unknown")}
+- **Pose source:** {report["artifacts"].get("pose_source", "unknown")}
 - **Main finding:** {summary["main_findings"][0] if summary["main_findings"] else "No major issue detected"}
 - **Run ID:** `{report["run_id"]}`
 
@@ -162,7 +164,7 @@ with gr.Blocks(title="Pozify") as demo:
             run_button = gr.Button("Analyze", variant="primary")
 
         with gr.Column(scale=2):
-            annotated_video = gr.Video(label="Annotated video placeholder")
+            annotated_video = gr.Video(label="Annotated video")
             summary_md = gr.Markdown(label="Summary")
 
     with gr.Tab("Final Report JSON"):

src/pozify/pipeline.py

@@ -86,14 +86,14 @@ def run_pipeline(
     issues = issue_marker.run(classification, reps, analysis, variation)
     write_artifact("issue_markers.json", issues)
 
-    annotated_video_path = annotated_renderer.run(manifest, issues, run_dir)
-    artifact_index.append(
-        {
-            "name": "annotated_video_placeholder.json",
-            "path": str(run_dir / "annotated_video_placeholder.json"),
-            "contract": "renderer_placeholder",
-        }
+    annotated_video_path = annotated_renderer.run(manifest, cleaned_pose_sequence, reps, issues, run_dir)
+
+    pose_source = (
+        cleaned_pose_sequence.frames[0].pose_quality.get("source")
+        if cleaned_pose_sequence.frames
+        else "none"
     )
+    analysis_mode = "mock" if mock_mode else "real"
 
     summary = coach_summary.run(profile, classification, reps, analysis, variation, issues)
     write_artifact("coach_summary.json", summary)
@@ -115,6 +115,9 @@ def run_pipeline(
         "artifacts": {
             "run_dir": str(run_dir),
             "annotated_video_path": annotated_video_path,
+            "rep_debug_path": str(run_dir / "rep_debug.json"),
+            "analysis_mode": analysis_mode,
+            "pose_source": pose_source,
         },
     }
     write_artifact("final_report.json", final_report)

src/pozify/steps/annotated_renderer.py

@@ -1,19 +1,173 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import Any
 
-from pozify.contracts import IssueMarkers, VideoManifest
+import cv2
 
+from pozify.contracts import IssueMarkers, PoseSequence, Reps, VideoManifest
 
-def run(manifest: VideoManifest, issues: IssueMarkers, run_dir: Path) -> str | None:
-    placeholder = {
-        "status": "mock_renderer",
-        "message": "Annotated renderer is not implemented yet. Returning original video path.",
-        "issue_count": len(issues.issues),
-    }
-    (run_dir / "annotated_video_placeholder.json").write_text(
-        __import__("json").dumps(placeholder, ensure_ascii=False, indent=2),
-        encoding="utf-8",
+
+SKELETON_EDGES = [
+    ("left_shoulder", "right_shoulder"),
+    ("left_hip", "right_hip"),
+    ("left_shoulder", "left_elbow"),
+    ("left_elbow", "left_wrist"),
+    ("right_shoulder", "right_elbow"),
+    ("right_elbow", "right_wrist"),
+    ("left_shoulder", "left_hip"),
+    ("right_shoulder", "right_hip"),
+    ("left_hip", "left_knee"),
+    ("left_knee", "left_ankle"),
+    ("right_hip", "right_knee"),
+    ("right_knee", "right_ankle"),
+]
+
+
+def _frame_landmark_points(
+    frame_landmarks: dict[str, dict[str, float]],
+    width: int,
+    height: int,
+) -> dict[str, tuple[int, int]]:
+    points: dict[str, tuple[int, int]] = {}
+    for name, values in frame_landmarks.items():
+        x = values.get("x")
+        y = values.get("y")
+        if x is None or y is None:
+            continue
+        points[name] = (int(round(x * width)), int(round(y * height)))
+    return points
+
+
+def _draw_pose(frame: Any, points: dict[str, tuple[int, int]]) -> None:
+    for start_name, end_name in SKELETON_EDGES:
+        start = points.get(start_name)
+        end = points.get(end_name)
+        if start is None or end is None:
+            continue
+        cv2.line(frame, start, end, (90, 220, 90), 2)
+
+    for point in points.values():
+        cv2.circle(frame, point, 3, (255, 240, 40), -1)
+
+
+def _rep_boundaries(reps: Reps) -> dict[int, list[str]]:
+    boundaries: dict[int, list[str]] = {}
+    for rep in reps.reps:
+        boundaries.setdefault(rep.start_frame, []).append(f"rep {rep.rep_id} start")
+        boundaries.setdefault(rep.mid_frame, []).append(f"rep {rep.rep_id} mid")
+        boundaries.setdefault(rep.end_frame, []).append(f"rep {rep.rep_id} end")
+    return boundaries
+
+
+def _draw_overlays(
+    frame: Any,
+    frame_index: int,
+    rep_count: int,
+    issue_count: int,
+    boundary_labels: dict[int, list[str]],
+) -> None:
+    cv2.putText(
+        frame,
+        f"Reps detected: {rep_count}",
+        (16, 28),
+        cv2.FONT_HERSHEY_SIMPLEX,
+        0.8,
+        (255, 255, 255),
+        2,
+        cv2.LINE_AA,
+    )
+    cv2.putText(
+        frame,
+        f"Issues: {issue_count}",
+        (16, 58),
+        cv2.FONT_HERSHEY_SIMPLEX,
+        0.7,
+        (200, 230, 255),
+        2,
+        cv2.LINE_AA,
     )
-    return manifest.video_path
+    labels = boundary_labels.get(frame_index, [])
+    for offset, label in enumerate(labels):
+        cv2.putText(
+            frame,
+            label,
+            (16, 96 + offset * 28),
+            cv2.FONT_HERSHEY_SIMPLEX,
+            0.75,
+            (80, 180, 255),
+            2,
+            cv2.LINE_AA,
+        )
+
+
+def run(
+    manifest: VideoManifest,
+    pose_sequence: PoseSequence,
+    reps: Reps,
+    issues: IssueMarkers,
+    run_dir: Path,
+) -> str | None:
+    if not manifest.analysis_allowed or not manifest.video_path:
+        return manifest.video_path
+
+    capture = cv2.VideoCapture(manifest.video_path)
+    if not capture.isOpened():
+        capture.release()
+        return manifest.video_path
+
+    fps = manifest.fps if manifest.fps > 0 else 30.0
+    width = manifest.width or int(capture.get(cv2.CAP_PROP_FRAME_WIDTH) or 0)
+    height = manifest.height or int(capture.get(cv2.CAP_PROP_FRAME_HEIGHT) or 0)
+    if width <= 0 or height <= 0:
+        capture.release()
+        return manifest.video_path
+
+    output_path = run_dir / "annotated_video.mp4"
+    writer = cv2.VideoWriter(
+        str(output_path),
+        cv2.VideoWriter_fourcc(*"mp4v"),
+        fps,
+        (width, height),
+    )
+    if not writer.isOpened():
+        capture.release()
+        return manifest.video_path
+
+    pose_by_frame = {frame.frame_index: frame for frame in pose_sequence.frames}
+    ordered_pose_frames = sorted(pose_sequence.frames, key=lambda frame: frame.frame_index)
+    pose_cursor = 0
+    last_pose_frame = None
+    boundary_labels = _rep_boundaries(reps)
+
+    try:
+        frame_index = 0
+        while True:
+            ok, frame = capture.read()
+            if not ok or frame is None:
+                break
+
+            if pose_cursor < len(ordered_pose_frames):
+                while (
+                    pose_cursor + 1 < len(ordered_pose_frames)
+                    and ordered_pose_frames[pose_cursor + 1].frame_index <= frame_index
+                ):
+                    pose_cursor += 1
+                candidate = ordered_pose_frames[pose_cursor]
+                if candidate.frame_index <= frame_index:
+                    last_pose_frame = candidate
+
+            exact_pose = pose_by_frame.get(frame_index)
+            active_pose = exact_pose or last_pose_frame
+            if active_pose is not None and active_pose.landmarks:
+                points = _frame_landmark_points(active_pose.landmarks, width, height)
+                _draw_pose(frame, points)
+
+            _draw_overlays(frame, frame_index, len(reps.reps), len(issues.issues), boundary_labels)
+            writer.write(frame)
+            frame_index += 1
+    finally:
+        writer.release()
+        capture.release()
 
+    return str(output_path)

tests/test_annotated_renderer.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+
+import cv2
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from pozify.contracts import IssueMarkers, PoseFrame, PoseSequence, Rep, Reps, VideoManifest
+from pozify.steps import annotated_renderer
+
+
+def _frame(frame_index: int) -> PoseFrame:
+    landmarks = {
+        "left_shoulder": {"x": 0.35, "y": 0.3},
+        "right_shoulder": {"x": 0.65, "y": 0.3},
+        "left_elbow": {"x": 0.3, "y": 0.45},
+        "right_elbow": {"x": 0.7, "y": 0.45},
+        "left_wrist": {"x": 0.28, "y": 0.6},
+        "right_wrist": {"x": 0.72, "y": 0.6},
+        "left_hip": {"x": 0.42, "y": 0.55},
+        "right_hip": {"x": 0.58, "y": 0.55},
+        "left_knee": {"x": 0.42, "y": 0.75},
+        "right_knee": {"x": 0.58, "y": 0.75},
+        "left_ankle": {"x": 0.42, "y": 0.92},
+        "right_ankle": {"x": 0.58, "y": 0.92},
+    }
+    return PoseFrame(
+        frame_index=frame_index,
+        timestamp_sec=round(frame_index / 30.0, 3),
+        landmarks=landmarks,
+        world_landmarks={},
+        pose_quality={"source": "fake_pose", "mean_visibility": 0.95},
+    )
+
+
+class AnnotatedRendererTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp_dir = tempfile.TemporaryDirectory()
+
+    def tearDown(self) -> None:
+        self.temp_dir.cleanup()
+
+    def _write_video(self, frame_count: int = 6) -> Path:
+        path = Path(self.temp_dir.name) / "input.mp4"
+        writer = cv2.VideoWriter(str(path), cv2.VideoWriter_fourcc(*"mp4v"), 30.0, (320, 240))
+        self.assertTrue(writer.isOpened())
+        for frame_index in range(frame_count):
+            frame = np.full((240, 320, 3), 90 + frame_index, dtype=np.uint8)
+            writer.write(frame)
+        writer.release()
+        return path
+
+    def test_renderer_writes_annotated_video(self) -> None:
+        video_path = self._write_video()
+        manifest = VideoManifest(
+            video_path=str(video_path),
+            fps=30.0,
+            duration_sec=0.2,
+            total_frames=6,
+            sampled_frames=6,
+            width=320,
+            height=240,
+            codec="mp4v",
+            container="mp4",
+            brightness_mean=100.0,
+            blur_laplacian_var=100.0,
+            quality_warnings=[],
+            analysis_allowed=True,
+        )
+        pose_sequence = PoseSequence(
+            frames=[_frame(index) for index in range(6)],
+            normalized=True,
+            smoothing_method="exponential_smoothing",
+            pose_valid_ratio=1.0,
+        )
+        reps = Reps(
+            exercise="push_up",
+            reps=[Rep(1, 0, 2, 5, 0.0, 0.067, 0.167)],
+            partial_reps=[],
+        )
+
+        output_path = annotated_renderer.run(
+            manifest,
+            pose_sequence,
+            reps,
+            IssueMarkers(issues=[]),
+            Path(self.temp_dir.name),
+        )
+
+        self.assertIsNotNone(output_path)
+        self.assertTrue(Path(str(output_path)).exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pipeline_contracts.py

@@ -149,7 +149,6 @@ class PipelineContractTests(unittest.TestCase):
                 "rep_analysis.json",
                 "variation.json",
                 "issue_markers.json",
-                "annotated_video_placeholder.json",
                 "coach_summary.json",
                 "verification.json",
                 "final_report.json",