diff --git a/src/machinevisiontoolbox/Sources.py b/src/machinevisiontoolbox/Sources.py index efe1366a..0babdf03 100644 --- a/src/machinevisiontoolbox/Sources.py +++ b/src/machinevisiontoolbox/Sources.py @@ -448,6 +448,7 @@ class VideoFile(ImageSource): :param filename: Path to video file :type filename: str :param kwargs: options applied to image frames, see :func:`~machinevisiontoolbox.base.imageio.convert` + :raises ValueError: if the video cannot be opened or its first frame cannot be read The resulting object is an iterator over the frames of the video file. The iterator returns :class:`~machinevisiontoolbox.Image` objects where: @@ -502,12 +503,20 @@ def __init__(self, filename: str, **kwargs: Any) -> None: # get the number of frames in the video # not sure it's always correct cap = cv2.VideoCapture(self.filename) - ret, frame = cap.read() - self.nframes = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) - self.shape = frame.shape - self.fps = int(cap.get(cv2.CAP_PROP_FPS)) + try: + if not cap.isOpened(): + raise ValueError(f"Cannot open video file {self.filename!r}") + ret, frame = cap.read() + if not ret or frame is None: + raise ValueError( + f"Cannot read the first frame of video file {self.filename!r}" + ) + self.nframes = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) + self.shape = frame.shape + self.fps = int(cap.get(cv2.CAP_PROP_FPS)) + finally: + cap.release() self.args = kwargs - cap.release() self.cap = None self.i = 0 diff --git a/tests/test_video_file_errors.py b/tests/test_video_file_errors.py new file mode 100644 index 00000000..841fb346 --- /dev/null +++ b/tests/test_video_file_errors.py @@ -0,0 +1,76 @@ +"""Video-file initialisation failures and capture lifetime.""" + +from pathlib import Path +from unittest.mock import Mock, patch + +import cv2 +import numpy as np +import pytest + +import machinevisiontoolbox.Sources as sources +from machinevisiontoolbox import VideoFile + + +@pytest.mark.parametrize("contents", [b"", b"not a video"]) +def test_invalid_video_file(tmp_path: Path, contents: bytes) -> None: + filename = tmp_path / "invalid.avi" + filename.write_bytes(contents) + + with pytest.raises(ValueError, match="Cannot open video file") as error: + VideoFile(str(filename)) + + assert repr(str(filename)) in str(error.value) + + +@pytest.mark.parametrize( + "opened, readable, message", + [ + (False, False, "Cannot open video file"), + (True, False, "Cannot read the first frame"), + (True, True, "Cannot read the first frame"), + ], +) +def test_failed_video_probe_releases_capture( + tmp_path: Path, opened: bool, readable: bool, message: str +) -> None: + filename = tmp_path / "invalid.avi" + filename.touch() + capture = Mock(spec=cv2.VideoCapture) + capture.isOpened.return_value = opened + capture.read.return_value = (readable, None) + capture.get.return_value = 0 + + with patch.object(sources.cv2, "VideoCapture", return_value=capture): + with pytest.raises(ValueError, match=message) as error: + VideoFile(str(filename)) + + assert repr(str(filename)) in str(error.value) + capture.release.assert_called_once_with() + capture.get.assert_not_called() + if not opened: + capture.read.assert_not_called() + + +def test_valid_video_probe_preserves_iteration(tmp_path: Path) -> None: + filename = tmp_path / "valid.avi" + writer = cv2.VideoWriter( + str(filename), cv2.VideoWriter_fourcc(*"MJPG"), 10, (16, 16) + ) + if not writer.isOpened(): + writer.release() + pytest.skip("MJPG video writer unavailable") + try: + for intensity in (40, 200): + writer.write(np.full((16, 16, 3), intensity, dtype=np.uint8)) + finally: + writer.release() + + with VideoFile(str(filename)) as video: + assert video.shape == (16, 16, 3) + assert video.nframes == 2 + assert video.fps == 10 + frames = list(video) + + assert [frame.id for frame in frames] == [0, 1] + np.testing.assert_allclose(frames[0].array, 40, atol=2) + np.testing.assert_allclose(frames[1].array, 200, atol=2)