Migrate read_video_properties from OpenCV to PyAV

src/caliscope/recording/synchronized_timestamps.py

@@ -210,7 +210,7 @@ def from_videos(cls, recording_dir: Path, cam_ids: Sequence[int]) -> Self:
             frame_count = props["frame_count"]
             fps = props["fps"]
 
-            if frame_count == 0:
+            if frame_count <= 0:
                 raise ValueError(f"Could not determine frame count for cam_{cam_id}")
 
             if fps <= 0:

src/caliscope/recording/video_utils.py

@@ -1,13 +1,14 @@
 """Video file utilities.
 
 Functions for reading video metadata without full frame decoding.
+Uses PyAV for container inspection -- no frames are decoded.
 """
 
 import logging
 from pathlib import Path
 from typing import TypedDict
 
-import cv2
+import av
 
 logger = logging.getLogger(__name__)
 
@@ -23,36 +24,71 @@ class VideoProperties(TypedDict):
 
 
 def read_video_properties(source_path: Path) -> VideoProperties:
-    """Read video metadata (fps, frame_count, dimensions).
+    """Read video metadata (fps, frame_count, dimensions) via PyAV.
 
-    Opens the video file briefly to extract metadata, then releases the handle.
+    Opens the video container briefly to inspect stream metadata,
+    then closes it. No frames are decoded.
 
-    Args:
-        source_path: Path to video file (.mp4, .avi, etc.)
-
-    Returns:
-        VideoProperties dict with keys: fps, frame_count, width, height, size
+    Falls back to duration-based frame count when the container
+    does not report stream.frames (same logic as FrameSource.frame_count).
 
     Raises:
-        ValueError: If video file cannot be opened
+        FileNotFoundError: If source_path does not exist.
+        ValueError: If the file cannot be opened as video, has no video
+            stream, or if fps/frame_count cannot be determined. All PyAV
+            exceptions are wrapped as ValueError to preserve the caller
+            contract.
     """
-    logger.info(f"Reading video properties from: {source_path}")
+    if not source_path.exists():
+        raise FileNotFoundError(f"Video file not found: {source_path}")
 
-    video = cv2.VideoCapture(str(source_path))
+    logger.info(f"Reading video properties from: {source_path}")
 
-    if not video.isOpened():
-        raise ValueError(f"Could not open the video file: {source_path}")
+    try:
+        container = av.open(str(source_path))
+    except Exception as e:
+        raise ValueError(
+            f"Could not open video file: {source_path}. The file may be corrupted or in an unsupported format."
+        ) from e
 
     try:
-        width = int(video.get(cv2.CAP_PROP_FRAME_WIDTH))
-        height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        if not container.streams.video:
+            raise ValueError(f"No video stream found in: {source_path}")
+
+        stream = container.streams.video[0]
+
+        # FPS from average_rate (rational number, e.g. 30000/1001)
+        if stream.average_rate is None or float(stream.average_rate) <= 0:
+            raise ValueError(
+                f"Could not determine frame rate for: {source_path}. "
+                f"The video file may be corrupted or in an unsupported format."
+            )
+        fps = float(stream.average_rate)
+
+        # Frame count: prefer stream metadata, fall back to duration * fps
+        # This matches FrameSource.frame_count (metadata estimate), not
+        # FrameSource.last_frame_index (keyframe-scan result).
+        frame_count = stream.frames
+        if frame_count == 0 and container.duration is not None:
+            duration_seconds = container.duration / 1_000_000
+            frame_count = int(duration_seconds * fps)
+
+        if frame_count <= 0:
+            raise ValueError(
+                f"Could not determine frame count for: {source_path}. "
+                f"stream.frames={stream.frames}, "
+                f"container.duration={container.duration}"
+            )
+
+        width = stream.width
+        height = stream.height
 
         return VideoProperties(
-            fps=video.get(cv2.CAP_PROP_FPS),
-            frame_count=int(video.get(cv2.CAP_PROP_FRAME_COUNT)),
+            fps=fps,
+            frame_count=frame_count,
             width=width,
             height=height,
             size=(width, height),
         )
     finally:
-        video.release()
+        container.close()

tests/test_video_utils.py

@@ -0,0 +1,58 @@
+"""Tests for video_utils.read_video_properties (PyAV backend)."""
+
+from pathlib import Path
+
+import pytest
+
+from caliscope.recording.video_utils import read_video_properties
+
+EXTRINSIC_DIR = Path(__file__).parent / "sessions" / "4_cam_recording" / "calibration" / "extrinsic"
+
+
+class TestReadVideoProperties:
+    """read_video_properties returns correct metadata and raises on bad input."""
+
+    def test_returns_correct_metadata_for_test_video(self):
+        """Reads a real test video and validates all fields are sensible."""
+        video_path = EXTRINSIC_DIR / "cam_0.mp4"
+        props = read_video_properties(video_path)
+
+        assert props["fps"] > 0
+        assert props["frame_count"] > 0
+        assert props["width"] > 0
+        assert props["height"] > 0
+        assert props["size"] == (props["width"], props["height"])
+
+    def test_file_not_found_raises(self, tmp_path: Path):
+        """FileNotFoundError for nonexistent path."""
+        with pytest.raises(FileNotFoundError, match="Video file not found"):
+            read_video_properties(tmp_path / "nonexistent.mp4")
+
+    def test_corrupt_file_raises_value_error(self, tmp_path: Path):
+        """ValueError (not av.error.*) for a file that isn't valid video."""
+        fake_video = tmp_path / "not_a_video.mp4"
+        fake_video.write_text("this is not a video file")
+
+        with pytest.raises(ValueError, match="Could not open video file"):
+            read_video_properties(fake_video)
+
+    def test_consistent_across_cameras(self):
+        """All 4 test cameras return plausible, consistent FPS."""
+        fps_values = []
+        for cam_id in range(4):
+            props = read_video_properties(EXTRINSIC_DIR / f"cam_{cam_id}.mp4")
+            fps_values.append(props["fps"])
+
+        # All cameras should have the same FPS (within rounding)
+        for fps in fps_values:
+            assert abs(fps - fps_values[0]) < 0.1
+
+
+if __name__ == "__main__":
+    debug_dir = Path(__file__).parent / "tmp"
+    debug_dir.mkdir(parents=True, exist_ok=True)
+
+    for cam_id in range(4):
+        path = EXTRINSIC_DIR / f"cam_{cam_id}.mp4"
+        props = read_video_properties(path)
+        print(f"cam_{cam_id}: {props}")