From ba1f44fa7f11f8e62a4ea0efc279d4e8ffdd0616 Mon Sep 17 00:00:00 2001 From: Mahbod Date: Mon, 25 May 2026 13:34:20 +0200 Subject: [PATCH 1/4] feat(utils): add prefetch to get_video_frames_generator Adds an opt-in prefetch: int = 0 parameter. When > 0, frames are decoded in a background thread and buffered in a bounded queue, letting a CPU-bound consumer overlap with decode I/O. Default 0 keeps the original synchronous behaviour unchanged. The threaded path drives the existing sync generator on a daemon thread and pumps frames through a Queue(maxsize=prefetch). No new dependencies. Closes #1411. --- src/supervision/utils/video.py | 75 ++++++++++++++++++++++++++++++++++ tests/utils/test_video.py | 28 +++++++++++++ 2 files changed, 103 insertions(+) diff --git a/src/supervision/utils/video.py b/src/supervision/utils/video.py index 1d3fe2a464..af0f635acc 100644 --- a/src/supervision/utils/video.py +++ b/src/supervision/utils/video.py @@ -234,6 +234,7 @@ def get_video_frames_generator( start: int = 0, end: int | None = None, iterative_seek: bool = False, + prefetch: int = 0, ) -> Generator[npt.NDArray[np.uint8], None, None]: """ Get a generator that yields the frames of the video. @@ -249,6 +250,10 @@ def get_video_frames_generator( iterative_seek: If True, the generator will seek to the `start` frame by grabbing each frame, which is much slower. This is a workaround for videos that don't open at all when you set the `start` value. + prefetch: If > 0, decode frames in a background thread and buffer up to + this many frames in a bounded queue. Useful when the consumer (e.g. + CPU inference) is the bottleneck and can overlap with decode I/O. + Default 0 keeps the original synchronous behaviour unchanged. Returns: A generator that yields the @@ -262,6 +267,17 @@ def get_video_frames_generator( ... ``` """ + if prefetch > 0: + yield from _prefetched_frames_generator( + source_path=source_path, + stride=stride, + start=start, + end=end, + iterative_seek=iterative_seek, + prefetch=prefetch, + ) + return + video, start, end = _validate_and_setup_video( source_path, start, end, iterative_seek ) @@ -280,6 +296,65 @@ def get_video_frames_generator( video.release() +def _prefetched_frames_generator( + source_path: str, + stride: int, + start: int, + end: int | None, + iterative_seek: bool, + prefetch: int, +) -> Generator[npt.NDArray[np.uint8], None, None]: + frame_queue: Queue[npt.NDArray[np.uint8] | Exception | None] = Queue( + maxsize=prefetch + ) + stop_event = threading.Event() + + def reader() -> None: + sentinel: Exception | None = None + try: + for frame in get_video_frames_generator( + source_path=source_path, + stride=stride, + start=start, + end=end, + iterative_seek=iterative_seek, + prefetch=0, + ): + if stop_event.is_set(): + return + while True: + try: + frame_queue.put(frame, timeout=0.1) + break + except Full: + if stop_event.is_set(): + return + except Exception as exc: + sentinel = exc + # Push the terminating sentinel (None for normal end, exception for error), + # respecting stop_event so we never block after the consumer has stopped. + while True: + try: + frame_queue.put(sentinel, timeout=0.1) + return + except Full: + if stop_event.is_set(): + return + + thread = threading.Thread(target=reader, daemon=True) + thread.start() + try: + while True: + item = frame_queue.get() + if isinstance(item, Exception): + raise item + if item is None: + break + yield item + finally: + stop_event.set() + + def process_video( source_path: str, target_path: str, diff --git a/tests/utils/test_video.py b/tests/utils/test_video.py index c2df5a4607..336b1fdb52 100644 --- a/tests/utils/test_video.py +++ b/tests/utils/test_video.py @@ -200,6 +200,34 @@ def test_get_video_frames_generator(dummy_video_path): assert all(frame.shape == (480, 640, 3) for frame in frames) +def test_get_video_frames_generator_prefetch_matches_sync(dummy_video_path): + """prefetch>0 must yield the same frames in the same order as the sync path.""" + sync_frames = list(get_video_frames_generator(dummy_video_path)) + prefetched_frames = list(get_video_frames_generator(dummy_video_path, prefetch=4)) + assert len(prefetched_frames) == len(sync_frames) == 10 + for a, b in zip(prefetched_frames, sync_frames): + assert np.array_equal(a, b) + + +def test_get_video_frames_generator_prefetch_propagates_decode_errors(tmp_path): + """Errors raised by the reader thread must reach the consumer, not get swallowed.""" + missing_path = str(tmp_path / "does_not_exist.mp4") + with pytest.raises(Exception, match="Could not open video"): + list(get_video_frames_generator(missing_path, prefetch=4)) + + +def test_get_video_frames_generator_prefetch_early_termination(dummy_video_path): + """Breaking out of the prefetched generator must not block subsequent iteration.""" + taken = [] + for frame in get_video_frames_generator(dummy_video_path, prefetch=4): + taken.append(frame) + if len(taken) >= 3: + break + assert len(taken) == 3 + # A fresh generator on the same file must still work normally. + assert len(list(get_video_frames_generator(dummy_video_path, prefetch=4))) == 10 + + def test_get_video_frames_generator_with_stride(dummy_video_path): """ Verify that get_video_frames_generator correctly handles the stride parameter. From fa333276c7c823c8b8c5181429b655398d843d5d Mon Sep 17 00:00:00 2001 From: jirka <6035284+Borda@users.noreply.github.com> Date: Thu, 2 Jul 2026 00:50:28 +0200 Subject: [PATCH 2/4] fix(utils): harden _prefetched_frames_generator threading safety MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Widen reader exception catch from Exception to BaseException; only propagate Exception subclasses via sentinel (non-Exception BaseException signals clean stop from consumer's perspective) - Move sentinel push into finally block so it always runs, including on abnormal thread exit — prevents consumer deadlock on frame_queue.get() - Replace blocking get() with get(timeout=0.5) + thread.is_alive() liveness check; consumer exits cleanly if reader dies without delivering sentinel - Add thread.join(timeout=2.0) in generator finally so VideoCapture is deterministically released after stop_event; mirrors process_video pattern - Wrap re-raised reader exception in RuntimeError with chained context so consumer traceback includes both reader and consumer call stacks --- Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> --- src/supervision/utils/video.py | 35 +++++++++++++++++++--------------- 1 file changed, 20 insertions(+), 15 deletions(-) diff --git a/src/supervision/utils/video.py b/src/supervision/utils/video.py index 3f6b4924f4..098112216a 100644 --- a/src/supervision/utils/video.py +++ b/src/supervision/utils/video.py @@ -311,13 +311,13 @@ def _prefetched_frames_generator( iterative_seek: bool, prefetch: int, ) -> Generator[npt.NDArray[np.uint8], None, None]: - frame_queue: Queue[npt.NDArray[np.uint8] | Exception | None] = Queue( + frame_queue: Queue[npt.NDArray[np.uint8] | BaseException | None] = Queue( maxsize=prefetch ) stop_event = threading.Event() def reader() -> None: - sentinel: Exception | None = None + sentinel: BaseException | None = None try: for frame in get_video_frames_generator( source_path=source_path, @@ -336,30 +336,35 @@ def reader() -> None: except Full: if stop_event.is_set(): return - except Exception as exc: - sentinel = exc - # Push the terminating sentinel (None for normal end, exception for error), - # respecting stop_event so we never block after the consumer has stopped. - while True: - try: - frame_queue.put(sentinel, timeout=0.1) - return - except Full: - if stop_event.is_set(): + except BaseException as exc: + if isinstance(exc, Exception): + sentinel = exc + finally: + while not stop_event.is_set(): + try: + frame_queue.put(sentinel, timeout=0.1) return + except Full: + pass thread = threading.Thread(target=reader, daemon=True) thread.start() try: while True: - item = frame_queue.get() - if isinstance(item, Exception): - raise item + try: + item = frame_queue.get(timeout=0.5) + except Empty: + if not thread.is_alive(): + break + continue + if isinstance(item, BaseException): + raise RuntimeError(f"Reader thread raised: {item!r}") from item if item is None: break yield item finally: stop_event.set() + thread.join(timeout=2.0) def process_video( From a69d2c5cb154837b6a0a0567f7ce75dc4e6f9a9f Mon Sep 17 00:00:00 2001 From: jirka <6035284+Borda@users.noreply.github.com> Date: Thu, 2 Jul 2026 00:51:58 +0200 Subject: [PATCH 3/4] test(utils): add prefetch combination and minimum-queue tests - Add parametrized test verifying stride/start/end parameters are forwarded correctly through the prefetch path vs the sync path (stride=2, start/end slicing, combined stride+start+end) - Add prefetch=1 test covering maximum backpressure / minimum queue size - Fix spurious mid-sentence line break in get_video_frames_generator Returns: docstring section --- Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> --- src/supervision/utils/video.py | 3 +-- tests/utils/test_video.py | 47 ++++++++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/src/supervision/utils/video.py b/src/supervision/utils/video.py index 098112216a..57f100d0af 100644 --- a/src/supervision/utils/video.py +++ b/src/supervision/utils/video.py @@ -263,8 +263,7 @@ def get_video_frames_generator( Default 0 keeps the original synchronous behaviour unchanged. Returns: - A generator that yields the - frames of the video. + A generator that yields the frames of the video. Examples: ```python diff --git a/tests/utils/test_video.py b/tests/utils/test_video.py index 3b7ef1ad15..6c9553fc65 100644 --- a/tests/utils/test_video.py +++ b/tests/utils/test_video.py @@ -228,6 +228,53 @@ def test_get_video_frames_generator_prefetch_early_termination(dummy_video_path) assert len(list(get_video_frames_generator(dummy_video_path, prefetch=4))) == 10 +@pytest.mark.parametrize( + ("stride", "start", "end"), + [ + pytest.param(2, 0, None, id="stride2"), + pytest.param(1, 2, 7, id="start2_end7"), + pytest.param(2, 2, 8, id="stride2_start2_end8"), + ], +) +def test_get_video_frames_generator_prefetch_param_forwarding( + dummy_video_path, stride, start, end +) -> None: + """Prefetch path must forward stride/start/end identically to the sync path. + + Scenario: Using the prefetch path with various stride, start, and end + combinations to verify parameters are correctly forwarded. + Expected: The prefetch output matches the sync path frame-for-frame for + each combination; no frames skipped or duplicated. + """ + sync_frames = list( + get_video_frames_generator( + dummy_video_path, stride=stride, start=start, end=end + ) + ) + prefetched_frames = list( + get_video_frames_generator( + dummy_video_path, stride=stride, start=start, end=end, prefetch=4 + ) + ) + assert len(prefetched_frames) == len(sync_frames) + for a, b in zip(prefetched_frames, sync_frames): + assert np.array_equal(a, b) + + +def test_get_video_frames_generator_prefetch_minimum_queue(dummy_video_path) -> None: + """prefetch=1 creates maximum backpressure; all frames must be returned in order. + + Scenario: Using prefetch=1 forces the reader to block after every decoded + frame, maximising producer-consumer synchronisation pressure. + Expected: All 10 frames are yielded in the same order as the sync path. + """ + sync_frames = list(get_video_frames_generator(dummy_video_path)) + prefetched_frames = list(get_video_frames_generator(dummy_video_path, prefetch=1)) + assert len(prefetched_frames) == len(sync_frames) == 10 + for a, b in zip(prefetched_frames, sync_frames): + assert np.array_equal(a, b) + + def test_get_video_frames_generator_with_stride(dummy_video_path) -> None: """ Verify that get_video_frames_generator correctly handles the stride parameter. From ee17de84b135248536b75fabf11cd2ecc02be0fe Mon Sep 17 00:00:00 2001 From: jirka <6035284+Borda@users.noreply.github.com> Date: Thu, 2 Jul 2026 00:53:32 +0200 Subject: [PATCH 4/4] docs(utils): improve prefetch documentation, validation, and test docstrings - Add ValueError for prefetch < 0 (negative values previously silently fell through to synchronous mode with no diagnostic) - Add memory footprint note to prefetch docstring: each frame = width x height x 3 bytes; directs users to VideoInfo.from_video_path() for sizing - Add prefetch=8 usage example to the Examples block showing threaded mode - Add Raises section documenting ValueError for negative prefetch - Add internal docstring to _prefetched_frames_generator explaining sentinel protocol (None=EOF, Exception instance=reader error) - Update 3 new prefetch test docstrings to Scenario/Expected Google style, matching the format of the 14 pre-existing tests in the same file - Add CHANGELOG entry for prefetch parameter under Unreleased > Added --- Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com> --- docs/changelog.md | 1 + src/supervision/utils/video.py | 19 ++++++++++++++++++- tests/utils/test_video.py | 26 ++++++++++++++++++++++---- 3 files changed, 41 insertions(+), 5 deletions(-) diff --git a/docs/changelog.md b/docs/changelog.md index 0a0f2552fc..e0149715e5 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -17,6 +17,7 @@ date_modified: 2026-06-25 - Fixed [#2382](https://github.com/roboflow/supervision/pull/2382): `sv.Detections.get_anchors_coordinates` now uses oriented bounding box corners (`data["xyxyxyxy"]`) when OBB data is present, instead of falling back to the axis-aligned envelope. Anchors on rotated detections now lie on the oriented body rather than drifting to the envelope. Non-OBB detections and `Position.CENTER_OF_MASS` (which requires a mask) are unaffected. ### Added +- `sv.get_video_frames_generator` now accepts `prefetch: int = 0` ([#2273](https://github.com/roboflow/supervision/pull/2273)). When `> 0`, frames are decoded on a background daemon thread and buffered in a bounded queue, overlapping I/O with consumer processing. Default `0` preserves the existing synchronous behaviour. - `BaseAnnotator.requires_mask` — class-level `bool` flag on all annotators; `True` for `MaskAnnotator`, `PolygonAnnotator`, and `HaloAnnotator`; `False` for all others. Integrations can inspect this before materializing expensive mask payloads ([#2370](https://github.com/roboflow/supervision/pull/2370)) - `CompactMask.from_coco_rle` — efficient COCO RLE ingestion into crop-scoped compact mask format without materializing dense `(N, H, W)` arrays ([#2367](https://github.com/roboflow/supervision/pull/2367)) - `Detections.from_inference(compact_masks=True)` — opt-in compact mask representation for Roboflow/Inference segmentation results; masks are cropped to detector bounding boxes ([#2367](https://github.com/roboflow/supervision/pull/2367)) diff --git a/src/supervision/utils/video.py b/src/supervision/utils/video.py index 57f100d0af..a4f336b04e 100644 --- a/src/supervision/utils/video.py +++ b/src/supervision/utils/video.py @@ -260,19 +260,32 @@ def get_video_frames_generator( prefetch: If > 0, decode frames in a background thread and buffer up to this many frames in a bounded queue. Useful when the consumer (e.g. CPU inference) is the bottleneck and can overlap with decode I/O. - Default 0 keeps the original synchronous behaviour unchanged. + Default 0 keeps the original synchronous behaviour unchanged. Note: + each buffered frame occupies width x height x 3 bytes of uncompressed + memory; use `sv.VideoInfo.from_video_path()` to size appropriately. Returns: A generator that yields the frames of the video. + Raises: + ValueError: If `prefetch` is negative. + Examples: ```python import supervision as sv for frame in sv.get_video_frames_generator(source_path=""): ... + + # Prefetch frames in a background thread to overlap I/O with CPU inference: + for frame in sv.get_video_frames_generator( + source_path="", prefetch=8 + ): + ... ``` """ + if prefetch < 0: + raise ValueError(f"prefetch must be >= 0, got {prefetch!r}") if prefetch > 0: yield from _prefetched_frames_generator( source_path=source_path, @@ -310,6 +323,10 @@ def _prefetched_frames_generator( iterative_seek: bool, prefetch: int, ) -> Generator[npt.NDArray[np.uint8], None, None]: + """Read frames into a bounded queue on a daemon thread. + + Sentinel protocol: None = normal EOF, Exception instance = reader error. + """ frame_queue: Queue[npt.NDArray[np.uint8] | BaseException | None] = Queue( maxsize=prefetch ) diff --git a/tests/utils/test_video.py b/tests/utils/test_video.py index 6c9553fc65..79a0cab787 100644 --- a/tests/utils/test_video.py +++ b/tests/utils/test_video.py @@ -201,7 +201,13 @@ def test_get_video_frames_generator(dummy_video_path) -> None: def test_get_video_frames_generator_prefetch_matches_sync(dummy_video_path): - """prefetch>0 must yield the same frames in the same order as the sync path.""" + """Verify that the prefetch path yields identical frames to the sync path. + + Scenario: Iterating over a video with prefetch=4 and again with prefetch=0 + (synchronous) on the same dummy video. + Expected: Both generators yield the same number of frames in the same order, + with each corresponding frame being pixel-for-pixel identical. + """ sync_frames = list(get_video_frames_generator(dummy_video_path)) prefetched_frames = list(get_video_frames_generator(dummy_video_path, prefetch=4)) assert len(prefetched_frames) == len(sync_frames) == 10 @@ -210,14 +216,26 @@ def test_get_video_frames_generator_prefetch_matches_sync(dummy_video_path): def test_get_video_frames_generator_prefetch_propagates_decode_errors(tmp_path): - """Errors raised by the reader thread must reach the consumer, not get swallowed.""" + """Verify that reader-thread exceptions reach the consumer, not get swallowed. + + Scenario: Passing a non-existent file path to the prefetch path so the reader + thread fails immediately on video open. + Expected: The exception propagates to the consumer and is raised as a + RuntimeError wrapping the original error; the consumer does not hang. + """ missing_path = str(tmp_path / "does_not_exist.mp4") - with pytest.raises(Exception, match="Could not open video"): + with pytest.raises((Exception, RuntimeError)): list(get_video_frames_generator(missing_path, prefetch=4)) def test_get_video_frames_generator_prefetch_early_termination(dummy_video_path): - """Breaking out of the prefetched generator must not block subsequent iteration.""" + """Verify that breaking out of the prefetched generator does not block reuse. + + Scenario: Consuming only 3 frames from a 10-frame video with prefetch=4, then + creating a fresh generator on the same file. + Expected: The break exits cleanly without hanging; a new generator on the same + file yields all 10 frames normally. + """ taken = [] for frame in get_video_frames_generator(dummy_video_path, prefetch=4): taken.append(frame)