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. An estimate from container metadata that can be zero, negative, or lower than the number of frames the video holds.

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

@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. An estimate from container metadata that can be zero,
            negative, or lower than the number of frames the video holds.

    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:
        """Read video metadata from a file path.

        Args:
            video_path: Path to the video file.

        Returns:
            A :class:`~supervision.utils.video.VideoInfo` instance with width,
                height, fps, and total_frames.

        Examples:
            ```python
            import supervision as sv

            video_info = sv.VideoInfo.from_video_path(video_path="<SOURCE_VIDEO_FILE>")
            # VideoInfo(width=3840, height=2160, fps=25.0, total_frames=538)
            ```
        """
        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

Methods:

from_video_path(video_path: str) -> VideoInfo classmethod

Read video metadata from a file path.

Parameters:

Name	Type	Description	Default
video_path	str	Path to the video file.	required

Returns:

Name	Type	Description
A	VideoInfo	class:~supervision.utils.video.VideoInfo instance with width, height, fps, and total_frames.

Examples:

import supervision as sv

video_info = sv.VideoInfo.from_video_path(video_path="<SOURCE_VIDEO_FILE>")
# VideoInfo(width=3840, height=2160, fps=25.0, total_frames=538)

Source code in src/supervision/utils/video.py

@classmethod
def from_video_path(cls, video_path: str) -> VideoInfo:
    """Read video metadata from a file path.

    Args:
        video_path: Path to the video file.

    Returns:
        A :class:`~supervision.utils.video.VideoInfo` instance with width,
            height, fps, and total_frames.

    Examples:
        ```python
        import supervision as sv

        video_info = sv.VideoInfo.from_video_path(video_path="<SOURCE_VIDEO_FILE>")
        # VideoInfo(width=3840, height=2160, fps=25.0, total_frames=538)
        ```
    """
    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)

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

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 writer without replacing an error from the managed block.

        A secondary release failure is logged; a cleanup-only failure propagates. The
        sink is marked closed in either case.
        """
        if self.__writer is not None:
            try:
                self.__writer.release()
            except Exception:
                # Context exit must preserve the exception already being propagated;
                # without a pending error, callers still need to see release failures.
                if exc_value is None:
                    raise
                logger.exception(
                    "Failed to release video writer while handling an error."
                )
            finally:
                self.__writer = None

Methods:

__enter__() -> VideoSink

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

Source code in src/supervision/utils/video.py

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 writer without replacing an error from the managed block.

A secondary release failure is logged; a cleanup-only failure propagates. The sink is marked closed in either case.

Source code in src/supervision/utils/video.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    exc_traceback: TracebackType | None,
) -> None:
    """Release the writer without replacing an error from the managed block.

    A secondary release failure is logged; a cleanup-only failure propagates. The
    sink is marked closed in either case.
    """
    if self.__writer is not None:
        try:
            self.__writer.release()
        except Exception:
            # Context exit must preserve the exception already being propagated;
            # without a pending error, callers still need to see release failures.
            if exc_value is None:
                raise
            logger.exception(
                "Failed to release video writer while handling an error."
            )
        finally:
            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

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

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

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

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

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, prefetch: int = 0) -> 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. A zero or negative frame count reported by the container means the count is unknown: end is not checked and reading stops at the stream 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
prefetch	int	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. This works best when the consumer releases the GIL during frame processing (common for numpy/PyTorch/ONNX C-extension calls). Pure Python per-frame consumers that hold the GIL (for example, heavy Python loops, PIL usage, or pandas apply) usually see little speedup. Default 0 keeps the original synchronous behaviour unchanged. Note: each buffered frame occupies width x height x 3 bytes of uncompressed memory; use :meth:~supervision.utils.video.VideoInfo.from_video_path to size appropriately.	0

Returns:

Type	Description
None	A generator that yields the frames of the video.

Raises:

Type	Description
Exception	If end exceeds a positive frame count reported by the container (raised on first iteration), or capture release fails without an earlier decoding error. Both are wrapped in RuntimeError when prefetch > 0.
ValueError	If prefetch is negative.
RuntimeError	If prefetch is greater than 0 and the background reader thread encounters a decode/open error, raised as RuntimeError(f"Reader thread raised: {item!r}") from item. Errors are drained after buffered frames, so when prefetch > 0 the consumer may yield up to prefetch additional good frames before the exception is raised. If capture release also fails after a decoding error, the cleanup failure is logged and the original decoding error is preserved.

Note

For live camera streams, use cv2.VideoCapture with an integer device index directly. :func:~supervision.utils.video.get_video_frames_generator is designed for file-based sources; cv2.VideoCapture must be released by the caller when done. This requires OpenCV to be installed — the PyAV-based fallback used when OpenCV is unavailable only supports file paths, not webcam device indexes; passing an integer source to it always leaves the capture closed (isOpened() returns False):

from supervision import _cv2 as cv2

cap = cv2.VideoCapture(0)  # 0 = default webcam
try:
    while cap.isOpened():
        success, frame = cap.read()
        if not success:
            break
        ...  # process frame
finally:
    cap.release()

Examples:

import supervision as sv

for frame in sv.get_video_frames_generator(source_path="<SOURCE_VIDEO_PATH>"):
    ...

# Prefetch frames in a background thread to overlap I/O with CPU inference:
for frame in sv.get_video_frames_generator(
    source_path="<SOURCE_VIDEO_PATH>", prefetch=8
):
    ...

Source code in src/supervision/utils/video.py

def get_video_frames_generator(
    source_path: str,
    stride: int = 1,
    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.

    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.
            A zero or negative frame count reported by the container means the
            count is unknown: `end` is not checked and reading stops at the
            stream 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.
        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.
            This works best when the consumer releases the GIL during frame
            processing (common for numpy/PyTorch/ONNX C-extension calls). Pure
            Python per-frame consumers that hold the GIL (for example, heavy
            Python loops, PIL usage, or pandas `apply`) usually see little
            speedup.
            Default 0 keeps the original synchronous behaviour unchanged. Note:
            each buffered frame occupies width x height x 3 bytes of uncompressed
            memory; use
            :meth:`~supervision.utils.video.VideoInfo.from_video_path`
            to size appropriately.

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

    Raises:
        Exception: If `end` exceeds a positive frame count reported by the
            container (raised on first iteration), or capture release fails without
            an earlier decoding error. Both are wrapped in `RuntimeError` when
            `prefetch` > 0.
        ValueError: If `prefetch` is negative.
        RuntimeError: If `prefetch` is greater than 0 and the background reader
            thread encounters a decode/open error, raised as
            `RuntimeError(f"Reader thread raised: {item!r}") from item`. Errors are
            drained after buffered frames, so when `prefetch` > 0 the consumer may
            yield up to `prefetch` additional good frames before the exception is
            raised. If capture release also fails after a decoding error, the
            cleanup failure is logged and the original decoding error is preserved.

    Note:
        For live camera streams, use `cv2.VideoCapture` with an integer device
        index directly.
        :func:`~supervision.utils.video.get_video_frames_generator` is designed
        for file-based
        sources; `cv2.VideoCapture` must be released by the caller when done.
        This requires OpenCV to be installed — the PyAV-based fallback used
        when OpenCV is unavailable only supports file paths, not webcam device
        indexes; passing an integer source to it always leaves the capture
        closed (`isOpened()` returns `False`):

        ```python
        from supervision import _cv2 as cv2

        cap = cv2.VideoCapture(0)  # 0 = default webcam
        try:
            while cap.isOpened():
                success, frame = cap.read()
                if not success:
                    break
                ...  # process frame
        finally:
            cap.release()
        ```

    Examples:
        ```python
        import supervision as sv

        for frame in sv.get_video_frames_generator(source_path="<SOURCE_VIDEO_PATH>"):
            ...

        # Prefetch frames in a background thread to overlap I/O with CPU inference:
        for frame in sv.get_video_frames_generator(
            source_path="<SOURCE_VIDEO_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,
            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
    )
    frame_position = start
    decoding_failed = False
    try:
        while True:
            success, frame = video.read()
            if not success or (end is not None and 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
    except Exception:
        decoding_failed = True
        raise
    finally:
        try:
            video.release()
        except Exception:
            # A secondary release error must not replace the pending decode error;
            # cleanup-only failures still propagate, including during generator close.
            if not decoding_failed:
                raise
            logger.exception("Failed to release video capture after a decoding error.")

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. Shutdown waits for any active write to finish before releasing the output sink.

On a processing or writing failure, shutdown stops the reader instead of decoding the rest of the source. An active decode is allowed to finish before the capture is released; queued writes finish unless the writer itself fails. If a callback fails, a secondary reader failure is logged before the original callback exception is re-raised.

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). A value larger than the number of frames in the video processes the whole video.	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. Remuxing is done with PyAV; no external ffmpeg executable is required. If 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

Raises:

Type	Description
RuntimeError	If the reader thread fails to open or decode the source video, raised as RuntimeError(f"Reader thread raised: {exc!r}") from the original exception. If writing a processed frame to target_path fails, raised as RuntimeError(f"Writer thread raised: {exc!r}") from the original exception. Exceptions raised by callback are re-raised unchanged.
TypeError	If callback returns None instead of the processed frame.

Example

import supervision as sv
from rfdetr import RFDETRMedium

model = RFDETRMedium()
box_annotator = sv.BoxAnnotator()

def callback(frame, frame_index):
    # Frames are BGR; RF-DETR expects RGB. Return the annotated frame.
    detections = model.predict(frame[:, :, ::-1])
    return box_annotator.annotate(frame.copy(), detections)

sv.process_video(
    source_path="source.mp4",
    target_path="target.mp4",
    callback=callback,
    preserve_audio=True,
)

Source code in src/supervision/utils/video.py

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. Shutdown waits for any active
       write to finish before releasing the output sink.

    On a processing or writing failure, shutdown stops the reader instead of
    decoding the rest of the source. An active decode is allowed to finish before
    the capture is released; queued writes finish unless the writer itself fails.
    If a callback fails, a secondary reader failure is logged before the original
    callback exception is re-raised.

    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). A value larger
            than the number of frames in the video processes the whole video.
        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. Remuxing is done with PyAV;
            no external `ffmpeg` executable is required. If 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

    Raises:
        RuntimeError: If the reader thread fails to open or decode the source
            video, raised as `RuntimeError(f"Reader thread raised: {exc!r}")`
            from the original exception. If writing a processed frame to
            `target_path` fails, raised as
            `RuntimeError(f"Writer thread raised: {exc!r}")` from the original
            exception. Exceptions raised by `callback` are re-raised unchanged.
        TypeError: If `callback` returns `None` instead of the processed frame.

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

        model = RFDETRMedium()
        box_annotator = sv.BoxAnnotator()

        def callback(frame, frame_index):
            # Frames are BGR; RF-DETR expects RGB. Return the annotated frame.
            detections = model.predict(frame[:, :, ::-1])
            return box_annotator.annotate(frame.copy(), detections)

        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)
    video_total_frames = video_info.total_frames or 0
    total_frames = video_total_frames
    if max_frames is not None:
        total_frames = min(video_total_frames, max_frames)

    # `max_frames` is a cap, not an exact count: a value larger than the video
    # must process the whole video. `get_video_frames_generator` raises when
    # `end` exceeds the frame count, so clamp it whenever the count is known.
    frames_end: int | None = max_frames
    if max_frames is not None and video_total_frames > 0:
        frames_end = total_frames

    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
    )

    reader_exception: Exception | None = None
    writer_exception: Exception | None = None
    stop_reader = threading.Event()

    def reader_thread() -> None:
        """Feed frames into the bounded queue until EOF, failure or cancellation."""
        nonlocal reader_exception
        frame_generator = None
        try:
            frame_generator = get_video_frames_generator(
                source_path=source_path,
                end=frames_end,
            )
            frame_index = 0
            while not stop_reader.is_set():
                try:
                    frame = next(frame_generator)
                except StopIteration:
                    break
                while not stop_reader.is_set():
                    try:
                        frame_read_queue.put((frame_index, frame), timeout=0.1)
                        break
                    except Full:
                        pass
                frame_index += 1
        except Exception as exc:
            # The main loop blocks on `frame_read_queue.get()` with no timeout,
            # so a reader failure must still enqueue the sentinel or the call
            # never returns. Surface the error once the pipeline has shut down.
            reader_exception = exc
        finally:
            if frame_generator is not None:
                try:
                    frame_generator.close()
                except Exception as exc:
                    if reader_exception is None:
                        reader_exception = exc
            # A cancelled consumer no longer drains the queue. Both frame and EOF
            # puts must therefore be interruptible, even when the queue is full.
            while not stop_reader.is_set():
                try:
                    frame_read_queue.put(None, timeout=0.1)
                    break
                except Full:
                    pass

    def writer_thread(video_sink: VideoSink) -> None:
        """Write processed frames until the sentinel, keeping the first write error.

        The main loop puts frames into the bounded write queue without a timeout, so
        after a failed write this thread keeps taking frames off the queue and discards
        them; otherwise a full queue blocks the main loop forever. The error is raised
        once the pipeline has shut down.
        """
        nonlocal writer_exception
        while True:
            frame = frame_write_queue.get()
            if frame is None:
                break
            if writer_exception is not None:
                continue
            try:
                video_sink.write_frame(frame=frame)
            except Exception as exc:
                writer_exception = exc

    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: BaseException | None = None
        try:
            while True:
                read_item = frame_read_queue.get()
                if read_item is None:
                    break

                frame_index, frame = read_item
                try:
                    processed_frame = callback(frame, frame_index)
                    # `None` is the writer's end-of-stream sentinel; queueing it
                    # would stop the writer while frames keep arriving.
                    if processed_frame is None:
                        raise TypeError(
                            f"`callback` returned None for frame {frame_index}; "
                            "it must return the processed frame."
                        )
                    frame_write_queue.put(processed_frame)
                    progress_bar.update(1)
                except BaseException as exc:
                    exception_in_worker = exc
                    break
                if writer_exception is not None:
                    break
        finally:
            stop_reader.set()
            frame_write_queue.put(None)
            reader_worker.join()
            writer_worker.join()
            progress_bar.close()
            if exception_in_worker is not None:
                # Callback errors stay primary, but secondary reader failures
                # must remain visible after both worker threads have stopped.
                if reader_exception is not None:
                    logger.error(
                        "Reader thread failed while handling a processing error.",
                        exc_info=(
                            type(reader_exception),
                            reader_exception,
                            reader_exception.__traceback__,
                        ),
                    )
                raise exception_in_worker
            if reader_exception is not None:
                raise RuntimeError(
                    f"Reader thread raised: {reader_exception!r}"
                ) from reader_exception
            if writer_exception is not None:
                raise RuntimeError(
                    f"Writer thread raised: {writer_exception!r}"
                ) from writer_exception

    if preserve_audio:
        _mux_audio(source_path=source_path, video_path=target_path)

Comments