diff --git a/Video_Demo_notebook.ipynb b/Video_Demo_notebook.ipynb
new file mode 100644
index 000000000..3031c3b71
--- /dev/null
+++ b/Video_Demo_notebook.ipynb
@@ -0,0 +1,297 @@
+{
+    "cells": [
+        {
+            "cell_type": "code",
+            "execution_count": 1,
+            "id": "fb864ed3",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "TqdmWarning: IProgress not found. Please update jupyter and ipywidgets. See https://ipywidgets.readthedocs.io/en/stable/user_install.html\n"
+                    ]
+                }
+            ],
+            "source": [
+                "from tqdm import tqdm\n",
+                "\n",
+                "import supervision as sv"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 2,
+            "id": "94cbdffa",
+            "metadata": {},
+            "outputs": [
+                {
+                    "data": {
+                        "text/plain": [
+                            "<Video gemist.mp4 : VideoInfo(width=848, height=480, fps=15, precise_fps=14.955430835259161, total_frames=151) : Backend: opencv>"
+                        ]
+                    },
+                    "execution_count": 2,
+                    "metadata": {},
+                    "output_type": "execute_result"
+                }
+            ],
+            "source": [
+                "video = sv.Video(\"gemist.mp4\")\n",
+                "video"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 3,
+            "id": "3cd5bbd6",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "100%|\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588| 151/151 [00:00<00:00, 1929.48it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "video = sv.Video(\"gemist.mp4\")\n",
+                "for frame in tqdm(video):\n",
+                "    pass"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 4,
+            "id": "5ca4d68c",
+            "metadata": {},
+            "outputs": [],
+            "source": [
+                "for frame in sv.Video(\"gemist.mp4\").frames(\n",
+                "    stride=5, start=100, end=500, resolution_wh=(1280, 720)\n",
+                "):\n",
+                "    pass"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 5,
+            "id": "90463868",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "Saving video: 151it [00:00, 540.72it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "import cv2\n",
+                "\n",
+                "\n",
+                "def blur(frame, i):\n",
+                "    return cv2.GaussianBlur(frame, (11, 11), 0)\n",
+                "\n",
+                "\n",
+                "sv.Video(\"gemist.mp4\").save(\"blurred.mp4\", callback=blur, show_progress=True)"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 6,
+            "id": "47382afa",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "Saving video: 151it [00:00, 539.70it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "sv.Video(\"gemist.mp4\").save(\n",
+                "    \"timelapse.mp4\", fps=60, callback=lambda f, i: f, show_progress=True\n",
+                ")"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 7,
+            "id": "11bbe111",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "Saving video: 151it [00:00, 625.05it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "from supervision import Video, VideoInfo\n",
+                "\n",
+                "# Need more elaboration on this\n",
+                "\n",
+                "source = Video(\"gemist.mp4\")\n",
+                "target_info = VideoInfo(width=200, height=600, fps=200)\n",
+                "\n",
+                "source.save(\"resized.mp4\", info=target_info, show_progress=True)"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 8,
+            "id": "f60323ab",
+            "metadata": {},
+            "outputs": [
+                {
+                    "data": {
+                        "text/plain": [
+                            "<Video gemist.mp4 : VideoInfo(width=848, height=480, fps=15, precise_fps=14.955430835259161, total_frames=151) : Backend: pyav>"
+                        ]
+                    },
+                    "execution_count": 8,
+                    "metadata": {},
+                    "output_type": "execute_result"
+                }
+            ],
+            "source": [
+                "import supervision as sv\n",
+                "\n",
+                "video = sv.Video(\"gemist.mp4\", backend=\"pyav\")\n",
+                "video"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 9,
+            "id": "de97b6c3",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "100%|\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588| 151/151 [00:00<00:00, 1969.45it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "from tqdm import tqdm\n",
+                "\n",
+                "import supervision as sv\n",
+                "\n",
+                "video = sv.Video(\"gemist.mp4\", backend=\"pyav\")\n",
+                "for frame in tqdm(video):\n",
+                "    pass"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 10,
+            "id": "726ffd1a",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "13it [00:00, 318.78it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "from tqdm import tqdm\n",
+                "\n",
+                "import supervision as sv\n",
+                "\n",
+                "for frame in tqdm(\n",
+                "    sv.Video(\"gemist.mp4\", backend=\"pyav\").frames(\n",
+                "        stride=5, start=100, end=500, resolution_wh=(1280, 720)\n",
+                "    )\n",
+                "):\n",
+                "    pass"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 11,
+            "id": "0da30fa8",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "Saving video: 151it [00:00, 413.05it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "def blur(frame, i):\n",
+                "    return cv2.GaussianBlur(frame, (11, 11), 0)\n",
+                "\n",
+                "\n",
+                "sv.Video(\"gemist.mp4\", backend=\"pyav\").save(\n",
+                "    \"blurred_av.mp4\", callback=blur, show_progress=True\n",
+                ")"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": 12,
+            "id": "ef5fb50e",
+            "metadata": {},
+            "outputs": [
+                {
+                    "name": "stderr",
+                    "output_type": "stream",
+                    "text": [
+                        "Saving video: 151it [00:00, 413.50it/s]\n"
+                    ]
+                }
+            ],
+            "source": [
+                "sv.Video(\"gemist.mp4\", backend=\"pyav\").save(\n",
+                "    \"timelapse.mp4\", fps=60, callback=lambda f, i: f, show_progress=True\n",
+                ")"
+            ]
+        },
+        {
+            "cell_type": "code",
+            "execution_count": null,
+            "id": "a8db8606",
+            "metadata": {},
+            "outputs": [],
+            "source": []
+        }
+    ],
+    "metadata": {
+        "kernelspec": {
+            "display_name": ".venv",
+            "language": "python",
+            "name": "python3"
+        },
+        "language_info": {
+            "codemirror_mode": {
+                "name": "ipython",
+                "version": 3
+            },
+            "file_extension": ".py",
+            "mimetype": "text/x-python",
+            "name": "python",
+            "nbconvert_exporter": "python",
+            "pygments_lexer": "ipython3",
+            "version": "3.11.13"
+        }
+    },
+    "nbformat": 4,
+    "nbformat_minor": 5
+}
diff --git a/pyproject.toml b/pyproject.toml
index 9bf3b24aa..29a231030 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -83,6 +83,9 @@ build = [
     "wheel>=0.40,<0.46",
     "build>=0.10,<1.3"
 ]
+video = [
+    "av>=15.0.0"
+]
 
 [tool.bandit]
 target = ["test", "supervision"]
diff --git a/supervision/__init__.py b/supervision/__init__.py
index 04d3fb254..3eba1208d 100644
--- a/supervision/__init__.py
+++ b/supervision/__init__.py
@@ -131,11 +131,11 @@
 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.core import Video, VideoInfo
 
 __all__ = [
     "LMM",
@@ -193,6 +193,7 @@
     "TriangleAnnotator",
     "VertexAnnotator",
     "VertexLabelAnnotator",
+    "Video",
     "VideoInfo",
     "VideoSink",
     "approximate_polygon",
diff --git a/supervision/utils/video.py b/supervision/utils/video.py
index 3b281b4e2..23759bffc 100644
--- a/supervision/utils/video.py
+++ b/supervision/utils/video.py
@@ -9,6 +9,8 @@
 import numpy as np
 from tqdm.auto import tqdm
 
+from supervision.utils.internal import deprecated
+
 
 @dataclass
 class VideoInfo:
@@ -60,6 +62,10 @@ def resolution_wh(self) -> tuple[int, int]:
         return self.width, self.height
 
 
+@deprecated(
+    "VideoSink is deprated and will be removed in a future version. "
+    "Use `sv.Video(path).save(...)` instead."
+)
 class VideoSink:
     """
     Context manager that saves video frames to a file using OpenCV.
@@ -141,6 +147,10 @@ def _validate_and_setup_video(
     return video, start, end
 
 
+@deprecated(
+    "get_video_frames_generator is deprated and will be removed in a future version. "
+    "Use `sv.Video(source).frames(...)` instead."
+)
 def get_video_frames_generator(
     source_path: str,
     stride: int = 1,
@@ -192,6 +202,10 @@ def get_video_frames_generator(
     video.release()
 
 
+@deprecated(
+    "get_video_frames_generator is deprated and will be removed in a future version."
+    "Use `sv.Video(source).save(target, callback=...)` instead."
+)
 def process_video(
     source_path: str,
     target_path: str,
diff --git a/supervision/video/__init__.py b/supervision/video/__init__.py
new file mode 100644
index 000000000..445a57c74
--- /dev/null
+++ b/supervision/video/__init__.py
@@ -0,0 +1,4 @@
+from .core import Video
+from .utils import VideoInfo
+
+__all__ = ["Video", "VideoInfo"]
diff --git a/supervision/video/backends/__init__.py b/supervision/video/backends/__init__.py
new file mode 100644
index 000000000..d834269c1
--- /dev/null
+++ b/supervision/video/backends/__init__.py
@@ -0,0 +1,7 @@
+BACKENDS = {
+    "opencv": "supervision.video.backends.opencv",
+    "pyav": "supervision.video.backends.pyav",
+}
+
+
+__all__ = ["BACKENDS"]
diff --git a/supervision/video/backends/base.py b/supervision/video/backends/base.py
new file mode 100644
index 000000000..7e8a5f4b9
--- /dev/null
+++ b/supervision/video/backends/base.py
@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Protocol, runtime_checkable
+
+import numpy as np
+
+from ..utils import VideoInfo
+
+
+@runtime_checkable
+class Backend(Protocol):
+    """Protocol for video backends used by ``supervision.video.Video``.
+
+    The high-level adapter instantiates a backend selected by name and only
+    calls the methods defined in this protocol. Other members are considered
+    private implementation details.
+    """
+
+    def __init__(self, source: str | int):
+        """Create a new backend for a source.
+
+        Args:
+            source: Either a path/URL (``str``) or a webcam index (``int``).
+        """
+
+    def info(self) -> VideoInfo:
+        """Return static information about the source (size, fps, frames).
+
+        Returns:
+            VideoInfo: Static properties of the video source.
+        """
+
+    def read(self) -> tuple[bool, np.ndarray]:
+        """Decode the next frame.
+
+        Returns:
+            tuple[bool, np.ndarray]: ``(success, frame)`` where frame is a
+            ``np.ndarray`` (H x W x 3) in backend-specific channel order.
+        """
+
+    def grab(self) -> bool:
+        """Grab the next frame without decoding pixels.
+
+        Useful to skip frames quickly (for example, when ``stride > 1``).
+
+        Returns:
+            bool: ``True`` if a frame position was advanced, ``False`` otherwise.
+        """
+
+    def seek(self, frame_idx: int) -> None:
+        """Seek so that the next call to ``read`` returns ``frame_idx``."""
+
+    def writer(
+        self,
+        path: str,
+        info: VideoInfo,
+        codec: str | None = None,
+    ) -> Writer:
+        """Return a writer that encodes frames to ``path``.
+
+        Args:
+            path: Target file path.
+            info: Expected output resolution and fps (copied from source by default).
+            codec: FourCC or codec name to override the backend default.
+
+        Returns:
+            Writer: A context-manager compatible writer instance.
+        """
+
+    def __iter__(self) -> Iterator[np.ndarray]:
+        """Yield successive frames until exhaustion.
+
+        This is considered convenience behavior; the default implementation is
+        sufficient for most backends.
+
+        Yields:
+            np.ndarray: The next decoded frame.
+        """
+
+
+@runtime_checkable
+class Writer(Protocol):
+    """Protocol for encoded video writers returned by ``Backend.writer``."""
+
+    def write(self, frame: np.ndarray, frame_number: int, callback) -> None:
+        """Write a single frame to the output stream."""
+
+    def close(self) -> None:
+        """Flush and close the underlying container or file descriptor."""
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False  # propagate exception (if any)
+
+
+class _NullWriter:
+    """Fallback writer that silently drops every frame."""
+
+    def write(self, frame: np.ndarray, frame_number: int, callback) -> None:
+        pass
+
+    def close(self) -> None:
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+__all__ = [
+    "Backend",
+    "Writer",
+]
diff --git a/supervision/video/backends/opencv.py b/supervision/video/backends/opencv.py
new file mode 100644
index 000000000..2169f1ecf
--- /dev/null
+++ b/supervision/video/backends/opencv.py
@@ -0,0 +1,203 @@
+from collections.abc import Callable, Iterator
+from typing import Any
+
+import cv2
+import numpy as np
+
+from ..utils import VideoInfo
+from .base import Writer
+
+
+class OpenCVWriter:
+    def __init__(self, vw: cv2.VideoWriter, info: VideoInfo):
+        """OpenCV-based video writer.
+
+        Args:
+            vw: An initialized ``cv2.VideoWriter`` instance.
+            info: Output video information used to validate/resize frames.
+        """
+        self._vw = vw
+        self.info = info
+
+    def write(
+        self,
+        frame: np.ndarray,
+        frame_number: int,
+        callback: Callable[[np.ndarray], None] | None = None,
+    ) -> None:
+        """Write a frame, applying an optional callback and resize if needed.
+
+        Args:
+            frame: Input frame array.
+            frame_number: Sequential frame number being written.
+            callback: Optional function ``(frame, frame_number) -> frame`` to
+                transform the frame before writing.
+        """
+        if callback:
+            frame = callback(frame, frame_number)
+        if frame.shape[0] != self.info.height or frame.shape[1] != self.info.width:
+            frame = cv2.resize(frame, (self.info.width, self.info.height))
+        self._vw.write(frame)
+
+    def close(self) -> None:
+        """Release the underlying ``cv2.VideoWriter``."""
+        self._vw.release()
+
+    def __enter__(self) -> Writer:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.close()
+
+
+class OpenCVBackend:
+    def __init__(self, source_path: str | int):
+        """Create a new backend for a source path or webcam index.
+
+        Args:
+            source_path: File path or stream URL (``str``) or webcam index
+                (``int``).
+        """
+        self.source_path = source_path
+        self.cap = cv2.VideoCapture(self.source_path)
+        if not self.cap.isOpened():
+            raise ValueError(f"Could not open video source {self.source_path}")
+
+    def info(self) -> VideoInfo:
+        """Return static information (width, height, fps, total_frames)."""
+        from ..core import VideoInfo
+
+        w = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        h = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+        precise_fps = self.cap.get(cv2.CAP_PROP_FPS)
+        fps = int(round(precise_fps, 0))
+        n = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        return VideoInfo(w, h, fps, precise_fps, n)
+
+    def read(self) -> tuple[bool, np.ndarray]:
+        """Decode the next frame."""
+        return self.cap.read()
+
+    def grab(self) -> bool:
+        """Grab the next frame without decoding pixels."""
+        return self.cap.grab()
+
+    def seek(self, frame_idx: int) -> None:
+        """Seek so that the next call to ``read`` returns ``frame_idx``."""
+        self.cap.set(cv2.CAP_PROP_POS_FRAMES, frame_idx)
+
+    def writer(self, path: str, info: VideoInfo, codec: str | None = None) -> Writer:
+        """Return a writer that encodes frames to a file path.
+
+        Args:
+            path: Target file path.
+            info: Expected output resolution and fps (copied from source by default).
+            codec: FourCC or codec name to override the backend default.
+        """
+        fourcc = (
+            cv2.VideoWriter_fourcc(*codec) if codec else cv2.VideoWriter_fourcc(*"mp4v")
+        )
+        vw = cv2.VideoWriter(path, fourcc, info.fps, (info.width, info.height))
+        return OpenCVWriter(vw, info)
+
+    def frames(
+        self,
+        stride: int = 1,
+        start: int = 0,
+        end: int | None = None,
+        resolution_wh: tuple[int, int] | None = None,
+        interpolation=cv2.INTER_LINEAR,
+    ) -> Iterator[np.ndarray]:
+        """Yield frames lazily, with optional skipping and resizing.
+
+        Args:
+            stride: Number of frames to skip between yielded frames (``1`` yields every frame).
+            start: First frame index (0-based) to yield.
+            end: Index after the last frame to yield. ``None`` means until exhaustion.
+            resolution_wh: Optional ``(width, height)`` to resize each yielded frame to.
+            interpolation: OpenCV interpolation flag used when resizing.
+
+        Yields:
+            np.ndarray: The next decoded (and optionally resized) video frame.
+        """
+        if stride < 1:
+            raise ValueError("stride must be >= 1")
+
+        info = self.info()
+        total = (
+            info.total_frames if info.total_frames and info.total_frames > 0 else None
+        )
+        if end is None and total is not None:
+            end = total
+        if start < 0 or start >= end:
+            return
+
+        # Position capture at the start frame
+        self.seek(start)
+        current_idx = start
+        infinate_stream = end is None
+
+        while infinate_stream or current_idx < end:
+            success, frame = self.read()
+            if not success:
+                break
+
+            if resolution_wh is not None and (
+                frame.shape[1] != resolution_wh[0] or frame.shape[0] != resolution_wh[1]
+            ):
+                frame = cv2.resize(frame, resolution_wh, interpolation=interpolation)
+
+            yield frame
+            current_idx += 1
+
+            # Efficiently skip stride-1 frames with grab()
+            skip = stride - 1
+            while skip and current_idx < end:
+                grabbed = self.grab()
+                if not grabbed:
+                    return
+                current_idx += 1
+                skip -= 1
+
+    def __iter__(self) -> Iterator[np.ndarray]:
+        """Yield successive frames until exhaustion.
+
+        This is considered convenience behavior; the default implementation is
+        sufficient for most backends.
+        """
+        while True:
+            success, frame = self.read()
+            if not success:
+                break
+            yield frame
+
+    def release(self):
+        """Release the video file."""
+        self.cap.release()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.release()
+
+    def __len__(self) -> int:
+        n = self.info().total_frames
+        if n is None or n < 0:
+            raise TypeError("length is unknown for this stream")
+        return n
+
+    def __getitem__(self, index: int) -> np.ndarray:
+        current = int(self.cap.get(cv2.CAP_PROP_POS_FRAMES))
+        self.cap.set(cv2.CAP_PROP_POS_FRAMES, index)
+        success, frame = self.read()
+        self.cap.set(
+            cv2.CAP_PROP_POS_FRAMES, current
+        )  # ? Do we want to restore the video to the original position?
+        if not success:
+            raise IndexError(f"Failed to read frame {index}")
+        return frame
+
+
+# Provide a consistent alias for the core loader
+Backend = OpenCVBackend
diff --git a/supervision/video/backends/pyav.py b/supervision/video/backends/pyav.py
new file mode 100644
index 000000000..100809558
--- /dev/null
+++ b/supervision/video/backends/pyav.py
@@ -0,0 +1,335 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from fractions import Fraction
+from typing import Any, cast
+
+import numpy as np
+
+try:
+    import av
+except ImportError as e:
+    raise ImportError(
+        "The pyav backend is not installed, "
+        "please install it using `pip install supervision[ffmpeg]`"
+    ) from e
+
+from supervision.video.backends.base import Writer
+from supervision.video.utils import VideoInfo
+
+
+class PyAVWriter:
+    def __init__(
+        self,
+        container: av.container.output.OutputContainer,
+        stream: av.video.stream.VideoStream,
+        info: VideoInfo,
+    ):
+        """PyAV based video writer.
+
+        Args:
+            container: An opened ``av.open(..., mode="w")`` output container.
+            stream: The created output video stream.
+            info: Output video information used to validate or resize frames.
+        """
+        self._container = container
+        self._stream = stream
+        self.info = info
+
+    def write(
+        self,
+        frame: np.ndarray,
+        frame_number: int,
+        callback: Any = None,
+    ) -> None:
+        """Encode a frame, applying an optional callback and resize if needed.
+
+        Args:
+            frame: Input frame array, expected BGR with shape (H, W, 3).
+            frame_number: Sequential frame number being written.
+            callback: Optional function ``(frame, frame_number) -> frame`` that
+                transforms the frame before writing.
+        """
+        if callback is not None:
+            frame = callback(frame, frame_number)
+
+        # Create a VideoFrame from the ndarray. We assume BGR input for parity with OpenCV.
+        vframe = av.VideoFrame.from_ndarray(frame, format="bgr24")
+
+        # Ensure expected output dimensions and pixel format for the encoder.
+        target_fmt = self._stream.pix_fmt or "yuv420p"
+        if (
+            vframe.width != self.info.width
+            or vframe.height != self.info.height
+            or vframe.format.name != target_fmt
+        ):
+            vframe = vframe.reformat(
+                width=self.info.width, height=self.info.height, format=target_fmt
+            )
+
+        # Encode and mux packets.
+        packets = self._stream.encode(vframe)
+        for packet in packets:
+            self._container.mux(packet)
+
+    def close(self) -> None:
+        """Flush and close the underlying output container."""
+        # Flush encoder
+        packets = self._stream.encode(None)
+        for packet in packets:
+            self._container.mux(packet)
+        self._container.close()
+
+    def __enter__(self) -> Writer:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        self.close()
+
+
+class PyAVBackend:
+    def __init__(self, source_path: str | int):
+        """Create a new backend for a source path or stream URL.
+
+        Args:
+            source_path: File path or stream URL. Integer webcam indexes are not
+                handled portably by PyAV, so pass a device URL if you need a camera.
+        """
+        self.source_path = source_path
+        if isinstance(source_path, int):
+            raise ValueError(
+                "Numeric webcam indexes are not supported by this PyAV backend. "
+                "Provide a device URL, or use the OpenCV backend for webcams."
+            )
+
+        try:
+            self.container: av.container.input.InputContainer = av.open(
+                self.source_path, mode="r"
+            )
+        except Exception as e:
+            raise ValueError(
+                f"Could not open video source {self.source_path!r}: {e}"
+            ) from e
+
+        # Pick the first video stream.
+        video_streams = [s for s in self.container.streams if s.type == "video"]
+        if not video_streams:
+            self.container.close()
+            raise ValueError(f"No video stream found in source {self.source_path!r}")
+
+        self.video_stream: av.video.stream.VideoStream = cast(
+            av.video.stream.VideoStream, video_streams[0]
+        )
+        self.video_stream.thread_type = "AUTO"  # Improve performance on some inputs
+        self._decoder = self.container.decode(video=self.video_stream.index)
+
+    def _compute_fps(self) -> tuple[int, float]:
+        avg = self.video_stream.average_rate
+        if avg is None:
+            avg = self.video_stream.guessed_rate
+        if avg is None:
+            precise = 30.0  # As a last resort, fall back to 30 fps
+        else:
+            # average_rate is a Fraction
+            precise = float(avg)
+        fps_int = round(precise) if precise > 0 else 30
+        return fps_int, precise
+
+    def info(self) -> VideoInfo:
+        """Return static information (width, height, fps, precise_fps, total_frames)."""
+        w = int(self.video_stream.codec_context.width or self.video_stream.width)
+        h = int(self.video_stream.codec_context.height or self.video_stream.height)
+        fps_int, precise = self._compute_fps()
+        n = (
+            int(self.video_stream.frames)
+            if self.video_stream.frames not in (None, 0)
+            else None
+        )
+        return VideoInfo(w, h, fps_int, precise, n)
+
+    def _reset_decoder(self) -> None:
+        self._decoder = self.container.decode(video=self.video_stream.index)
+
+    def _next_avframe(self) -> av.VideoFrame | None:
+        try:
+            return next(self._decoder)
+        except StopIteration:
+            return None
+
+    def read(self) -> tuple[bool, np.ndarray]:
+        """Decode the next frame as a BGR numpy array."""
+        frame = self._next_avframe()
+        if frame is None:
+            return False, np.empty((0, 0, 3), dtype=np.uint8)
+        arr = frame.to_ndarray(format="bgr24")
+        return True, arr
+
+    def grab(self) -> bool:
+        """Advance to the next frame without materializing pixel data."""
+        return self._next_avframe() is not None
+
+    def _pts_from_frame_index(self, frame_idx: int) -> int:
+        fps_int, precise = self._compute_fps()
+        time_base: Fraction = self.video_stream.time_base or Fraction(
+            1, max(fps_int, 1)
+        )
+        # seconds to PTS units
+        seconds = frame_idx / (precise if precise > 0 else max(fps_int, 1))
+        pts = round(seconds / float(time_base))
+        return pts
+
+    def seek(self, frame_idx: int) -> None:
+        """Seek so that the next call to ``read`` returns ``frame_idx``.
+
+        This computes a timestamp from the frame index using the stream frame rate,
+        performs an accurate seek on the video stream, then resets the decoder.
+        """
+        pts = self._pts_from_frame_index(frame_idx)
+        # Use any_frame=False to seek to keyframes, then decoder will advance
+        self.container.seek(
+            pts, stream=self.video_stream, any_frame=False, backward=True
+        )
+        self._reset_decoder()
+
+    # ----------- Writer factory -----------
+    def writer(self, path: str, info: VideoInfo, codec: str | None = None) -> Writer:
+        """Return a writer that encodes frames to a file path.
+
+        Args:
+            path: Target file path.
+            info: Expected output resolution and fps.
+            codec: FFmpeg encoder name. Examples: "libx264", "h264", "hevc", "mpeg4".
+        """
+        out = av.open(path, mode="w")
+        enc = cast(
+            av.video.stream.VideoStream,
+            out.add_stream(codec or "libx264", rate=info.fps),
+        )
+        enc.width = info.width
+        enc.height = info.height
+        enc.pix_fmt = "yuv420p"  # Use a broadly compatible pixel format
+        enc.options = {
+            "movflags": "+faststart"
+        }  # Improve default compatibility for MP4
+        return cast(Writer, PyAVWriter(out, enc, info))
+
+    def frames(
+        self,
+        stride: int = 1,
+        start: int = 0,
+        end: int | None = None,
+        resolution_wh: tuple[int, int] | None = None,
+        _interpolation: Any = None,  # Kept for API parity. PyAV scales internally.
+    ) -> Iterator[np.ndarray]:
+        """Yield frames lazily with optional skipping and resizing.
+
+        Args:
+            stride: Number of frames to skip between yielded frames. One yields every frame.
+            start: First frame index to yield.
+            end: Index after the last frame to yield. None means until exhaustion.
+            resolution_wh: Optional (width, height) to resize each yielded frame to.
+            interpolation: Ignored. Present only for API parity with the OpenCV backend.
+        Yields:
+            np.ndarray: The next decoded and optionally resized frame in BGR order.
+        """
+        if stride < 1:
+            raise ValueError("stride must be >= 1")
+
+        info = self.info()
+        total = (
+            info.total_frames if info.total_frames and info.total_frames > 0 else None
+        )
+        if end is None and total is not None:
+            end = total
+        if end is not None and (start < 0 or start >= end):
+            return
+
+        # Position decoder at the start frame
+        self.seek(start)
+        current_idx = start
+        infinite_stream = end is None
+
+        while infinite_stream or current_idx < end:  # type: ignore[operator]
+            vf = self._next_avframe()
+            if vf is None:
+                break
+
+            if resolution_wh is not None and (
+                vf.width != resolution_wh[0] or vf.height != resolution_wh[1]
+            ):
+                vf = vf.reformat(
+                    width=resolution_wh[0], height=resolution_wh[1], format="bgr24"
+                )
+                arr = vf.to_ndarray(format="bgr24")
+            else:
+                arr = vf.to_ndarray(format="bgr24")
+
+            yield arr
+            current_idx += 1
+
+            # Efficiently skip stride - 1 frames
+            skip = stride - 1
+            while skip > 0 and (
+                infinite_stream or (end is not None and current_idx < end)
+            ):
+                if not self.grab():
+                    return
+                current_idx += 1
+                skip -= 1
+
+    def __iter__(self) -> Iterator[np.ndarray]:
+        """Yield successive frames until exhaustion."""
+        while True:
+            ok, frame = self.read()
+            if not ok:
+                break
+            yield frame
+
+    # ----------- Resource management -----------
+    def release(self) -> None:
+        """Close the input container."""
+        try:
+            self.container.close()
+        finally:
+            self._decoder = iter(())
+
+    def __enter__(self) -> PyAVBackend:
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.release()
+
+    # ----------- Random access helpers -----------
+    def __len__(self) -> int:
+        n = self.info().total_frames
+        if n is None or n < 0:
+            raise TypeError("length is unknown for this stream")
+        return n
+
+    def __getitem__(self, index: int) -> np.ndarray:
+        """Return the frame at the given index without disturbing current decode state."""
+        # Open a temporary container to avoid changing the main decoder position
+        try:
+            with av.open(self.source_path, mode="r") as tmp:
+                vstreams = [s for s in tmp.streams if s.type == "video"]
+                if not vstreams:
+                    raise IndexError(f"No video stream in source {self.source_path!r}")
+                vs = vstreams[0]
+
+                # Compute PTS and seek
+                avg = vs.average_rate or vs.guessed_rate or Fraction(30, 1)
+                time_base: Fraction = vs.time_base or Fraction(1, int(avg))
+                seconds = index / float(avg)
+                pts = round(seconds / float(time_base))
+                tmp.seek(pts, stream=vs, any_frame=False, backward=True)
+
+                for frame in tmp.decode(video=vs.index):
+                    return frame.to_ndarray(format="bgr24")
+        except Exception as e:
+            raise IndexError(f"Failed to read frame {index}: {e}") from e
+
+        raise IndexError(f"Failed to read frame {index}")
+
+
+# Provide a consistent alias for the core loader
+Backend = PyAVBackend
diff --git a/supervision/video/core.py b/supervision/video/core.py
new file mode 100644
index 000000000..78abcfa4e
--- /dev/null
+++ b/supervision/video/core.py
@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import importlib
+from collections.abc import Generator
+from dataclasses import replace
+
+import numpy as np
+from tqdm.auto import tqdm
+
+from .utils import VideoInfo
+
+
+class Video:
+    def __init__(self, video_path: str, backend: str | None = None):
+        """High-level video reader and writer.
+
+        This class provides a unified interface over pluggable backends for
+        reading frames, iterating, slicing, and saving videos.
+
+        Args:
+            video_path: Path or identifier of the source video. This can be a
+                file path, URL, or camera index depending on the backend.
+            backend: Optional backend name (for example, ``"pyav"``). If not
+                provided, ``"opencv"`` is used.
+        """
+        self.video_path = video_path
+        self._backend_name = backend or "opencv"
+        self._backend = self.__get_backend()
+
+    def __len__(self) -> int:
+        """Return the number of frames if known.
+
+        Returns:
+            int: The total number of frames.
+
+        Raises:
+            TypeError: If the underlying stream does not expose a finite length.
+        """
+        return len(self._backend)
+
+    def __iter__(self):
+        """Return an iterator over decoded frames as ``np.ndarray``.
+
+        Yields:
+            np.ndarray: The next decoded frame in BGR/RGB format depending on
+                the backend.
+        """
+        return iter(self._backend)
+
+    def __getitem__(self, index: int) -> np.ndarray:
+        """Return the frame at a specific index.
+
+        Args:
+            index: Zero-based frame index to retrieve.
+
+        Returns:
+            np.ndarray: The decoded frame at ``index``.
+
+        Raises:
+            IndexError: If the index is out of bounds or cannot be read.
+        """
+        return self._backend[index]
+
+    def __repr__(self) -> str:
+        """Return a concise representation including path and info."""
+        return (
+            f"<Video {self.video_path} : {self.info} : Backend: {self._backend_name}>"
+        )
+
+    def __get_backend(self):
+        """Instantiate the selected backend implementation.
+
+        Returns:
+            Backend: The instantiated backend object matching ``self._backend_name``.
+
+        Raises:
+            ValueError: If an unknown backend name is provided.
+        """
+        from .backends import BACKENDS
+
+        try:
+            module_path = BACKENDS[self._backend_name]
+        except KeyError as exc:
+            raise ValueError(
+                f"Unknown backend '{self._backend_name}'. "
+                f"Available backends: {', '.join(BACKENDS.keys())}"
+            ) from exc
+        module = importlib.import_module(module_path)
+        self._backend = module.Backend(str(self.video_path))
+        return self._backend
+
+    @property
+    def info(self) -> VideoInfo:
+        """Return static information about the video source.
+
+        Returns:
+            VideoInfo: Width, height, frames per second, and total frames when
+            available.
+        """
+        return self._backend.info()
+
+    def frames(
+        self,
+        stride: int = 1,
+        start: int = 0,
+        end: int | None = None,
+        resolution_wh: tuple[int, int] | None = None,
+    ) -> Generator[np.ndarray]:
+        """Yield frames lazily with optional skipping and resizing.
+
+        Args:
+            stride: Number of frames to skip between yielded frames (``1`` yields
+                every frame).
+            start: First frame index (0-based) to yield.
+            end: Index after the last frame to yield. ``None`` means until
+                exhaustion.
+            resolution_wh: Optional ``(width, height)`` to resize each yielded
+                frame to.
+
+        Yields:
+            np.ndarray: The next decoded (and optionally resized) video frame.
+        """
+        yield from self._backend.frames(stride, start, end, resolution_wh)
+
+    def save(
+        self,
+        path: str,
+        callback=None,
+        show_progress=True,
+        info: VideoInfo | None = None,
+        **kwargs,
+    ):
+        """Encode and save frames from this video to a new file.
+
+        Args:
+            path: Output file path.
+            callback: Optional callable to transform frames before writing. It
+                receives ``(frame, frame_number)`` and should return a frame.
+            show_progress: If ``True``, display a progress bar while saving.
+            info: Optional ``VideoInfo`` describing the desired output
+                properties. If omitted, the source video info is used.
+            **kwargs: Additional ``VideoInfo`` fields to override (for example,
+                ``fps``, ``width``, ``height``). Only keys present in
+                ``VideoInfo`` are applied.
+
+        Returns:
+            None
+        """
+        updated_info = info or self.info
+        updated_info = replace(
+            updated_info,
+            **{k: v for k, v in kwargs.items() if k in self.info.__dataclass_fields__},
+        )
+        with self._backend.writer(path, updated_info) as writer:
+            for i, frame in enumerate(
+                tqdm(self.frames(), desc="Saving video", disable=not show_progress)
+            ):
+                writer.write(frame, frame_number=i, callback=callback)
diff --git a/supervision/video/utils.py b/supervision/video/utils.py
new file mode 100644
index 000000000..21e87cf43
--- /dev/null
+++ b/supervision/video/utils.py
@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import cast
+
+
+@dataclass
+class VideoInfo:
+    """Static information about a video.
+
+    Stores the width, height, frames-per-second values, and optionally the
+    precise frames-per-second and total number of frames.
+
+    Attributes:
+        width: Width of the video in pixels.
+        height: Height of the video in pixels.
+        fps: Rounded frames per second of the video.
+        precise_fps: Exact frames per second value when available.
+        total_frames: Total number of frames in the video if known.
+
+    Examples:
+        ```python
+        import supervision as sv
+
+        video_info = sv.VideoInfo.from_video_path(video_path="/path/to/video.mp4")
+
+        print(video_info)
+        # VideoInfo(width=3840, height=2160, fps=25, precise_fps=25.0, total_frames=538)
+
+        print(video_info.resolution_wh)
+        # (3840, 2160)
+        ```
+    """
+
+    width: int
+    height: int
+    fps: int
+    precise_fps: float | None = None
+    total_frames: int | None = None
+
+    @classmethod
+    def from_video_path(cls, video_path: str, backend: str | None = None) -> VideoInfo:
+        """Construct a ``VideoInfo`` from a video file or stream.
+
+        Args:
+            video_path: Path or URL to the video file/stream.
+            backend: Optional backend name to use (for example, ``"opencv"``).
+
+        Returns:
+            VideoInfo: Parsed static information from the source.
+        """
+        from supervision.video.core import Video  # Avoid circular import
+
+        return cast("VideoInfo", Video(video_path, backend=backend).info)
+
+    @property
+    def resolution_wh(self) -> tuple[int, int]:
+        """Return the resolution as ``(width, height)``.
+
+        Returns:
+            tuple[int, int]: A tuple of width and height in pixels.
+        """
+        return self.width, self.height