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.

Author: Ace3Z

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,39 @@ 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] | None] = Queue(maxsize=prefetch)
+
+    def reader() -> 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,
+            ):
+                frame_queue.put(frame)
+        finally:
+            frame_queue.put(None)
+
+    thread = threading.Thread(target=reader, daemon=True)
+    thread.start()
+    while True:
+        frame = frame_queue.get()
+        if frame is None:
+            break
+        yield frame
+
+
 def process_video(
     source_path: str,
     target_path: str,

tests/utils/test_video.py

@@ -200,6 +200,15 @@ 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_with_stride(dummy_video_path):
     """
     Verify that get_video_frames_generator correctly handles the stride parameter.