Skip to content

Video Utils

VideoInfo

supervision.utils.video.VideoInfo dataclass

A class to store video information, including width, height, fps and total number of frames.

Attributes:

Name Type Description
width int

width of the video in pixels
height int

height of the video in pixels
fps float

frames per second of the video as a float. Common values include 23.976, 24.0, 25.0, 29.97, 30.0, 59.94, and 60.0.
total_frames int | None

total number of frames in the video, default is None

Examples:

import supervision as sv

video_info = sv.VideoInfo.from_video_path(video_path="<SOURCE_VIDEO_FILE>")

video_info
# VideoInfo(width=3840, height=2160, fps=25.0, total_frames=538)

video_info.resolution_wh
# (3840, 2160)
Source code in src/supervision/utils/video.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class VideoInfo:
    """
    A class to store video information, including width, height, fps and
        total number of frames.

    Attributes:
        width: width of the video in pixels
        height: height of the video in pixels
        fps: frames per second of the video as a float. Common values include
            23.976, 24.0, 25.0, 29.97, 30.0, 59.94, and 60.0.
        total_frames: total number of frames in the video,
            default is None

    Examples:
        ```python
        import supervision as sv

        video_info = sv.VideoInfo.from_video_path(video_path="<SOURCE_VIDEO_FILE>")

        video_info
        # VideoInfo(width=3840, height=2160, fps=25.0, total_frames=538)

        video_info.resolution_wh
        # (3840, 2160)
        ```
    """

    width: int
    height: int
    fps: float
    total_frames: int | None = None

    @classmethod
    def from_video_path(cls, video_path: str) -> VideoInfo:
        video = cv2.VideoCapture(video_path)
        if not video.isOpened():
            raise Exception(f"Could not open video at {video_path}")

        width = int(video.get(cv2.CAP_PROP_FRAME_WIDTH))
        height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
        fps = float(video.get(cv2.CAP_PROP_FPS))
        total_frames = int(video.get(cv2.CAP_PROP_FRAME_COUNT))
        video.release()
        return VideoInfo(width, height, fps, total_frames)

    @property
    def resolution_wh(self) -> tuple[int, int]:
        return self.width, self.height

VideoSink

supervision.utils.video.VideoSink

Context manager that saves video frames to a file using OpenCV.

Attributes:

Name Type Description
target_path

The path to the output file where the video will be saved.
video_info

Information about the video resolution, fps, and total frame count.
codec

FOURCC code for video format
Example 
import supervision as sv

video_info = sv.VideoInfo.from_video_path("<SOURCE_VIDEO_PATH>")
frames_generator = sv.get_video_frames_generator("<SOURCE_VIDEO_PATH>")

with sv.VideoSink(target_path="<TARGET_VIDEO_PATH>", video_info=video_info) as sink:
    for frame in frames_generator:
        sink.write_frame(frame=frame)
Source code in src/supervision/utils/video.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
class VideoSink:
    """
    Context manager that saves video frames to a file using OpenCV.

    Attributes:
        target_path: The path to the output file where the video will be saved.
        video_info: Information about the video resolution, fps,
            and total frame count.
        codec: FOURCC code for video format

    Example:
        ```python
        import supervision as sv

        video_info = sv.VideoInfo.from_video_path("<SOURCE_VIDEO_PATH>")
        frames_generator = sv.get_video_frames_generator("<SOURCE_VIDEO_PATH>")

        with sv.VideoSink(target_path="<TARGET_VIDEO_PATH>", video_info=video_info) as sink:
            for frame in frames_generator:
                sink.write_frame(frame=frame)
        ```
    """  # noqa: E501 // docs

    def __init__(
        self, target_path: str, video_info: VideoInfo, codec: str = "mp4v"
    ) -> None:
        self.target_path = target_path
        self.video_info = video_info
        self.__codec = codec
        self.__fourcc: int = 0
        self.__writer: cv2.VideoWriter | None = None

    def __enter__(self) -> VideoSink:
        """Open the underlying video writer for context-managed frame output."""
        fourcc_fn = cast(
            Callable[[str, str, str, str], int], getattr(cv2, "VideoWriter_fourcc")
        )
        try:
            self.__fourcc = int(fourcc_fn(*self.__codec))
        except TypeError as e:
            logger.warning("%s. Defaulting to mp4v...", str(e))
            self.__fourcc = int(fourcc_fn(*"mp4v"))
        self.__writer = cv2.VideoWriter(
            self.target_path,
            self.__fourcc,
            self.video_info.fps,
            self.video_info.resolution_wh,
        )
        # OpenCV can construct a writer object that is not usable for the target path.
        if not self.__writer.isOpened():
            self.__writer.release()
            self.__writer = None
            raise RuntimeError(f"Could not open video writer for {self.target_path}")
        return self

    def write_frame(self, frame: npt.NDArray[np.uint8]) -> None:
        """
        Writes a single video frame to the target video file.

        Args:
            frame: The video frame to be written to the file. The frame
                must be in BGR color format.
        """
        # Preserve the context-manager invariant instead of silently dropping frames.
        if self.__writer is None:
            raise RuntimeError("write_frame requires an open VideoSink context.")
        self.__writer.write(frame)

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        exc_traceback: TracebackType | None,
    ) -> None:
        """Release the underlying video writer when leaving the context."""
        if self.__writer is not None:
            self.__writer.release()
            self.__writer = None

Methods:

__enter__() -> VideoSink

Open the underlying video writer for context-managed frame output.

Source code in src/supervision/utils/video.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def __enter__(self) -> VideoSink:
    """Open the underlying video writer for context-managed frame output."""
    fourcc_fn = cast(
        Callable[[str, str, str, str], int], getattr(cv2, "VideoWriter_fourcc")
    )
    try:
        self.__fourcc = int(fourcc_fn(*self.__codec))
    except TypeError as e:
        logger.warning("%s. Defaulting to mp4v...", str(e))
        self.__fourcc = int(fourcc_fn(*"mp4v"))
    self.__writer = cv2.VideoWriter(
        self.target_path,
        self.__fourcc,
        self.video_info.fps,
        self.video_info.resolution_wh,
    )
    # OpenCV can construct a writer object that is not usable for the target path.
    if not self.__writer.isOpened():
        self.__writer.release()
        self.__writer = None
        raise RuntimeError(f"Could not open video writer for {self.target_path}")
    return self

__exit__(exc_type: type[BaseException] | None, exc_value: BaseException | None, exc_traceback: TracebackType | None) -> None

Release the underlying video writer when leaving the context.

Source code in src/supervision/utils/video.py 
145
146
147
148
149
150
151
152
153
154
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    exc_traceback: TracebackType | None,
) -> None:
    """Release the underlying video writer when leaving the context."""
    if self.__writer is not None:
        self.__writer.release()
        self.__writer = None

write_frame(frame: npt.NDArray[np.uint8]) -> None

Writes a single video frame to the target video file.

Parameters:

Name Type Description Default
frame
NDArray[uint8]

The video frame to be written to the file. The frame must be in BGR color format.

 required
Source code in src/supervision/utils/video.py 
132
133
134
135
136
137
138
139
140
141
142
143
def write_frame(self, frame: npt.NDArray[np.uint8]) -> None:
    """
    Writes a single video frame to the target video file.

    Args:
        frame: The video frame to be written to the file. The frame
            must be in BGR color format.
    """
    # Preserve the context-manager invariant instead of silently dropping frames.
    if self.__writer is None:
        raise RuntimeError("write_frame requires an open VideoSink context.")
    self.__writer.write(frame)

FPSMonitor

supervision.utils.video.FPSMonitor

A class for monitoring frames per second (FPS) to benchmark latency.

Source code in src/supervision/utils/video.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
class FPSMonitor:
    """
    A class for monitoring frames per second (FPS) to benchmark latency.
    """

    def __init__(self, sample_size: int = 30) -> None:
        """
        Args:
            sample_size: The maximum number of observations for latency
                benchmarking.

        Examples:
            ```python
            import supervision as sv

            frames_generator = sv.get_video_frames_generator(
                source_path="<SOURCE_FILE_PATH>")
            fps_monitor = sv.FPSMonitor()

            for frame in frames_generator:
                # your processing code here
                fps_monitor.tick()
                fps = fps_monitor.fps
            ```
        """
        self.all_timestamps: deque[float] = deque(maxlen=sample_size)

    @property
    def fps(self) -> float:
        """
        Computes and returns the average FPS based on the stored time stamps.

        Returns:
            The average FPS across the recorded intervals. Returns 0.0 if fewer
            than two time stamps are stored.
        """
        if len(self.all_timestamps) < 2:
            return 0.0
        taken_time = self.all_timestamps[-1] - self.all_timestamps[0]
        frame_intervals = len(self.all_timestamps) - 1
        return frame_intervals / taken_time if taken_time != 0 else 0.0

    def tick(self) -> None:
        """
        Adds a new time stamp to the deque for FPS calculation.
        """
        self.all_timestamps.append(time.monotonic())

    def reset(self) -> None:
        """
        Clears all the time stamps from the deque.
        """
        self.all_timestamps.clear()

Attributes

fps: float property

Computes and returns the average FPS based on the stored time stamps.

Returns:

Type Description
float

The average FPS across the recorded intervals. Returns 0.0 if fewer
float

than two time stamps are stored.

Methods:

__init__(sample_size: int = 30) -> None

Parameters:

Name Type Description Default
sample_size
int

The maximum number of observations for latency benchmarking.

 30

Examples:

import supervision as sv

frames_generator = sv.get_video_frames_generator(
    source_path="<SOURCE_FILE_PATH>")
fps_monitor = sv.FPSMonitor()

for frame in frames_generator:
    # your processing code here
    fps_monitor.tick()
    fps = fps_monitor.fps
Source code in src/supervision/utils/video.py 
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
def __init__(self, sample_size: int = 30) -> None:
    """
    Args:
        sample_size: The maximum number of observations for latency
            benchmarking.

    Examples:
        ```python
        import supervision as sv

        frames_generator = sv.get_video_frames_generator(
            source_path="<SOURCE_FILE_PATH>")
        fps_monitor = sv.FPSMonitor()

        for frame in frames_generator:
            # your processing code here
            fps_monitor.tick()
            fps = fps_monitor.fps
        ```
    """
    self.all_timestamps: deque[float] = deque(maxlen=sample_size)

reset() -> None

Clears all the time stamps from the deque.

Source code in src/supervision/utils/video.py 
529
530
531
532
533
def reset(self) -> None:
    """
    Clears all the time stamps from the deque.
    """
    self.all_timestamps.clear()

tick() -> None

Adds a new time stamp to the deque for FPS calculation.

Source code in src/supervision/utils/video.py 
523
524
525
526
527
def tick(self) -> None:
    """
    Adds a new time stamp to the deque for FPS calculation.
    """
    self.all_timestamps.append(time.monotonic())

get_video_frames_generator

supervision.utils.video.get_video_frames_generator(source_path: str, stride: int = 1, start: int = 0, end: int | None = None, iterative_seek: bool = False) -> Generator[npt.NDArray[np.uint8], None, None]

Get a generator that yields the frames of the video.

Parameters:

Name Type Description Default

source_path

str

The path of the video file.

 required

stride

int

Indicates the interval at which frames are returned, skipping stride - 1 frames between each.

 1

start

int

Indicates the starting position from which video should generate frames

 0

end

int | None

Indicates the ending position at which video should stop generating frames. If None, video will be read to the end.

 None

iterative_seek

bool

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.

 False

Returns:

Type Description
None

A generator that yields the frames of the video.
Note

The underlying cv2.VideoCapture is always released when the generator is exhausted or closed, even if the consumer breaks out of iteration early.

Examples:

import supervision as sv

for frame in sv.get_video_frames_generator(source_path="<SOURCE_VIDEO_PATH>"):
    ...
Source code in src/supervision/utils/video.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def get_video_frames_generator(
    source_path: str,
    stride: int = 1,
    start: int = 0,
    end: int | None = None,
    iterative_seek: bool = False,
) -> Generator[npt.NDArray[np.uint8], None, None]:
    """
    Get a generator that yields the frames of the video.

    Args:
        source_path: The path of the video file.
        stride: Indicates the interval at which frames are returned,
            skipping stride - 1 frames between each.
        start: Indicates the starting position from which
            video should generate frames
        end: Indicates the ending position at which video
            should stop generating frames. If None, video will be read to the end.
        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.

    Returns:
        A generator that yields the
            frames of the video.

    Note:
        The underlying `cv2.VideoCapture` is always released when the generator
        is exhausted or closed, even if the consumer breaks out of iteration early.

    Examples:
        ```python
        import supervision as sv

        for frame in sv.get_video_frames_generator(source_path="<SOURCE_VIDEO_PATH>"):
            ...
        ```
    """
    video, start, end = _validate_and_setup_video(
        source_path, start, end, iterative_seek
    )
    frame_position = start
    try:
        while True:
            success, frame = video.read()
            if not success or frame_position >= end:
                break
            if frame is not None:
                yield cast(npt.NDArray[np.uint8], frame)
            for _ in range(stride - 1):
                success = video.grab()
                if not success:
                    break
            frame_position += stride
    finally:
        video.release()

process_video

supervision.utils.video.process_video(source_path: str, target_path: str, callback: Callable[[npt.NDArray[np.uint8], int], npt.NDArray[np.uint8]], *, max_frames: int | None = None, prefetch: int = 32, writer_buffer: int = 32, show_progress: bool = False, progress_message: str = 'Processing video', preserve_audio: bool = False) -> None

Process video frames asynchronously using a threaded pipeline.

This function orchestrates a three-stage pipeline to optimize video processing throughput:

  1. Reader thread: Continuously reads frames from the source video file and enqueues them into a bounded queue (frame_read_queue). The queue size is limited by the prefetch parameter to control memory usage.
  2. Main thread (Processor): Dequeues frames from frame_read_queue, applies the user-defined callback function to process each frame, then enqueues the processed frames into another bounded queue (frame_write_queue) for writing. The processing happens in the main thread, simplifying use of stateful objects without synchronization.
  3. Writer thread: Dequeues processed frames from frame_write_queue and writes them sequentially to the output video file.

Parameters:

Name Type Description Default

source_path

str

Path to the input video file.

 required

target_path

str

Path where the processed video will be saved.

 required

callback

Callable[[NDArray[uint8], int], NDArray[uint8]]

Function called for each frame, accepting the frame as a numpy array and its zero-based index, returning the processed frame.

 required

max_frames

int | None

Optional maximum number of frames to process. If None, the entire video is processed (default).

 None

prefetch

int

Maximum number of frames buffered by the reader thread. Controls memory use; default is 32.

 32

writer_buffer

int

Maximum number of frames buffered before writing. Controls output buffer size; default is 32.

 32

show_progress

bool

Whether to display a tqdm progress bar during processing. Default is False.

 False

progress_message

str

Description shown in the progress bar.

 'Processing video'

preserve_audio

bool

If True, copy the audio stream from source_path into target_path after frame processing. Requires ffmpeg on PATH (e.g. apt install ffmpeg, brew install ffmpeg). If ffmpeg is not found or the mux step fails, a warning is logged and the output video is saved without audio — no exception is raised. Audio is truncated to match the processed video duration. Default is False.

 False

Returns:

Type Description
None

None
Example 
import supervision as sv
from rfdetr import RFDETRMedium

model = RFDETRMedium()

def callback(frame, frame_index):
    return model.predict(frame)

sv.process_video(
    source_path="source.mp4",
    target_path="target.mp4",
    callback=callback,
    preserve_audio=True,
)
Source code in src/supervision/utils/video.py 
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
def process_video(
    source_path: str,
    target_path: str,
    callback: Callable[[npt.NDArray[np.uint8], int], npt.NDArray[np.uint8]],
    *,
    max_frames: int | None = None,
    prefetch: int = 32,
    writer_buffer: int = 32,
    show_progress: bool = False,
    progress_message: str = "Processing video",
    preserve_audio: bool = False,
) -> None:
    """
    Process video frames asynchronously using a threaded pipeline.

    This function orchestrates a three-stage pipeline to optimize video processing
    throughput:

    1. Reader thread: Continuously reads frames from the source video file and
       enqueues them into a bounded queue (`frame_read_queue`). The queue size is
       limited by the `prefetch` parameter to control memory usage.
    2. Main thread (Processor): Dequeues frames from `frame_read_queue`, applies the
       user-defined `callback` function to process each frame, then enqueues the
       processed frames into another bounded queue (`frame_write_queue`) for writing.
       The processing happens in the main thread, simplifying use of stateful objects
       without synchronization.
    3. Writer thread: Dequeues processed frames from `frame_write_queue` and writes
       them sequentially to the output video file.

    Args:
        source_path: Path to the input video file.
        target_path: Path where the processed video will be saved.
        callback: Function called for
            each frame, accepting the frame as a numpy array and its zero-based index,
            returning the processed frame.
        max_frames: Optional maximum number of frames to process.
            If None, the entire video is processed (default).
        prefetch: Maximum number of frames buffered by the reader thread.
            Controls memory use; default is 32.
        writer_buffer: Maximum number of frames buffered before writing.
            Controls output buffer size; default is 32.
        show_progress: Whether to display a tqdm progress bar during processing.
            Default is False.
        progress_message: Description shown in the progress bar.
        preserve_audio: If True, copy the audio stream from `source_path` into
            `target_path` after frame processing. Requires `ffmpeg` on PATH
            (e.g. `apt install ffmpeg`, `brew install ffmpeg`). If ffmpeg is
            not found or the mux step fails, a warning is logged and the output
            video is saved without audio — no exception is raised. Audio is
            truncated to match the processed video duration. Default is False.

    Returns:
        None

    Example:
        ```python
        import supervision as sv
        from rfdetr import RFDETRMedium

        model = RFDETRMedium()

        def callback(frame, frame_index):
            return model.predict(frame)

        sv.process_video(
            source_path="source.mp4",
            target_path="target.mp4",
            callback=callback,
            preserve_audio=True,
        )
        ```
    """
    video_info = VideoInfo.from_video_path(video_path=source_path)
    total_frames = (
        min(video_info.total_frames or 0, max_frames)
        if max_frames is not None
        else video_info.total_frames or 0
    )

    frame_read_queue: Queue[tuple[int, npt.NDArray[np.uint8]] | None] = Queue(
        maxsize=prefetch
    )
    frame_write_queue: Queue[npt.NDArray[np.uint8] | None] = Queue(
        maxsize=writer_buffer
    )

    def reader_thread() -> None:
        frame_generator = get_video_frames_generator(
            source_path=source_path,
            end=max_frames,
        )
        for frame_index, frame in enumerate(frame_generator):
            frame_read_queue.put((frame_index, frame))
        frame_read_queue.put(None)

    def writer_thread(video_sink: VideoSink) -> None:
        while True:
            frame = frame_write_queue.get()
            if frame is None:
                break
            video_sink.write_frame(frame=frame)

    reader_worker = threading.Thread(target=reader_thread, daemon=True)
    with VideoSink(target_path=target_path, video_info=video_info) as video_sink:
        writer_worker = threading.Thread(
            target=writer_thread,
            args=(video_sink,),
            daemon=True,
        )

        reader_worker.start()
        writer_worker.start()

        progress_bar = tqdm(
            total=total_frames,
            disable=not show_progress,
            desc=progress_message,
        )

        exception_in_worker: Exception | None = None
        read_finished = False

        try:
            while True:
                read_item = frame_read_queue.get()
                if read_item is None:
                    read_finished = True
                    break

                frame_index, frame = read_item
                try:
                    processed_frame = callback(frame, frame_index)
                    frame_write_queue.put(processed_frame)
                    progress_bar.update(1)
                except Exception as exc:
                    exception_in_worker = exc
                    break
        finally:
            try:
                frame_write_queue.put(None, timeout=1)
            except Full:
                # Best effort: if the writer is stuck and the queue never drains,
                # do not block shutdown forever trying to enqueue the sentinel.
                pass
            if not read_finished:
                while True:
                    # Use timeout to prevent indefinite blocking if reader thread fails
                    try:
                        read_item = frame_read_queue.get(timeout=1)
                        if read_item is None:
                            break
                    # If we timeout waiting for a frame, only assume failure if reader
                    # thread is no longer alive. Otherwise, keep waiting as the reader
                    # may simply be slow (for example, due to a slow source).
                    except Empty:
                        if not reader_worker.is_alive():
                            break
                        # Reader is still alive; continue waiting for frames.
                        continue
            reader_worker.join(timeout=10)
            writer_worker.join(timeout=10)
            progress_bar.close()
            if exception_in_worker is not None:
                raise exception_in_worker

    if preserve_audio:
        if writer_worker.is_alive():
            logger.warning(
                "Writer thread did not finish in time; skipping audio mux "
                "to avoid reading an incomplete output file."
            )
        else:
            _mux_audio(source_path=source_path, video_path=target_path)

Comments