feat(pipeline): emit live annotated frames for preview

src/job.py

@@ -54,6 +54,7 @@ class Job:
     error: str | None = None
     created_at: float = field(default_factory=time.time)
     completed_at: float | None = None
+    latest_frame: bytes | None = None
 
 
 class JobQueue:
@@ -153,11 +154,16 @@ class JobQueue:
                     f"{model_cfg.stem}_annotated.mp4",
                 )
 
+                def _on_frame(jpeg_bytes: bytes) -> None:
+                    with self._lock:
+                        job.latest_frame = jpeg_bytes
+
                 result: PipelineResult = run_pipeline(
                     video_path=job.video_path,
                     model_config=model_cfg,
                     output_path=output_path,
                     class_filter=class_filter,
+                    frame_callback=_on_frame,
                 )
 
                 with self._lock:
@@ -178,6 +184,7 @@ class JobQueue:
                 if job.status != JobStatus.CANCELLED:
                     job.status = JobStatus.COMPLETED
                 job.progress = 1.0
+                job.latest_frame = None
 
         except Exception as e:
             with self._lock:

src/pipeline.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
 
 import cv2
@@ -59,7 +60,8 @@ def run_pipeline(
     sack_conf: float = 0.4,
     truck_conf: float = 0.5,
     truck_det_interval: int = 15,
-    progress_callback=None,
+    progress_callback: Callable[[int, int], None] | None = None,
+    frame_callback: Callable[[bytes], None] | None = None,
 ) -> PipelineResult:
     """Process a video file through the counting pipeline.
 
@@ -72,6 +74,7 @@ def run_pipeline(
         truck_conf: Truck detection confidence threshold (default: 0.5).
         truck_det_interval: Run truck detection every N frames.
         progress_callback: Optional fn(frame_idx, total_frames) called per frame.
+        frame_callback: Optional fn(jpeg_bytes) called every 10th frame.
 
     Returns:
         PipelineResult with counting summary.
@@ -203,6 +206,11 @@ def run_pipeline(
 
             writer.write_frame(viz)
 
+            if frame_callback and frame_idx % 10 == 0:
+                ok, jpeg = cv2.imencode('.jpg', viz)
+                if ok:
+                    frame_callback(jpeg.tobytes())
+
             if progress_callback:
                 progress_callback(frame_idx, total_frames)
 

tests/test_job.py

@@ -16,6 +16,29 @@ def test_job_initial_status():
     assert job.status == JobStatus.PENDING
 
 
+def test_job_has_latest_frame():
+    """Job dataclass has latest_frame field defaulting to None."""
+    job = Job(
+        job_id="test-1",
+        video_path="/tmp/test.mp4",
+        model_configs=[],
+        output_dir="/tmp/output",
+    )
+    assert job.latest_frame is None
+
+
+def test_job_latest_frame_settable():
+    """Job.latest_frame can be set to bytes."""
+    job = Job(
+        job_id="test-1",
+        video_path="/tmp/test.mp4",
+        model_configs=[],
+        output_dir="/tmp/output",
+    )
+    job.latest_frame = b"\xff\xd8\xff\xe0"
+    assert job.latest_frame == b"\xff\xd8\xff\xe0"
+
+
 def test_queue_add_job():
     """Adding a job returns the job with PENDING status."""
     q = JobQueue(output_dir="/tmp/output")

tests/test_pipeline.py

@@ -69,3 +69,12 @@ def test_run_pipeline_no_model_raises(tmp_path):
             model_config=ModelConfig(filename="test.pt", path="/nonexistent.pt", stem="test", known_classes=["sack"]),
             output_path=str(tmp_path / "out.mp4"),
         )
+
+
+def test_run_pipeline_accepts_frame_callback(tmp_path):
+    """run_pipeline signature accepts frame_callback parameter."""
+    import inspect
+    sig = inspect.signature(run_pipeline)
+    assert "frame_callback" in sig.parameters
+    param = sig.parameters["frame_callback"]
+    assert param.default is None