feat: annotated video writer wraps OpenCV VideoWriter

src/video_writer.py

@@ -0,0 +1,53 @@
+"""Annotated video writer — wraps OpenCV VideoWriter for output."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import cv2
+import numpy as np
+
+
+class AnnotatedVideoWriter:
+    """Writes annotated frames to an MP4 file.
+
+    Args:
+        output_path: Destination .mp4 file path.
+        fps: Frames per second for the output video.
+        frame_size: (width, height) tuple.
+        codec: FourCC codec string (default "mp4v").
+    """
+
+    def __init__(
+        self,
+        output_path: str,
+        fps: float,
+        frame_size: tuple[int, int],
+        codec: str = "mp4v",
+    ) -> None:
+        if fps <= 0:
+            raise ValueError(f"fps must be > 0, got {fps}")
+        self._path = Path(output_path)
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+
+        w, h = frame_size
+        fourcc = cv2.VideoWriter_fourcc(*codec)
+        self._writer = cv2.VideoWriter(str(self._path), fourcc, fps, (w, h))
+        self._frame_count = 0
+
+        if not self._writer.isOpened():
+            raise RuntimeError(f"Failed to open VideoWriter for {self._path}")
+
+    def write_frame(self, frame: np.ndarray) -> None:
+        """Write one frame. Frame size must match constructor frame_size."""
+        self._writer.write(frame)
+        self._frame_count += 1
+
+    def finish(self) -> None:
+        """Release the writer. Idempotent — safe to call multiple times."""
+        if self._writer is not None and self._writer.isOpened():
+            self._writer.release()
+
+    @property
+    def frame_count(self) -> int:
+        return self._frame_count

tests/test_video_writer.py

@@ -0,0 +1,39 @@
+"""Tests for AnnotatedVideoWriter (src/video_writer.py)."""
+
+import cv2
+import numpy as np
+import pytest
+from src.video_writer import AnnotatedVideoWriter
+
+
+def test_writer_creates_output_file(tmp_path):
+    out = tmp_path / "test_output.mp4"
+    writer = AnnotatedVideoWriter(str(out), fps=25.0, frame_size=(640, 480))
+    frame = np.zeros((480, 640, 3), dtype=np.uint8)
+    writer.write_frame(frame)
+    writer.finish()
+    assert out.exists()
+    assert out.stat().st_size > 0
+
+
+def test_writer_multiple_frames(tmp_path):
+    out = tmp_path / "multi.mp4"
+    writer = AnnotatedVideoWriter(str(out), fps=25.0, frame_size=(320, 240))
+    for _ in range(10):
+        writer.write_frame(np.zeros((240, 320, 3), dtype=np.uint8))
+    writer.finish()
+    assert out.exists()
+
+
+def test_writer_close_idempotent(tmp_path):
+    out = tmp_path / "idem.mp4"
+    writer = AnnotatedVideoWriter(str(out), fps=25.0, frame_size=(320, 240))
+    writer.write_frame(np.zeros((240, 320, 3), dtype=np.uint8))
+    writer.finish()
+    writer.finish()  # second call should not raise
+    assert out.exists()
+
+
+def test_writer_invalid_fps():
+    with pytest.raises(ValueError):
+        AnnotatedVideoWriter("/tmp/x.mp4", fps=0.0, frame_size=(640, 480))