From 5d6dce527ff035221bcb308892e0ef52a12ea8e4 Mon Sep 17 00:00:00 2001 From: Keigh Rim Date: Mon, 5 Oct 2026 11:20:55 -0400 Subject: [PATCH 1/2] fixed vdh bug when trying to extract images from invalid video frames --- mmif/utils/video_document_helper.py | 255 ++++++++++++--- tests/test_utils.py | 461 ++++++++++++++++++++++++---- 2 files changed, 622 insertions(+), 94 deletions(-) diff --git a/mmif/utils/video_document_helper.py b/mmif/utils/video_document_helper.py index bb1f5b1a..e6360784 100644 --- a/mmif/utils/video_document_helper.py +++ b/mmif/utils/video_document_helper.py @@ -1,6 +1,8 @@ import contextvars import importlib +import shlex import sys +from contextlib import contextmanager from enum import Enum import math @@ -27,6 +29,70 @@ 'offset spans (see issue #379).' ) +_MEDIA_CHECK_HINT = ( + 'The file may be truncated or corrupted; container metadata does not ' + 'reveal this, so to verify the file run ' + '`ffmpeg -v error -i {path} -map 0:v:0 -f null -` and treat any stderr ' + 'output as a problem (ffmpeg exits with 0 either way).' +) + + +def _media_check_hint(path: str) -> str: + """ + Formats the "verify this file" hint for a resolved document path. + + :param path: resolved local path of the video document + :returns: hint text, with the path quoted for shell use + :rtype: str + """ + return _MEDIA_CHECK_HINT.format(path=shlex.quote(path)) + + +def _stream_rate(stream): + """ + Resolves the frame rate of a video stream, as an exact fraction. + + Prefers ffmpeg's guessed rate (``av_guess_frame_rate``), which falls back + to the stream's base rate (``r_frame_rate``) when the average rate + (``avg_frame_rate``) is unset. Matroska/webm commonly leave + ``average_rate`` as ``None`` (which would crash ``float()``) or 0, so it + cannot be used directly; both are falsy, so the chain yields the first + usable one. The result stays a ``Fraction`` for exact frame-count math. + + :param stream: PyAV video stream + :returns: frame rate, or a falsy value when the stream reports none + """ + return stream.guessed_rate or stream.average_rate or stream.base_rate + + +@contextmanager +def _ffmpeg_op(action: str, video_document: Document, path: str): + """ + Turns a failed FFmpeg operation into an actionable ``RuntimeError``. + + Any :py:class:`av.error.FFmpegError` (and the ``IndexError`` of a file + that opens but holds no video stream) is re-raised as a ``RuntimeError`` + naming the document, the media check hint, and the original message. The + original message is repeated in the new one on purpose, because + downstream error reporting (e.g. ``clams-python``'s error view) records + only the outermost exception, so a chained cause would be dropped. + + :param action: what was attempted, as a verb phrase (e.g. ``'open'``) + :param video_document: :py:class:`~mmif.serialize.annotation.Document` + holding the video document being read + :param path: resolved local path of the video document + :raises RuntimeError: if the wrapped operation fails + """ + av = _check_cv_dep('av') + try: + yield + except (av.error.FFmpegError, IndexError) as e: + raise RuntimeError( + f'Cannot {action} video {video_document.id} ({path}). ' + f'{_media_check_hint(path)} ' + f'(original error: {type(e).__name__}: {e})' + ) from e + def _check_cv_dep(dep): """Import a CV dependency, raising ImportError with a helpful message.""" @@ -88,21 +154,28 @@ def open_container(video_document: Document): :returns: open PyAV :py:class:`av.container.InputContainer` :rtype: av.container.InputContainer :raises ValueError: if ``video_document`` is missing or of the wrong type + :raises RuntimeError: if the file cannot be opened and read as a video, + for instance an mp4 whose ``moov`` atom is cut off, or a prefix too + short to hold a video stream """ av = _check_cv_dep('av') if video_document is None or video_document.at_type != DocumentTypes.VideoDocument: raise ValueError(f'The document does not exist.') - container = av.open(video_document.location_path(nonexist_ok=False)) - stream = container.streams.video[0] + path = video_document.location_path(nonexist_ok=False) + container = None + with _ffmpeg_op('open', video_document, path): + try: + container = av.open(path) + stream = container.streams.video[0] + except BaseException: + # a file can open and still hold no video stream, in which case + # the container must not be left dangling + if container is not None: + container.close() + raise time_base = float(stream.time_base) - # Frame rate: prefer ffmpeg's guessed rate (av_guess_frame_rate), which - # falls back to the stream's base rate (r_frame_rate) when the average - # rate (avg_frame_rate) is unset. Matroska/webm commonly leave - # `average_rate` as None (which would crash float()) or 0, so it cannot be - # used directly. `rate` stays a Fraction for exact frame-count math; both - # None and a zero rate are falsy, so the chain yields the first usable one. - rate = stream.guessed_rate or stream.average_rate or stream.base_rate + rate = _stream_rate(stream) fps = round(float(rate), 2) if rate else 0.0 # Resolve duration (in seconds) from the most reliable source available. # `stream.duration` is exact for CFR H.264/MP4, but Matroska/webm leave it @@ -121,13 +194,13 @@ def open_container(video_document: Document): if stream.frames > 0: frame_count = stream.frames elif duration_s is not None and rate: - frame_count = int(round(duration_s * float(rate))) + frame_count = round(duration_s * float(rate)) else: frame_count = 0 if duration_s is not None: - duration_ms = int(round(duration_s * 1000)) + duration_ms = round(duration_s * 1000) elif frame_count > 0 and fps > 0: - duration_ms = int(round(frame_count / fps * 1000)) + duration_ms = round(frame_count / fps * 1000) else: duration_ms = 0 video_document.add_property(FPS_DOCPROP_KEY, fps) @@ -161,6 +234,78 @@ def get_framerate(video_document: Document) -> float: container.close() +def _decode_frames(container, stream, decode_errors: List[str]): + """ + Iterates decoded frames of a video stream, stopping at the first error. + + Decoding stops at the first :py:class:`av.error.FFmpegError`, which a + truncated or corrupted file can raise at any point, but the decoder is + flushed first so frames that are held back for reordering are not lost. + The error text is appended to ``decode_errors`` for the caller to report. + + :param container: open PyAV :py:class:`av.container.InputContainer` + :param stream: video stream to decode + :param decode_errors: list that collects decoding error texts + :returns: generator of :py:class:`av.video.frame.VideoFrame` + """ + av = _check_cv_dep('av') + try: + for frame in container.decode(stream): + yield frame + except av.error.FFmpegError as e: + decode_errors.append(f'{type(e).__name__}: {e}') + try: + for frame in stream.codec_context.decode(None): + yield frame + except av.error.FFmpegError as flush_e: + decode_errors.append(f'{type(flush_e).__name__}: {flush_e}') + + +_MAX_LISTED_TIMEPOINTS = 5 + + +def _warn_missing_images( + video_document: Document, + path: str, + missing_ms: List[int], + n_missing_positions: int, + decode_errors: List[str], +) -> None: + """ + Issues a single warning for timepoints that have no image. + + A missing timepoint that falls before the document's ``duration``, or a + decoding error, means the file holds fewer images than its container + metadata promises, hence the media check hint. Timepoints past a known + duration are simply out of range, and get no hint. + + :param video_document: :py:class:`~mmif.serialize.annotation.Document` + holding a video document + :param path: resolved local path of the video document + :param missing_ms: sorted distinct timepoints (in ms) that have no image + :param n_missing_positions: number of positions in the returned list that + hold ``None``, which exceeds ``len(missing_ms)`` when timepoints repeat + :param decode_errors: decoding error texts collected while decoding + """ + if len(missing_ms) <= _MAX_LISTED_TIMEPOINTS: + where = ', '.join(f'{ms}ms' for ms in missing_ms) + else: + where = (f'{len(missing_ms)} distinct timepoints between ' + f'{missing_ms[0]}ms and {missing_ms[-1]}ms') + msg = (f'No image is available for {n_missing_positions} timepoint(s) ' + f'({where}) of video {video_document.id}; the returned list holds ' + f'`None` at those positions.') + duration_ms = video_document.get_property(DURATION_DOCPROP_KEY) + # a zero duration means the container reports none at all, so the + # out-of-range test cannot be applied and the file stays suspect + if decode_errors or not duration_ms or missing_ms[0] < duration_ms: + msg += f' {_media_check_hint(path)}' + if decode_errors: + msg += f' (first decoding error: {decode_errors[0]})' + # helper -> public extraction function -> the caller to be pointed at + warnings.warn(msg, stacklevel=3) + + def extract_images_from_timepoints( video_document: Document, timepoints_ms: Iterable[int], @@ -173,13 +318,20 @@ def extract_images_from_timepoints( presentation timestamp (PTS) is closest to it. Duplicate timepoints produce duplicate images at the same list positions as the input. + The returned list always has one entry per requested timepoint. An entry + is ``None`` when no image is available, namely when the timepoint is + more than one frame duration past the last decodable frame. This happens + for timepoints past the end of the video, and also within a truncated or + corrupted file, where decoding stops early. Such a call issues a single + warning naming the missing timepoints. + :param video_document: :py:class:`~mmif.serialize.annotation.Document` holding a video document (``"@type": ".../VideoDocument/..."``) :param timepoints_ms: iterable of timepoint values in milliseconds :param as_PIL: return :py:class:`PIL.Image.Image` (RGB) instead of :py:class:`~numpy.ndarray` (BGR) :returns: images in the same order (and with the same multiplicity) as - ``timepoints_ms`` + ``timepoints_ms``, with ``None`` where no image is available :rtype: list """ original_timepoints = list(timepoints_ms) @@ -190,17 +342,34 @@ def extract_images_from_timepoints( Image = _check_cv_dep('PIL.Image') if as_PIL else None container = open_container(video_document) + path = video_document.location_path() result_map = {} + decode_errors: List[str] = [] try: stream = container.streams.video[0] time_base = float(stream.time_base) + rate = _stream_rate(stream) + # Tolerance for a target that falls outside the decoded frames. A + # frame covers the interval up to the next one, so a target within + # that span still belongs to it; this is the normal case for the last + # frame of a healthy video, whose header duration usually runs a + # frame longer than the frame's own PTS. `frame.duration` is used when + # the container provides it, with the stream rate as the fallback. + default_tolerance_ticks = (round(1 / float(rate) / time_base) + if rate else 0) + + def _tolerance_ticks(frame): + return (getattr(frame, 'duration', None) + or default_tolerance_ticks) + # convert each target ms to stream ticks (PTS units) - target_ticks = [int(round(t_ms / 1000.0 / time_base)) + target_ticks = [round(t_ms / 1000.0 / time_base) for t_ms in unique_sorted_ms] # seek to the nearest keyframe at or before the earliest target - container.seek(target_ticks[0], backward=True, any_frame=False, - stream=stream) + with _ffmpeg_op('seek in', video_document, path): + container.seek(target_ticks[0], backward=True, any_frame=False, + stream=stream) targets = iter(zip(unique_sorted_ms, target_ticks)) cur_ms, cur_pts = next(targets, (None, None)) @@ -211,13 +380,19 @@ def _emit(frame, t_ms): result_map[t_ms] = (frame.to_image() if as_PIL else frame.to_ndarray(format='bgr24')) - for frame in container.decode(stream): + for frame in _decode_frames(container, stream, decode_errors): if frame.pts is None: continue pts = frame.pts while cur_ms is not None and pts >= cur_pts: - # pick whichever of (prev, current) is closer to target - if prev_pts is None or (pts - cur_pts) <= (cur_pts - prev_pts): + if prev_pts is None: + # nothing decoded before the target, which happens when + # the seek lands after it or the start of the stream is + # missing; the same distance limit applies as at the end + if pts - cur_pts <= _tolerance_ticks(frame): + _emit(frame, cur_ms) + # otherwise pick whichever of (prev, current) is closer + elif (pts - cur_pts) <= (cur_pts - prev_pts): _emit(frame, cur_ms) else: _emit(prev_frame, cur_ms) @@ -227,24 +402,24 @@ def _emit(frame, t_ms): if cur_ms is None: break - # targets past the last decoded frame: fall back to the last frame + # Targets past the last decoded frame belong to that frame only while + # they fall within its duration; anything further has no image, and is + # left out of `result_map` to become a `None` entry below. while cur_ms is not None: - if prev_frame is not None: - warnings.warn( - f'Timepoint {cur_ms}ms is beyond the video duration; ' - f'returning the last decoded frame for {video_document.id}.' - ) + if (prev_frame is not None + and cur_pts - prev_pts <= _tolerance_ticks(prev_frame)): _emit(prev_frame, cur_ms) - else: - warnings.warn( - f'No frames decoded for timepoint {cur_ms}ms from ' - f'video {video_document.id}.' - ) cur_ms, cur_pts = next(targets, (None, None)) finally: container.close() - return [result_map[t] for t in original_timepoints if t in result_map] + missing_ms = [t for t in unique_sorted_ms if t not in result_map] + if missing_ms: + n_missing_positions = sum(1 for t in original_timepoints + if t not in result_map) + _warn_missing_images(video_document, path, missing_ms, + n_missing_positions, decode_errors) + return [result_map.get(t) for t in original_timepoints] def _tp_ids_to_timepoints_ms(mmif: Mmif, tp_ids: List[str]) -> List[int]: @@ -407,7 +582,9 @@ def extract_images_by_count_with_sources( :param as_PIL: return :py:class:`~PIL.Image.Image` instead of :py:class:`~numpy.ndarray` :return: tuple of (list of images, list of selected target TP IDs); - the two lists are parallel + the two lists are parallel, and an image is ``None`` when none is + available for that timepoint (see + :py:func:`extract_images_from_timepoints`) :rtype: tuple """ if 'targets' not in annotation.properties: @@ -460,7 +637,8 @@ def extract_images_by_count( :param fraction: fraction of targets to include (ideally) :param as_PIL: return :py:class:`~PIL.Image.Image` instead of :py:class:`~numpy.ndarray` - :return: list of images + :return: list of images, with ``None`` where none is available (see + :py:func:`extract_images_from_timepoints`) :rtype: list """ images, _ = extract_images_by_count_with_sources( @@ -496,8 +674,10 @@ def extract_images_by_mode_with_sources( :param as_PIL: return :py:class:`PIL.Image.Image` instead of :py:class:`~numpy.ndarray` :return: tuple of (list of images, list of sources); the two lists - are parallel. May be ``([], [])`` for ``REPRESENTATIVES`` mode - when no representatives exist. + are parallel, and an image is ``None`` when none is available for + that timepoint (see :py:func:`extract_images_from_timepoints`). + May be ``([], [])`` for ``REPRESENTATIVES`` mode when no + representatives exist. :rtype: tuple """ if mode is None: @@ -540,8 +720,9 @@ def extract_images_by_mode( default :param as_PIL: return :py:class:`PIL.Image.Image` instead of :py:class:`~numpy.ndarray` - :return: list of images (may be empty for ``REPRESENTATIVES`` mode - when no representatives exist) + :return: list of images, with ``None`` where none is available (see + :py:func:`extract_images_from_timepoints`); may be empty for + ``REPRESENTATIVES`` mode when no representatives exist :rtype: list """ images, _ = extract_images_by_mode_with_sources( diff --git a/tests/test_utils.py b/tests/test_utils.py index 2bae378f..ec06c0a8 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -3,6 +3,7 @@ import pathlib import tempfile import unittest +import warnings from pathlib import Path from unittest import mock @@ -13,6 +14,7 @@ from mmif import ( AnnotationTypes, Document, + DocumentTypes, Mmif ) from mmif.utils import sequence_helper as sqh @@ -45,13 +47,14 @@ def test_convert(self, value): self.assertAlmostEqual(value, out_t, delta=1000/self.FPS) # up to 1 frame error +@pytest.mark.filterwarnings('ignore::DeprecationWarning') class TestVideoDocumentHelper(unittest.TestCase): def setUp(self): self.fps = 29.97 self.mmif_obj = Mmif(validate=False) self.a_view = self.mmif_obj.new_view() self.video_doc = Document({ - "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", + "@type": DocumentTypes.VideoDocument, "properties": { "mime": "video", "id": "d1", @@ -61,7 +64,7 @@ def setUp(self): self.video_doc.add_property('fps', self.fps) self.mmif_obj.add_document(self.video_doc) - def test_extract_mid_frame(self): + def test_get_mid_framenum(self): tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=100, end=200, timeUnit='frame', document=self.video_doc.id) self.assertEqual(150, vdh.get_mid_framenum(self.mmif_obj, tf)) tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=0, end=200, timeUnit='frame', document=self.video_doc.id) @@ -69,8 +72,7 @@ def test_extract_mid_frame(self): tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=0, end=3, timeUnit='seconds', document=self.video_doc.id) self.assertEqual(vdh.convert(1.5, 's', 'f', self.fps), vdh.get_mid_framenum(self.mmif_obj, tf)) - @pytest.mark.filterwarnings('ignore::DeprecationWarning') - def test_extract_representative_frame(self): + def test_get_representative_framenum(self): tp = self.a_view.new_annotation(AnnotationTypes.TimePoint, timePoint=1500, timeUnit='milliseconds', document=self.video_doc.id) tf = self.a_view.new_annotation(AnnotationTypes.TimeFrame, start=1000, end=2000, timeUnit='milliseconds', document=self.video_doc.id) tf.add_property('representatives', [tp.id]) @@ -88,23 +90,26 @@ def test_extract_representative_frame(self): def test_get_framerate(self): self.assertAlmostEqual(29.97, vdh.get_framerate(self.video_doc), places=0) - @pytest.mark.filterwarnings('ignore::DeprecationWarning') - def test_frames_to_seconds(self): - self.assertAlmostEqual(3.337, vdh.framenum_to_second(self.video_doc, 100), places=0) - - @pytest.mark.filterwarnings('ignore::DeprecationWarning') - def test_frames_to_milliseconds(self): - self.assertAlmostEqual(3337.0, vdh.framenum_to_millisecond(self.video_doc, 100), places=0) - - @pytest.mark.filterwarnings('ignore::DeprecationWarning') - def test_seconds_to_frames(self): - self.assertAlmostEqual(100, vdh.second_to_framenum(self.video_doc, 3.337), places=0) - - @pytest.mark.filterwarnings('ignore::DeprecationWarning') - def test_milliseconds_to_frames(self): - self.assertAlmostEqual(100, vdh.millisecond_to_framenum(self.video_doc, 3337.0), places=0) + def test_get_framerate_without_cached_property(self): + # without an fps-like property, the rate comes from the container + fixture = pathlib.Path(__file__).parent / 'black-2997fps.mp4' + vd = Document({ + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": "r1", + "location": f"file://{fixture}"}, + }) + self.assertAlmostEqual(29.97, vdh.get_framerate(vd), places=1) + # any of the accepted spellings is honored without opening the file + for key in ('frameRate', 'frames_per_second', 'frame-per-second'): + with self.subTest(key=key): + aliased = Document({ + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": "r2", + "location": "file:///nonexistent.mp4", + key: 12.5}, + }) + self.assertEqual(12.5, vdh.get_framerate(aliased)) - @pytest.mark.filterwarnings('ignore::DeprecationWarning') def test_convert_roundtrip(self): # ms for 1 frame tolerance = 1000 / self.video_doc.get_property('fps') @@ -113,7 +118,6 @@ def test_convert_roundtrip(self): m2f2m = vdh.framenum_to_millisecond(self.video_doc, m2f) self.assertAlmostEqual(ms, m2f2m, delta=tolerance) - @pytest.mark.filterwarnings('ignore::DeprecationWarning') def test_sample_frames(self): s_frame = vdh.second_to_framenum(self.video_doc, 3) e_frame = vdh.second_to_framenum(self.video_doc, 5.5) @@ -147,7 +151,6 @@ def test_convert_timeframe(self): for times in zip((3.337, 6.674), vdh.convert_timeframe(self.mmif_obj, timeframe_ann, 's')): self.assertAlmostEqual(*times, places=0) - @pytest.mark.filterwarnings('ignore::DeprecationWarning') def test_extract_frames_as_images(self): frame_list = [5, 10, 15] target_images = vdh.extract_frames_as_images(self.video_doc, frame_list, as_PIL=False) @@ -163,45 +166,79 @@ def test_extract_frames_as_images(self): frame_list.append(tot_fcount + 1) new_target_images = vdh.extract_frames_as_images(self.video_doc, frame_list, as_PIL=False) self.assertEqual(4, len(frame_list)) + # the deprecated cv2 path keeps dropping unavailable frames, so its + # list shortens where the current API would hold a `None` instead self.assertEqual(3, len(new_target_images)) - def test_open_container(self): - # open_container sets fps/frameCount/duration as informational props + def test_deprecated_capture(self): + fixture = pathlib.Path(__file__).parent / 'black-2997fps.mp4' vd = Document({ - "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", - "properties": { - "mime": "video", - "id": "o1", - "location": f"file://{pathlib.Path(__file__).parent}/black-2997fps.mp4", - } + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": "c1", + "location": f"file://{fixture}"}, }) - c = vdh.open_container(vd) + with pytest.warns(DeprecationWarning, match='open_container'): + capture = vdh.capture(vd) try: self.assertAlmostEqual(29.97, vd.get_property('fps'), places=1) - self.assertGreater(vd.get_property('frameCount'), 0) self.assertGreater(vd.get_property('duration'), 0) finally: - c.close() + capture.release() + + def test_deprecated_frame_extractors(self): + tp = self.a_view.new_annotation( + AnnotationTypes.TimePoint, timePoint=1500, + timeUnit='milliseconds', document=self.video_doc.id) + tf = self.a_view.new_annotation( + AnnotationTypes.TimeFrame, start=1000, end=2000, + timeUnit='milliseconds', document=self.video_doc.id) + tf.add_property('representatives', [tp.id]) + self.assertIsNotNone(vdh.extract_mid_frame(self.mmif_obj, tf)) + self.assertIsNotNone( + vdh.extract_representative_frame(self.mmif_obj, tf)) + self.assertEqual( + [vdh.millisecond_to_framenum(self.video_doc, 1500)], + vdh.get_representative_framenums(self.mmif_obj, tf)) - def test_open_container_matroska(self): - # regression for issue #390: Matroska/webm leave the per-stream - # `frames`/`duration` fields unset, so open_container must fall back to - # the container-level duration instead of caching frameCount/duration 0. + def test_open_container(self): + # open_container sets fps/frameCount/duration as informational props. + # The webm case is a regression for issue #390: Matroska leaves the + # per-stream `frames`/`duration` fields unset, so open_container must + # fall back to the container-level duration instead of caching 0. + for fixture in ('black-2997fps.mp4', 'black-2997fps.webm'): + with self.subTest(fixture=fixture): + path = pathlib.Path(__file__).parent / fixture + vd = Document({ + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": "o1", + "location": f"file://{path}"}, + }) + c = vdh.open_container(vd) + try: + self.assertAlmostEqual(29.97, vd.get_property('fps'), + places=1) + self.assertGreater(vd.get_property('frameCount'), 0) + self.assertGreater(vd.get_property('duration'), 0) + finally: + c.close() + + def test_open_container_rejects_non_video(self): + for bad_doc in (None, Document({ + "@type": "http://mmif.clams.ai/vocabulary/TextDocument/v1", + "properties": {"mime": "text", "id": "t1"}})): + with self.subTest(doc=bad_doc): + with pytest.raises(ValueError): + vdh.open_container(bad_doc) + + def test_open_container_missing_file(self): + # a missing file is not a broken file, and must stay distinguishable vd = Document({ - "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", - "properties": { - "mime": "video", - "id": "o1", - "location": f"file://{pathlib.Path(__file__).parent}/black-2997fps.webm", - } + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": "m1", + "location": "file:///nonexistent-video.mp4"}, }) - c = vdh.open_container(vd) - try: - self.assertAlmostEqual(29.97, vd.get_property('fps'), places=1) - self.assertGreater(vd.get_property('frameCount'), 0) - self.assertGreater(vd.get_property('duration'), 0) - finally: - c.close() + with pytest.raises(FileNotFoundError): + vdh.open_container(vd) def test_sample_timepoints(self): # half-open interval; step in ms @@ -216,18 +253,31 @@ def test_sample_timepoints(self): vdh.sample_timepoints(0, 100, -10) def test_extract_images_from_timepoints(self): - # basic: three distinct timepoints + # in-range timepoints all yield an image, and nothing is reported ms_list = [1000, 2000, 3000] - imgs = vdh.extract_images_from_timepoints( - self.video_doc, ms_list, as_PIL=False) - self.assertEqual(3, len(imgs)) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + imgs = vdh.extract_images_from_timepoints( + self.video_doc, ms_list, as_PIL=False) + self.assertEqual(len(ms_list), len(imgs)) + self.assertTrue(all(img is not None for img in imgs)) + self.assertEqual([], [w for w in caught + if 'No image is available' in str(w.message)]) # empty input self.assertEqual( [], vdh.extract_images_from_timepoints(self.video_doc, [])) - # duplicates preserved in input order + # duplicates yield the same image object at each repeated position dup_ms = [500, 250, 500, 750, 250] dup_imgs = vdh.extract_images_from_timepoints(self.video_doc, dup_ms) - self.assertEqual(5, len(dup_imgs)) + self.assertEqual(len(dup_ms), len(dup_imgs)) + self.assertIs(dup_imgs[0], dup_imgs[2]) + self.assertIs(dup_imgs[1], dup_imgs[4]) + + def test_extract_images_as_PIL(self): + Image = pytest.importorskip('PIL.Image') + imgs = vdh.extract_images_from_timepoints( + self.video_doc, [1000], as_PIL=True) + self.assertIsInstance(imgs[0], Image.Image) def _make_timepoints(self, count): # Explicit aid avoids a pre-existing clams-vocabulary / mmif-python @@ -418,7 +468,7 @@ def test_pts_offset_regression(self): import av fixture = pathlib.Path(__file__).parent / 'testsrc-2997fps-ptsoffset.mp4' vd = Document({ - "@type": "http://mmif.clams.ai/vocabulary/VideoDocument/v1", + "@type": DocumentTypes.VideoDocument, "properties": { "mime": "video", "id": "p1", @@ -463,6 +513,303 @@ def test_pts_offset_regression(self): 'PTS-offset videos') +class TestBrokenVideoDocument(unittest.TestCase): + """ + Covers videos that hold fewer images than their container metadata + promises (issue #400). A byte-truncated file surfaces differently per + container and per cut position, so each path gets its own fixture. + """ + + fixtures_dir = pathlib.Path(__file__).parent + + @classmethod + def setUpClass(cls): + # fixtures are byte-identical for every test, so they are built once + cls.tmpdir = tempfile.TemporaryDirectory() + cls.addClassCleanup(cls.tmpdir.cleanup) + tmp = pathlib.Path(cls.tmpdir.name) + healthy_mp4 = cls.fixtures_dir / 'black-2997fps.mp4' + faststart = tmp / 'faststart.mp4' + cls._remux_faststart(healthy_mp4, faststart) + cls.cuts = { + # cut mid-packet: the mp4 demuxer hands the partial packet to the + # decoder, which raises + 'unclean.mp4': (cls._truncate(faststart, tmp / 'unclean.mp4'), + True), + # cut at a packet boundary: a clean EOF, indistinguishable from + # the end of a healthy video + 'clean.mp4': (cls._truncate(faststart, tmp / 'clean.mp4', + cls._clean_cut_offset(faststart)), + False), + # Matroska drops the incomplete block, so any cut is clean + 'cut.webm': (cls._truncate(cls.fixtures_dir / 'black-2997fps.webm', + tmp / 'cut.webm'), + False), + } + # `moov` sits at the end of the shipped fixtures, so a cut copy of the + # original cannot be opened at all + cls.trailing_moov = cls._truncate(healthy_mp4, tmp / 'moovend.mp4') + # too short to hold a video stream, yet still openable + cls.tiny = cls._truncate(faststart, tmp / 'tiny.mp4', 200) + cls.healthy = healthy_mp4 + + @staticmethod + def _doc(path, aid='d1'): + return Document({ + "@type": DocumentTypes.VideoDocument, + "properties": {"mime": "video", "id": aid, + "location": f"file://{path}"} + }) + + @staticmethod + def _remux_faststart(src, dst): + # a cut copy of an mp4 is only decodable at all when the `moov` atom + # sits at the front; the shipped fixtures carry it at the end + import av + faststart = {'movflags': 'faststart'} + with av.open(str(src)) as inp, \ + av.open(str(dst), 'w', options=faststart) as out: + in_stream = inp.streams.video[0] + try: + out_stream = out.add_stream_from_template(in_stream) + except AttributeError: # PyAV < 14 + out_stream = out.add_stream(template=in_stream) + for packet in inp.demux(in_stream): + if packet.dts is None: + continue + packet.stream = out_stream + out.mux(packet) + + @staticmethod + def _truncate(src, dst, nbytes=None): + data = pathlib.Path(src).read_bytes() + dst.write_bytes(data[:nbytes if nbytes else len(data) // 2]) + return dst + + @staticmethod + def _clean_cut_offset(path): + # end of the last complete packet before the halfway point + import av + half = pathlib.Path(path).stat().st_size // 2 + with av.open(str(path)) as container: + stream = container.streams.video[0] + return next(p.pos for p in container.demux(stream) + if p.pos is not None and p.pos >= half) + + @staticmethod + def _last_decodable_ms(path): + # ground truth for the assertions below, computed without vdh; the + # decoder is flushed after an error, as vdh does, so that frames held + # back for reordering count here too + import av + last = None + + def frames(container, stream): + try: + for packet in container.demux(stream): + yield from packet.decode() + except av.error.FFmpegError: + try: + yield from stream.codec_context.decode(None) + except av.error.FFmpegError: + pass + + with av.open(str(path)) as container: + stream = container.streams.video[0] + for frame in frames(container, stream): + if frame.pts is not None: + pts_ms = round(frame.pts * stream.time_base * 1000) + last = max(last or 0, pts_ms) + return last + + @staticmethod + def _missing_image_warnings(caught): + # the module also emits unrelated warnings (e.g. vocabulary version + # mismatches), and a single call must report missing images once + return [str(w.message) for w in caught + if 'No image is available' in str(w.message)] + + def _extract(self, doc, timepoints, **kwargs): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + images = vdh.extract_images_from_timepoints( + doc, timepoints, **kwargs) + return images, self._missing_image_warnings(caught) + + def _frame_ms(self, doc): + return int(1000 / vdh.get_framerate(doc)) + + def test_truncated_file_yields_none_past_last_frame(self): + import av + for name, (path, expect_decode_error) in self.cuts.items(): + with self.subTest(fixture=name): + doc = self._doc(path) + last_ok = self._last_decodable_ms(path) + self.assertIsNotNone(last_ok, + 'fixture should decode at least once') + frame_ms = self._frame_ms(doc) + # the last decodable frame covers one frame duration, and + # nothing beyond it; 2ms of slack absorbs ms/tick rounding + timepoints = [0, last_ok, last_ok + frame_ms - 2, + last_ok + frame_ms + 2, 59000] + images, messages = self._extract(doc, timepoints) + self.assertEqual(len(timepoints), len(images)) + for i in (0, 1, 2): + self.assertIsNotNone(images[i], f'position {i}') + self.assertIsNone(images[3]) + self.assertIsNone(images[4]) + # within the tolerance, the last frame itself is reused + self.assertEqual(images[1].tobytes(), images[2].tobytes()) + self.assertEqual(1, len(messages)) + self.assertIn(vdh._media_check_hint(str(path)), messages[0]) + self.assertEqual( + expect_decode_error, + av.error.InvalidDataError.__name__ in messages[0]) + self.assertEqual(expect_decode_error, + 'first decoding error:' in messages[0]) + + def test_missing_only_past_duration_is_not_flagged(self): + # a file that merely ends is not suspect, even after a clean cut + path, _ = self.cuts['cut.webm'] + doc = self._doc(path) + vdh.open_container(doc).close() + duration = doc.get_property('duration') + images, messages = self._extract(doc, [duration + 10000]) + self.assertEqual([None], images) + self.assertEqual(1, len(messages)) + self.assertNotIn(vdh._media_check_hint(str(path)), messages[0]) + + def test_missing_before_and_past_duration_is_flagged(self): + # the earliest missing timepoint decides, and both are listed + path, _ = self.cuts['cut.webm'] + doc = self._doc(path) + vdh.open_container(doc).close() + duration = doc.get_property('duration') + last_ok = self._last_decodable_ms(path) + timepoints = [last_ok + 5000, duration + 10000] + images, messages = self._extract(doc, timepoints) + self.assertEqual([None, None], images) + self.assertEqual(1, len(messages)) + self.assertIn(vdh._media_check_hint(str(path)), messages[0]) + self.assertIn(f'{timepoints[0]}ms, {timepoints[1]}ms', messages[0]) + + def test_repeated_missing_timepoint_counts_positions(self): + path, _ = self.cuts['cut.webm'] + images, messages = self._extract(self._doc(path), [59000] * 3) + self.assertEqual([None, None, None], images) + self.assertIn('for 3 timepoint(s)', messages[0]) + + def test_missing_entry_with_as_PIL(self): + Image = pytest.importorskip('PIL.Image') + path, _ = self.cuts['cut.webm'] + images, _ = self._extract(self._doc(path), [0, 59000], as_PIL=True) + self.assertIsInstance(images[0], Image.Image) + self.assertIsNone(images[1]) + + def test_tolerance_falls_back_to_stream_rate(self): + # `frame.duration` only exists in PyAV >= 15; older versions must get + # the same tolerance from the stream frame rate + class _NoDuration: + def __init__(self, frame): + self._frame = frame + self.duration = None + + def __getattr__(self, name): + return getattr(self._frame, name) + + real_decode_frames = vdh._decode_frames + + def stripped(container, stream, decode_errors): + for frame in real_decode_frames(container, stream, decode_errors): + yield _NoDuration(frame) + + doc = self._doc(self.healthy) + vdh.open_container(doc).close() + duration = doc.get_property('duration') + with mock.patch.object(vdh, '_decode_frames', stripped): + images, messages = self._extract(doc, [duration - 1]) + self.assertIsNotNone(images[0]) + self.assertEqual([], messages) + + def test_unopenable_files_raise_with_hint(self): + cases = {'trailing moov': self.trailing_moov, 'tiny prefix': self.tiny} + for label, path in cases.items(): + with self.subTest(case=label): + with pytest.raises(RuntimeError) as excinfo: + vdh.open_container(self._doc(path)) + self.assertIn(vdh._media_check_hint(str(path)), + str(excinfo.value)) + # get_framerate goes through open_container, so it reports the same way + with pytest.raises(RuntimeError): + vdh.get_framerate(self._doc(self.trailing_moov)) + + def test_failed_seek_raises_with_hint(self): + # PyAV containers are immutable extension types, so the failure is + # injected through a proxy around the real container + import av + path, _ = self.cuts['cut.webm'] + doc = self._doc(path) + + class _SeekFails: + def __init__(self, container): + self._container = container + + def __getattr__(self, name): + return getattr(self._container, name) + + def seek(self, *args, **kwargs): + raise av.error.InvalidDataError(1094995529, 'seek failed') + + real_open_container = vdh.open_container + + def wrapped(video_document): + return _SeekFails(real_open_container(video_document)) + + with mock.patch.object(vdh, 'open_container', wrapped): + with pytest.raises(RuntimeError) as excinfo: + vdh.extract_images_from_timepoints(doc, [1000]) + self.assertIn(vdh._media_check_hint(str(path)), str(excinfo.value)) + self.assertIn('seek failed', str(excinfo.value)) + + def test_empty_request_never_opens_the_file(self): + doc = self._doc(self.trailing_moov) + self.assertEqual([], vdh.extract_images_from_timepoints(doc, [])) + + def test_none_stays_parallel_with_sources(self): + # the breakage that motivated issue #400: apps pair images with + # timepoints by list index + mmif_obj = Mmif(validate=False) + doc = self._doc(self.healthy) + mmif_obj.add_document(doc) + vdh.open_container(doc).close() + duration = doc.get_property('duration') + view = mmif_obj.new_view() + tp_ids = [] + for i, ms in enumerate([1000, duration + 10000, duration + 20000]): + tp = view.new_annotation( + AnnotationTypes.TimePoint, aid=f'tp_{i}', timePoint=ms, + timeUnit='milliseconds', document=doc.id) + tp_ids.append(tp.id) + time_frame = view.new_annotation( + AnnotationTypes.TimeFrame, aid='tf_0', document=doc.id, + timeUnit='milliseconds', start=1000, end=duration + 20000, + targets=tp_ids, representatives=tp_ids) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + by_mode = vdh.extract_images_by_mode_with_sources( + mmif_obj, time_frame, mode=vdh.SamplingMode.REPRESENTATIVES) + by_count = vdh.extract_images_by_count_with_sources( + mmif_obj, time_frame, fraction=1.0) + for label, (images, sources) in {'by_mode': by_mode, + 'by_count': by_count}.items(): + with self.subTest(entrypoint=label): + self.assertEqual(tp_ids, list(sources)) + self.assertEqual(len(sources), len(images)) + self.assertIsNotNone(images[0]) + self.assertIsNone(images[1]) + self.assertIsNone(images[2]) + + class TestSequenceHelper(unittest.TestCase): def test_validate_labelset(self): From 1326027df099aef5c501e3e42fed7aa22ed3239b Mon Sep 17 00:00:00 2001 From: Keigh Rim Date: Mon, 5 Oct 2026 11:36:24 -0400 Subject: [PATCH 2/2] pinned pyav, so we can reliable use duration feature --- mmif/utils/video_document_helper.py | 3 +-- pyproject.toml | 2 +- tests/test_utils.py | 29 +++++++++++++++++++++++++---- 3 files changed, 27 insertions(+), 7 deletions(-) diff --git a/mmif/utils/video_document_helper.py b/mmif/utils/video_document_helper.py index e6360784..65edc93d 100644 --- a/mmif/utils/video_document_helper.py +++ b/mmif/utils/video_document_helper.py @@ -359,8 +359,7 @@ def extract_images_from_timepoints( if rate else 0) def _tolerance_ticks(frame): - return (getattr(frame, 'duration', None) - or default_tolerance_ticks) + return frame.duration or default_tolerance_ticks # convert each target ms to stream ticks (PTS units) target_ticks = [round(t_ms / 1000.0 / time_base) diff --git a/pyproject.toml b/pyproject.toml index 32985ec9..97611807 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,7 @@ source = "https://github.com/clamsproject/mmif-python" mmif-spec = "https://mmif.clams.ai/1.2.0" [project.optional-dependencies] -cv = ["pillow", "opencv-python", "ffmpeg-python", "wurlitzer", "av"] +cv = ["pillow", "opencv-python", "ffmpeg-python", "wurlitzer", "av>=15"] seq = ["numpy"] docs = ["sphinx>=7.0,<8.0", "furo", "m2r2", "autodoc-pydantic"] test = [ diff --git a/tests/test_utils.py b/tests/test_utils.py index ec06c0a8..f4f6e185 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -549,7 +549,9 @@ def setUpClass(cls): # `moov` sits at the end of the shipped fixtures, so a cut copy of the # original cannot be opened at all cls.trailing_moov = cls._truncate(healthy_mp4, tmp / 'moovend.mp4') - # too short to hold a video stream, yet still openable + # opens fine, holds no video stream; the realistic case is an audio + # file registered as a VideoDocument, and a tiny prefix behaves alike + cls.audio_only = cls._write_audio_only(tmp / 'audio.wav') cls.tiny = cls._truncate(faststart, tmp / 'tiny.mp4', 200) cls.healthy = healthy_mp4 @@ -580,6 +582,23 @@ def _remux_faststart(src, dst): packet.stream = out_stream out.mux(packet) + @staticmethod + def _write_audio_only(dst): + import av + import numpy as np + rate = 8000 + with av.open(str(dst), 'w') as out: + stream = out.add_stream('pcm_s16le', rate=rate, layout='mono') + frame = av.AudioFrame.from_ndarray( + np.zeros((1, rate), dtype='int16'), format='s16', + layout='mono') + frame.sample_rate = rate + for packet in stream.encode(frame): + out.mux(packet) + for packet in stream.encode(None): + out.mux(packet) + return dst + @staticmethod def _truncate(src, dst, nbytes=None): data = pathlib.Path(src).read_bytes() @@ -707,8 +726,8 @@ def test_missing_entry_with_as_PIL(self): self.assertIsNone(images[1]) def test_tolerance_falls_back_to_stream_rate(self): - # `frame.duration` only exists in PyAV >= 15; older versions must get - # the same tolerance from the stream frame rate + # a frame may carry no duration of its own, in which case the same + # tolerance must come from the stream frame rate class _NoDuration: def __init__(self, frame): self._frame = frame @@ -732,7 +751,9 @@ def stripped(container, stream, decode_errors): self.assertEqual([], messages) def test_unopenable_files_raise_with_hint(self): - cases = {'trailing moov': self.trailing_moov, 'tiny prefix': self.tiny} + cases = {'trailing moov': self.trailing_moov, + 'audio only': self.audio_only, + 'tiny prefix': self.tiny} for label, path in cases.items(): with self.subTest(case=label): with pytest.raises(RuntimeError) as excinfo: