Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion src/docgen/av_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

from __future__ import annotations

import math
import re
from dataclasses import dataclass, field
from pathlib import Path
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
62 changes: 62 additions & 0 deletions tests/test_av_sync.py
Original file line number Diff line number Diff line change
@@ -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
Loading