Recover truncated video decodes via an ffmpeg fallback

CHANGELOG.md

@@ -60,3 +60,10 @@ All notable changes to this project are documented here. The format is based on
   stream's true extent rather than the container-header duration. Some files
   (notably certain AVIs) report a bogus multi-hour duration from a malformed
   header, which previously produced a grid of thousands of empty bins.
+- `lowlevel.extract_visual` no longer loses the tail of a clip when OpenCV stops
+  decoding partway through. Some variable-rate AVIs make OpenCV give up early and
+  report no error, so the frames it did return covered only the front of the clip
+  and every later bin came out NaN while the audio features (decoded separately by
+  ffmpeg) looked fine. The decode is now checked against the container's declared
+  frame count, and a short read is retried with ffmpeg, which reads these files
+  whole. A `RuntimeWarning` reports the recovery instead of leaving it silent.

src/affectlens/lowlevel.py

@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import subprocess
+import warnings
 import wave
 from pathlib import Path
 
@@ -29,6 +30,10 @@
 from . import midlevel
 from .config import ExtractionConfig
 
+# Fraction of a container's declared frames OpenCV must actually decode before we
+# trust the decode; below this, ``extract_visual`` retries with ffmpeg.
+_DECODE_SHORTFALL = 0.98
+
 
 # --------------------------------------------------------------------------- #
 # Visual
@@ -43,41 +48,69 @@ def _colorfulness(bgr: np.ndarray) -> float:
     return float(std + 0.3 * mean)
 
 
-def extract_visual(path: str | Path, config: ExtractionConfig | None = None) -> pd.DataFrame:
-    """Per-sampled-frame visual features.
+def _video_meta(path: str | Path) -> tuple[float, int]:
+    """Native frame rate and the container's declared frame count."""
+    cap = cv2.VideoCapture(str(path))
+    if not cap.isOpened():
+        raise RuntimeError(f"could not open video: {path}")
+    fps = cap.get(cv2.CAP_PROP_FPS) or 0.0
+    declared = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
+    cap.release()
+    return fps, max(0, declared)
 
-    Low-level columns: t, luminance, contrast, colorfulness, saturation,
-    edge_density, motion (mean absolute inter-frame difference of luminance).
-    Mid-level columns (see ``midlevel.py``): flow_magnitude, flow_looming, and
-    flow_coherence (dense optical flow; only when ``config.optical_flow``),
-    scene_cut (shot-boundary score), spatial_detail (high-spatial-frequency
-    energy), chroma_rg / chroma_by (signed cone-opponent colour axes), and
-    face_count / face_prominence (YuNet; only when ``config.detect_faces`` and the
-    bundled model is available).
-    """
-    config = config or ExtractionConfig()
+
+def _iter_frames_cv2(path: str | Path):
+    """Yield BGR frames via OpenCV, the fast path for well-formed containers."""
     cap = cv2.VideoCapture(str(path))
     if not cap.isOpened():
         raise RuntimeError(f"could not open video: {path}")
+    try:
+        while True:
+            ok, frame = cap.read()
+            if not ok:
+                break
+            yield frame
+    finally:
+        cap.release()
 
-    native_fps = cap.get(cv2.CAP_PROP_FPS) or 0.0
-    sample_fps = config.visual_sample_fps or native_fps or 8.0
-    frame_step = max(1, int(round(native_fps / sample_fps))) if native_fps else 1
 
-    # One face detector for the whole clip (None if disabled or model missing).
-    face_det = midlevel.load_face_detector() if config.detect_faces else None
+def _iter_frames_ffmpeg(path: str | Path):
+    """Yield BGR frames via the bundled ffmpeg.
+
+    The fallback decoder: ffmpeg reads containers whole that OpenCV abandons
+    partway (see ``extract_visual``), at the cost of a slower pipe.
+    """
+    reader = imageio_ffmpeg.read_frames(str(path), pix_fmt="bgr24")
+    try:
+        meta = next(reader)  # first yield is metadata, not a frame
+        width, height = meta["size"]
+        for buf in reader:
+            yield np.frombuffer(buf, dtype=np.uint8).reshape(height, width, 3)
+    finally:
+        reader.close()  # tear down the ffmpeg subprocess and its pipes
 
+
+def _visual_rows(
+    frames,
+    native_fps: float,
+    sample_fps: float,
+    frame_step: int,
+    config: ExtractionConfig,
+    face_det,
+) -> tuple[list[dict], int]:
+    """Compute per-sampled-frame features over a frame iterator.
+
+    Returns the feature rows and how many frames the decoder actually yielded,
+    which ``extract_visual`` uses to spot a decode that stopped early.
+    """
     rows: list[dict] = []
     prev_gray: np.ndarray | None = None
     prev_gray_u8: np.ndarray | None = None
     prev_small: np.ndarray | None = None
-    idx = 0
-    while True:
-        ok, frame = cap.read()
-        if not ok:
-            break
+    n_decoded = 0
+    for idx, frame in enumerate(frames):
+        n_decoded += 1
         if idx % frame_step != 0:
-            idx += 1
             continue
         t = idx / native_fps if native_fps else idx / sample_fps
         gray_u8 = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
@@ -130,8 +163,65 @@ def extract_visual(path: str | Path, config: ExtractionConfig | None = None) ->
             }
         )
         prev_gray, prev_gray_u8, prev_small = gray, gray_u8, small
-        idx += 1
-    cap.release()
+    return rows, n_decoded
+
+
+def extract_visual(path: str | Path, config: ExtractionConfig | None = None) -> pd.DataFrame:
+    """Per-sampled-frame visual features.
+
+    Low-level columns: t, luminance, contrast, colorfulness, saturation,
+    edge_density, motion (mean absolute inter-frame difference of luminance).
+    Mid-level columns (see ``midlevel.py``): flow_magnitude, flow_looming, and
+    flow_coherence (dense optical flow; only when ``config.optical_flow``),
+    scene_cut (shot-boundary score), spatial_detail (high-spatial-frequency
+    energy), chroma_rg / chroma_by (signed cone-opponent colour axes), and
+    face_count / face_prominence (YuNet; only when ``config.detect_faces`` and the
+    bundled model is available).
+    """
+    config = config or ExtractionConfig()
+    native_fps, declared_frames = _video_meta(path)
+    sample_fps = config.visual_sample_fps or native_fps or 8.0
+    frame_step = max(1, int(round(native_fps / sample_fps))) if native_fps else 1
+
+    # One face detector for the whole clip (None if disabled or model missing).
+    face_det = midlevel.load_face_detector() if config.detect_faces else None
+
+    rows, n_decoded = _visual_rows(
+        _iter_frames_cv2(path), native_fps, sample_fps, frame_step, config, face_det
+    )
+
+    # Some containers (notably variable-rate AVIs) make OpenCV give up partway
+    # through with no error at all, which leaves the tail of the clip with no
+    # frames and, once binned, a block of NaN features. ffmpeg reads those files
+    # whole, so retry with it and keep whichever decode got further.
+    if declared_frames and n_decoded < declared_frames * _DECODE_SHORTFALL:
+        try:
+            alt_rows, alt_decoded = _visual_rows(
+                _iter_frames_ffmpeg(path), native_fps, sample_fps, frame_step, config, face_det
+            )
+        except Exception as exc:  # ffmpeg unavailable or refused the file
+            warnings.warn(
+                f"{Path(path).name}: OpenCV decoded {n_decoded} of {declared_frames} frames "
+                f"and the ffmpeg fallback failed ({exc}); features cover only part of the clip.",
+                RuntimeWarning,
+                stacklevel=2,
+            )
+        else:
+            if alt_decoded > n_decoded:
+                warnings.warn(
+                    f"{Path(path).name}: OpenCV decoded only {n_decoded} of {declared_frames} "
+                    f"frames; re-decoded with ffmpeg and recovered {alt_decoded}.",
+                    RuntimeWarning,
+                    stacklevel=2,
+                )
+                rows, n_decoded = alt_rows, alt_decoded
+            else:
+                warnings.warn(
+                    f"{Path(path).name}: only {n_decoded} of {declared_frames} declared frames "
+                    "could be decoded by either backend; features may not span the whole clip.",
+                    RuntimeWarning,
+                    stacklevel=2,
+                )
 
     if not rows:
         raise RuntimeError(f"no frames decoded from {path}")

tests/test_pipeline.py

@@ -12,7 +12,7 @@
 # Make the src-layout package importable without an install.
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 
-from affectlens import align, encoding, highlevel, ratings
+from affectlens import align, encoding, highlevel, lowlevel, ratings
 from affectlens.config import ExtractionConfig
 from affectlens.ratings import RatingSchema
 from affectlens.synthetic import build_synthetic_dataset
@@ -258,6 +258,45 @@ def test_extract_all_without_ratings():
         assert any(c.startswith("audio__") for c in cf.X.columns)
 
 
+def test_visual_decode_falls_back_when_opencv_truncates():
+    """A clip OpenCV abandons partway must not silently lose its tail.
+
+    Some variable-rate AVIs make OpenCV stop decoding early and report no error,
+    which used to leave the rest of the clip with no frames and a block of NaN
+    features once binned. Here the OpenCV path is truncated to half its frames,
+    so ``extract_visual`` should notice that it decoded fewer than the container
+    declares, re-decode with ffmpeg, and cover the whole clip again.
+    """
+    import warnings
+
+    with build_synthetic_dataset(n_clips=1) as (clips_dir, _ratings_csv):
+        clip = sorted(Path(clips_dir).glob("*.mp4"))[0]
+        full = lowlevel.extract_visual(clip)
+
+        real_iter = lowlevel._iter_frames_cv2
+
+        def truncated(path):
+            frames = list(real_iter(path))
+            return iter(frames[: len(frames) // 2])
+
+        lowlevel._iter_frames_cv2 = truncated
+        try:
+            with warnings.catch_warnings(record=True) as caught:
+                warnings.simplefilter("always")
+                recovered = lowlevel.extract_visual(clip)
+        finally:
+            lowlevel._iter_frames_cv2 = real_iter
+
+    # The tail is back: same coverage as an untruncated decode ...
+    assert len(recovered) == len(full)
+    assert recovered["t"].max() == full["t"].max()
+    assert not recovered.isna().any().any()
+    # ... and the recovery was announced rather than silently papered over.
+    assert any(
+        issubclass(w.category, RuntimeWarning) and "ffmpeg" in str(w.message) for w in caught
+    )
+
+
 if __name__ == "__main__":
     import inspect
     import tempfile