- detect_interval=2 on both pipelines; stabilizer 10-frame hold bridges skipped frames (GPU inference cut ~half during active batch) - sup_det_interval 5 -> 10 (still <= stabilizer hold / synthetic max_age) - preview_every_n default 2 -> 5 - AnnotatedVideoWriter: codec auto-chain gstreamer_nvenc -> avc1 -> mp4v, max_height downscale (even dims, INTER_AREA); backend logged - output_max_height plumbed Job -> web checkbox (720p) -> CLI --output-height - README: 720p option, CLI flag - 153 tests pass (+13)
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
"""Tests for AnnotatedVideoWriter (src/video_writer.py)."""
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
from src.video_writer import AnnotatedVideoWriter
|
|
|
|
|
|
def probe(path):
|
|
cap = cv2.VideoCapture(str(path))
|
|
assert cap.isOpened(), f"cannot reopen {path}"
|
|
info = (
|
|
int(cap.get(cv2.CAP_PROP_FRAME_COUNT)),
|
|
int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
|
|
int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
|
|
)
|
|
cap.release()
|
|
return info
|
|
|
|
|
|
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))
|
|
|
|
|
|
def test_max_height_downscale(tmp_path):
|
|
out = tmp_path / "down.mp4"
|
|
writer = AnnotatedVideoWriter(
|
|
str(out), fps=25.0, frame_size=(640, 480), max_height=240
|
|
)
|
|
assert writer.written_size == (320, 240)
|
|
for _ in range(3):
|
|
writer.write_frame(np.zeros((480, 640, 3), dtype=np.uint8))
|
|
writer.finish()
|
|
assert out.exists() and out.stat().st_size > 0
|
|
count, w, h = probe(out)
|
|
assert count == 3
|
|
assert (w, h) == writer.written_size
|
|
|
|
|
|
def test_no_max_height_keeps_native_size(tmp_path):
|
|
out = tmp_path / "native.mp4"
|
|
writer = AnnotatedVideoWriter(str(out), fps=25.0, frame_size=(640, 480))
|
|
assert writer.written_size == (640, 480)
|
|
writer.write_frame(np.zeros((480, 640, 3), dtype=np.uint8))
|
|
writer.finish()
|
|
assert probe(out)[1:] == (640, 480)
|
|
|
|
|
|
def test_even_dimensions_after_downscale(tmp_path):
|
|
out = tmp_path / "even.mp4"
|
|
writer = AnnotatedVideoWriter(
|
|
str(out), fps=25.0, frame_size=(641, 481), max_height=240
|
|
)
|
|
tw, th = writer.written_size
|
|
assert tw % 2 == 0 and th % 2 == 0
|
|
assert th <= 240
|
|
writer.write_frame(np.zeros((481, 641, 3), dtype=np.uint8))
|
|
writer.finish()
|
|
assert probe(out)[1:] == writer.written_size
|
|
|
|
|
|
def test_auto_codec_records_backend(tmp_path):
|
|
out = tmp_path / "auto.mp4"
|
|
writer = AnnotatedVideoWriter(str(out), fps=25.0, frame_size=(640, 480))
|
|
assert isinstance(writer.backend, str)
|
|
assert writer.backend
|
|
writer.write_frame(np.zeros((480, 640, 3), dtype=np.uint8))
|
|
writer.finish()
|
|
|
|
|
|
def test_explicit_codec_forces_fourcc(tmp_path):
|
|
out = tmp_path / "explicit.mp4"
|
|
writer = AnnotatedVideoWriter(
|
|
str(out), fps=25.0, frame_size=(640, 480), codec="mp4v"
|
|
)
|
|
assert writer.backend == "mp4v"
|
|
writer.write_frame(np.zeros((480, 640, 3), dtype=np.uint8))
|
|
writer.finish()
|
|
assert out.exists() and out.stat().st_size > 0
|