diff --git a/src/docgen/av_sync.py b/src/docgen/av_sync.py index a402e6c..04d9463 100644 --- a/src/docgen/av_sync.py +++ b/src/docgen/av_sync.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math import re from dataclasses import dataclass, field from pathlib import Path @@ -35,6 +36,19 @@ def _ocr_keyword_from_label(label: str) -> str | None: return max(tokens, key=len) +def _require_positive_fps(cap: Any, raw: float) -> float: + """A zero or missing frame rate must not be treated as 30 fps.""" + try: + fps = float(raw) + except (TypeError, ValueError) as exc: + cap.release() + raise ValueError("video frame rate is not a number") from exc + if not math.isfinite(fps) or fps <= 0: + cap.release() + raise ValueError("video frame rate must be a positive finite number") + return fps + + class AVSyncValidator: def __init__(self, config: Config) -> None: self.config = config @@ -84,7 +98,7 @@ def validate_segment_with_timing( if not cap.isOpened(): report.passed = False return report - fps = cap.get(cv2.CAP_PROP_FPS) or 30 + fps = _require_positive_fps(cap, cap.get(cv2.CAP_PROP_FPS)) for anchor in anchors: # Sample frames within tolerance window diff --git a/tests/test_av_sync.py b/tests/test_av_sync.py new file mode 100644 index 0000000..32790fc --- /dev/null +++ b/tests/test_av_sync.py @@ -0,0 +1,62 @@ +"""AV sync must not invent a frame rate when the video does not report one.""" + +from __future__ import annotations + +from pathlib import Path + +import cv2 +import pytest +import yaml + +from docgen.av_sync import AVSyncValidator +from docgen.config import Config + + +def _cfg(tmp_path: Path) -> Config: + raw = {"segments": {"all": ["01"], "default": ["01"]}} + path = tmp_path / "docgen.yaml" + path.write_text(yaml.dump(raw), encoding="utf-8") + return Config.from_yaml(path) + + +class _Cap: + def __init__(self, *_args: object, **_kwargs: object) -> None: + self.fps = 0.0 + self.sampled = False + + def isOpened(self) -> bool: + return True + + def get(self, _prop: int) -> float: + return self.fps + + def set(self, *_args: object) -> None: + self.sampled = True + + def read(self) -> tuple[bool, None]: + self.sampled = True + return False, None + + def release(self) -> None: + return None + + +def test_zero_fps_is_refused(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + cap = _Cap() + monkeypatch.setattr(cv2, "VideoCapture", lambda *_a, **_k: cap) + words = {"words": [{"word": "pipeline", "start": 1.0}, {"word": "checkout", "start": 2.0}]} + with pytest.raises(ValueError, match="positive finite"): + AVSyncValidator(_cfg(tmp_path)).validate_segment_with_timing("01", tmp_path / "clip.mp4", words) + assert cap.sampled is False + + +def test_positive_fps_still_samples(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + cap = _Cap() + cap.fps = 30.0 + monkeypatch.setattr(cv2, "VideoCapture", lambda *_a, **_k: cap) + words = {"words": [{"word": "pipeline", "start": 1.0}, {"word": "checkout", "start": 2.0}]} + report = AVSyncValidator(_cfg(tmp_path)).validate_segment_with_timing( + "01", tmp_path / "clip.mp4", words + ) + assert cap.sampled is True + assert report.passed is False