FEATURE: New Video API

pyproject.toml

@@ -46,7 +46,7 @@ dependencies = [
     "pillow>=9.4",
     "requests>=2.26.0",
     "tqdm>=4.62.3",
-    "opencv-python>=4.5.5.64"
+    "opencv-python>=4.5.5.64",
 ]
 
 [project.urls]
@@ -58,6 +58,8 @@ Documentation = "https://supervision.roboflow.com/latest/"
 metrics = [
     "pandas>=2.0.0",
 ]
+ffmpeg = ["av (>=15.0.0)"]
+rich_display = ["ipython (>=8.15,<9.0)"]
 
 [dependency-groups]
 dev = [

supervision/__init__.py

@@ -130,11 +130,12 @@
 from supervision.utils.notebook import plot_image, plot_images_grid
 from supervision.utils.video import (
     FPSMonitor,
-    VideoInfo,
     VideoSink,
     get_video_frames_generator,
     process_video,
 )
+from supervision.video import Video, VideoInfo
+from supervision.video.backend import VideoBackendType
 
 __all__ = [
     "LMM",
@@ -192,6 +193,8 @@
     "TriangleAnnotator",
     "VertexAnnotator",
     "VertexLabelAnnotator",
+    "Video",
+    "VideoBackendType",
     "VideoInfo",
     "VideoSink",
     "approximate_polygon",

supervision/utils/video.py

@@ -9,7 +9,13 @@
 import numpy as np
 from tqdm.auto import tqdm
 
+from supervision.utils.internal import deprecated
 
+
+@deprecated(
+    "`process_video` is deprecated and will be removed in "
+    "`supervision-0.32.0`. Use `sv.VideoInfo` instead."
+)
 @dataclass
 class VideoInfo:
     """
@@ -60,6 +66,10 @@ def resolution_wh(self) -> tuple[int, int]:
         return self.width, self.height
 
 
+@deprecated(
+    "`process_video` is deprecated and will be removed in "
+    "`supervision-0.32.0`. Use `sv.Video().save` instead."
+)
 class VideoSink:
     """
     Context manager that saves video frames to a file using OpenCV.
@@ -141,6 +151,10 @@ def _validate_and_setup_video(
     return video, start, end
 
 
+@deprecated(
+    "`process_video` is deprecated and will be removed in "
+    "`supervision-0.32.0`. Use `sv.Video().frame()` or `sv.Video()` instead."
+)
 def get_video_frames_generator(
     source_path: str,
     stride: int = 1,
@@ -192,6 +206,10 @@ def get_video_frames_generator(
     video.release()
 
 
+@deprecated(
+    "`process_video` is deprecated and will be removed in "
+    "`supervision-0.32.0`. Use `sv.Video().save` instead."
+)
 def process_video(
     source_path: str,
     target_path: str,

supervision/video/__init__.py

@@ -0,0 +1,4 @@
+from supervision.video.core import Video
+from supervision.video.utils import SourceType, VideoInfo
+
+__all__ = ["SourceType", "Video", "VideoInfo"]

supervision/video/backend/__init__.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Literal, Union
+
+from supervision.video.backend.opencv import OpenCVBackend, OpenCVWriter
+from supervision.video.backend.pyav import pyAVBackend, pyAVWriter
+
+VideoBackendTypes = Union[OpenCVBackend, pyAVBackend]
+VideoWriterTypes = Union[OpenCVWriter, pyAVWriter]
+
+
+class VideoBackendType(Enum):
+    """
+    Enumeration of Backends.
+
+    Attributes:
+        PYAV (str): PyAV backend (powered by FFmpeg, supports audio rendering)
+        OPENCV (str): OpenCV backend
+
+    """
+
+    PYAV = "pyav"
+    OPENCV = "opencv"
+
+    @classmethod
+    def list(cls):
+        return list(map(lambda c: c.value, cls))
+
+    @classmethod
+    def from_value(cls, value: VideoBackendType | str) -> VideoBackendType:
+        if isinstance(value, cls):
+            return value
+        if isinstance(value, str):
+            value = value.lower()
+            try:
+                return cls(value)
+            except ValueError:
+                raise ValueError(f"Invalid value: {value}. Must be one of {cls.list()}")
+        raise ValueError(
+            f"Invalid value type: {type(value)}. Must be an instance of "
+            f"{cls.__name__} or str."
+        )
+
+
+VideoBackendDict = {
+    VideoBackendType.PYAV: pyAVBackend,
+    VideoBackendType.OPENCV: OpenCVBackend,
+}
+
+VideoWriterDict = {
+    VideoBackendType.PYAV: pyAVWriter,
+    VideoBackendType.OPENCV: OpenCVWriter,
+}
+
+__all__ = [
+    "VideoBackendDict",
+    "VideoBackendType",
+    "VideoBackendTypes",
+    "VideoWriterDict",
+    "VideoWriterTypes",
+]

supervision/video/backend/base.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+import numpy as np
+
+from supervision.video.utils import VideoInfo
+
+
+class BaseBackend(ABC):
+    def __init__(self):
+        self.cap = None
+        self.video_info: VideoInfo | None = None
+        self.writer: BaseWriter | None = None
+        self.path = None
+
+    @abstractmethod
+    def open(self, path: str | int) -> None:
+        pass
+
+    @abstractmethod
+    def isOpened(self) -> bool:
+        pass
+
+    @abstractmethod
+    def _set_video_info(self) -> VideoInfo:
+        pass
+
+    @abstractmethod
+    def info(self) -> VideoInfo:
+        pass
+
+    @abstractmethod
+    def read(self) -> tuple[bool, np.ndarray]:
+        pass
+
+    @abstractmethod
+    def grab(self) -> bool:
+        pass
+
+    @abstractmethod
+    def seek(self, frame_idx: int) -> None:
+        pass
+
+    @abstractmethod
+    def release(self) -> None:
+        pass
+
+
+class BaseWriter(ABC):
+    @abstractmethod
+    def __init__(
+        self,
+        filename: str,
+        fps: int,
+        frame_size: tuple[int, int],
+        codec: str | None = None,
+        backend: BaseBackend | None = None,
+        render_audio: bool | None = None,
+    ):
+        pass
+
+    @abstractmethod
+    def __enter__(self):
+        pass
+
+    @abstractmethod
+    def __exit__(self, exc_type, exc_value, traceback):
+        pass
+
+    @abstractmethod
+    def write(self, frame: np.ndarray) -> None:
+        pass
+
+    @abstractmethod
+    def close(self) -> None:
+        pass