pvio.video — Video Backends

Abstract base class and concrete backends for video sources.

Video is the base class for all video sources. Subclass it and implement _validate_init_params, _load_metadata, and _read_frame to add a custom backend.

pvio.video.Video

Video(
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
)

Bases: ABC

Base class for a video source (a file or a directory of images).

Nothing is loaded at construction time. Call .setup() before reading frames.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the video source. Interpretation depends on the backend subclass.	required
frame_range	tuple[int, int] | None	If specified, only frames in [start, end) are exposed. Defaults to the full video.	None

Source code in src/pvio/video.py

def __init__(
    self,
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
):
    r"""Base class for a video source (a file or a directory of images).

    Nothing is loaded at construction time. Call `.setup()` before reading frames.

    Args:
        path (Path | str): Path to the video source. Interpretation depends on
            the backend subclass.
        frame_range (tuple[int, int] | None): If specified, only frames in
            [start, end) are exposed. Defaults to the full video.
    """
    self.path = Path(path)
    self.frame_range_requested = frame_range

    self.__setup_done = False
    self.frame_range_effective: tuple[int, int] | None = None
    self.n_frames_in_source_video: int | None = None
    self.n_frames_in_range: int | None = None
    self.frame_size: tuple[int, int] | None = None
    self.fps: float | None = None

setup

setup() -> None

Validate arguments, load metadata, and prepare this video for reading.

Must be called after __init__ and before reading frames. Calling :meth:read_frame without a prior setup triggers an automatic setup with a warning.

Raises:

Type	Description
Exception	Whatever the backend's :meth:_validate_init_params, :meth:_load_metadata, or :meth:_post_setup raises on failure.

Source code in src/pvio/video.py

def setup(self) -> None:
    """Validate arguments, load metadata, and prepare this video for reading.

    Must be called after ``__init__`` and before reading frames. Calling
    :meth:`read_frame` without a prior ``setup`` triggers an automatic
    setup with a warning.

    Raises:
        Exception: Whatever the backend's :meth:`_validate_init_params`,
            :meth:`_load_metadata`, or :meth:`_post_setup` raises on failure.
    """
    if self.__setup_done:
        logger.warning(
            f"Video at path {self.path} is already set up. Skipping redundant call."
        )
        return

    # Run-time check of whether arguments to `__init__` are valid
    self._validate_init_params()

    # Load video metadata - this might take a non-negligible amount time depending
    # on the backend subclass
    n_frames_total, frame_size, fps = self._load_metadata()
    self.n_frames_in_source_video = n_frames_total
    self.frame_size = frame_size
    self.fps = fps

    # Resolve effective frame range
    resolved_range = self._resolve_effective_frame_range(
        self.frame_range_requested, n_frames_total
    )
    self.frame_range_effective = resolved_range
    self.n_frames_in_range = resolved_range[1] - resolved_range[0]

    # Post-setup hook for backend subclasses; raises on failure.
    self._post_setup()

    self.__setup_done = True

read_frame

read_frame(
    index: int, transform: Callable | None = None
) -> torch.Tensor

Read a single frame as a CHW float32 tensor.

Parameters:

Name	Type	Description	Default
index	int	Virtual frame index; 0 is the start of the effective frame range.	required
transform	Callable | None	Optional callable applied to the CHW float32 tensor before returning.	None

Returns:

Type	Description
Tensor	Frame as a CHW float32 tensor with values in [0, 1].

Source code in src/pvio/video.py

def read_frame(self, index: int, transform: Callable | None = None) -> torch.Tensor:
    """Read a single frame as a CHW float32 tensor.

    Args:
        index: Virtual frame index; 0 is the start of the effective
            frame range.
        transform: Optional callable applied to the CHW float32 tensor
            before returning.

    Returns:
        Frame as a CHW float32 tensor with values in ``[0, 1]``.
    """
    if not self.__setup_done:
        logger.warning(
            f"Video at path {self.path} is not set up yet. Call `.setup()` first. "
            f"This might take some time, depending on the backend subclass."
        )
        self.setup()

    return self._read_frame(index, transform=transform)

close

close() -> None

Release any resources held by this Video object.

Source code in src/pvio/video.py

def close(self) -> None:
    """Release any resources held by this Video object."""
    pass

_validate_init_params abstractmethod

_validate_init_params() -> None

Validate the path for this backend. Raise on any error; return None.

Source code in src/pvio/video.py

@abstractmethod
def _validate_init_params(self) -> None:
    """Validate the path for this backend. Raise on any error; return None."""
    pass

_load_metadata abstractmethod

_load_metadata() -> tuple[int, tuple[int, int], float]

Return frame count, frame size, and FPS for the video source.

Returns:

Type	Description
int	A 3-tuple (n_frames_total, (height, width), fps).
tuple[int, int]	n_frames_total is the full frame count of the source, unclipped
float	by frame_range. fps may be None if not applicable (e.g.
tuple[int, tuple[int, int], float]	for image directories).

Source code in src/pvio/video.py

@abstractmethod
def _load_metadata(self) -> tuple[int, tuple[int, int], float]:
    """Return frame count, frame size, and FPS for the video source.

    Returns:
        A 3-tuple ``(n_frames_total, (height, width), fps)``.
        *n_frames_total* is the full frame count of the source, unclipped
        by *frame_range*. *fps* may be ``None`` if not applicable (e.g.
        for image directories).
    """
    pass

_read_frame abstractmethod

_read_frame(
    index: int, transform: Callable | None = None
) -> torch.Tensor

Return the frame at virtual index index as a CHW float tensor in [0, 1]. Apply transform after loading if provided.

index 0 corresponds to the start of the effective frame range.

Source code in src/pvio/video.py

@abstractmethod
def _read_frame(
    self, index: int, transform: Callable | None = None
) -> torch.Tensor:
    """Return the frame at virtual index `index` as a CHW float tensor in [0, 1].
    Apply `transform` after loading if provided.

    `index` 0 corresponds to the start of the effective frame range."""
    pass

_post_setup

_post_setup() -> None

Optional backend-specific logic run at the end of .setup().

Raise on failure; the default is a no-op.

Source code in src/pvio/video.py

def _post_setup(self) -> None:
    """Optional backend-specific logic run at the end of `.setup()`.

    Raise on failure; the default is a no-op."""
    return

pvio.video.EncodedVideo

EncodedVideo(
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
    buffer_size: int = 64,
    cache_metadata: bool = True,
    use_cached_metadata: bool = True,
    device: str | None = None,
    num_ffmpeg_threads: int = 1,
)

Bases: Video

Video backend for "real" video files (e.g., mp4, mkv, etc.) using TorchCodec.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the video file.	required
frame_range	tuple[int, int] | None	If specified, only frames in this range [start, end) are considered part of the video.	None
buffer_size	int	Frames to decode in one batch. Larger values reduce decoding overhead at the cost of memory. Optimal size depends on keyframe intervals and available RAM; 64 (default) works well in most cases.	64
num_ffmpeg_threads	int	FFmpeg decoding threads for CPU decoding. The default of 1 is right for the DataLoader regime this class is designed for — many videos decoded in parallel across worker processes, where one thread per decode avoids oversubscribing cores. When reading a single video from a single process, pass 0 (auto) to enable frame-parallel decoding, which is several times faster. (For pure sequential access also consider :func:pvio.io.iter_frames_from_video, which avoids this class's random-access machinery entirely.) Ignored on CUDA.	1
cache_metadata	bool	Whether to cache video metadata to disk for faster subsequent metadata reads.	True
use_cached_metadata	bool	Whether to use cached metadata when available. Set to False to force re-load metadata.	True
device	str | None	Decode device for TorchCodec. None (default) or "auto" auto-selects "cuda" when a CUDA GPU is available and "cpu" otherwise; pass "cpu" or "cuda" to force a choice. Exact (frame-accurate) seeking is preserved on either device. GPU decoding returns frames already resident in GPU memory. Note that CUDA cannot be initialised inside forked DataLoader worker processes, so when this object is iterated under num_workers > 0 the decoder automatically downgrades to CPU in the workers; use the single-process path (num_workers=0, as SimpleVideoCollectionLoader arranges automatically) to keep decoding on the GPU.	None

Source code in src/pvio/video.py

def __init__(
    self,
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
    buffer_size: int = 64,
    cache_metadata: bool = True,
    use_cached_metadata: bool = True,
    device: str | None = None,
    num_ffmpeg_threads: int = 1,
):
    """Video backend for "real" video files (e.g., mp4, mkv, etc.) using TorchCodec.

    Args:
        path (Path | str): Path to the video file.
        frame_range (tuple[int, int] | None): If specified, only frames in this
            range [start, end) are considered part of the video.
        buffer_size (int): Frames to decode in one batch. Larger values reduce
            decoding overhead at the cost of memory. Optimal size depends on
            keyframe intervals and available RAM; 64 (default) works well in most
            cases.
        num_ffmpeg_threads (int): FFmpeg decoding threads for CPU decoding.
            The default of 1 is right for the DataLoader regime this class is
            designed for — many videos decoded in parallel across worker
            processes, where one thread per decode avoids oversubscribing
            cores. When reading a single video from a single process, pass
            ``0`` (auto) to enable frame-parallel decoding, which is several
            times faster. (For pure sequential access also consider
            :func:`pvio.io.iter_frames_from_video`, which avoids this class's
            random-access machinery entirely.) Ignored on CUDA.
        cache_metadata (bool): Whether to cache video metadata to disk for faster
            subsequent metadata reads.
        use_cached_metadata (bool): Whether to use cached metadata when available.
            Set to False to force re-load metadata.
        device (str | None): Decode device for TorchCodec. ``None`` (default)
            or ``"auto"`` auto-selects ``"cuda"`` when a CUDA GPU is available
            and ``"cpu"`` otherwise; pass ``"cpu"`` or ``"cuda"`` to force a
            choice. Exact
            (frame-accurate) seeking is preserved on either device. GPU
            decoding returns frames already resident in GPU memory. Note that
            CUDA cannot be initialised inside forked DataLoader worker
            processes, so when this object is iterated under
            ``num_workers > 0`` the decoder automatically downgrades to CPU in
            the workers; use the single-process path (``num_workers=0``, as
            ``SimpleVideoCollectionLoader`` arranges automatically) to keep
            decoding on the GPU.
    """
    super().__init__(path, frame_range)
    self.buffer_size = buffer_size
    self.cache_metadata = cache_metadata
    self.use_cached_metadata = use_cached_metadata
    self.device = _accel.resolve_decode_device(device)
    self.num_ffmpeg_threads = num_ffmpeg_threads

    # The following are to be managed by `.read_frame()`
    self._decoder: VideoDecoder | None = None
    self._buffer: dict[int, torch.Tensor] = {}

pvio.video.ImageDirVideo

ImageDirVideo(
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
    frame_id_regex: str
    | Pattern
    | None = "frame\\D*(\\d+)(?!\\d)",
)

Bases: Video

Video backend for directories containing frames as individual images.

Parameters:

Name	Type	Description	Default
path	Path | str	Path to the directory containing frames.	required
frame_range	tuple[int, int] | None	If specified, only frames in this range [start, end) are exposed. Note: start/end index the sorted position of the frames (0-based), not the physical frame id parsed from the filename. So with files whose parsed ids are 0, 5, 10, 15, 20, frame_range=(1, 3) exposes the 2nd and 3rd files (parsed ids 5 and 10), matching how EncodedVideo treats frame_range as a positional slice. The parsed ids are used only for ordering and remain accessible via frame_id_vir2phy / phy_frame_id_to_path.	None
frame_id_regex	str | Pattern | None	Regex used to parse the frame index from each filename (must capture exactly one group). The default handles names like "frame1.jpg", "frame002.tif", "frame_3.png", "frame-4.bmp", "frame_id=05.ext". If None, files are sorted alphabetically and indices are assigned 0, 1, 2, …	'frame\\D*(\\d+)(?!\\d)'

Source code in src/pvio/video.py

def __init__(
    self,
    path: Path | str,
    frame_range: tuple[int, int] | None = None,
    frame_id_regex: str | re.Pattern | None = r"frame\D*(\d+)(?!\d)",
):
    r"""Video backend for directories containing frames as individual images.

    Args:
        path (Path | str): Path to the directory containing frames.
        frame_range (tuple[int, int] | None): If specified, only frames in this
            range [start, end) are exposed. Note: start/end index the **sorted
            position** of the frames (0-based), not the physical frame id parsed
            from the filename. So with files whose parsed ids are 0, 5, 10, 15, 20,
            ``frame_range=(1, 3)`` exposes the 2nd and 3rd files (parsed ids 5 and
            10), matching how `EncodedVideo` treats frame_range as a positional
            slice. The parsed ids are used only for ordering and remain accessible
            via `frame_id_vir2phy` / `phy_frame_id_to_path`.
        frame_id_regex (str | re.Pattern | None): Regex used to parse the frame
            index from each filename (must capture exactly one group). The default
            handles names like "frame1.jpg", "frame002.tif", "frame_3.png",
            "frame-4.bmp", "frame_id=05.ext". If None, files are sorted
            alphabetically and indices are assigned 0, 1, 2, …
    """
    super().__init__(path, frame_range)
    if frame_id_regex is not None and isinstance(frame_id_regex, str):
        frame_id_regex = re.compile(frame_id_regex)
    self.frame_id_regex: re.Pattern | None = frame_id_regex

    # Directory listing cached by `._load_metadata()` and reused by
    # `._post_setup()` so the directory is scanned only once.
    self._image_files: list[Path] = []

    # The following mappings are to be populated in `._post_setup()`
    self.phy_frame_id_to_path: dict[int, Path] = {}
    self.vir_frame_id_to_path: dict[int, Path] = {}
    self.frame_id_vir2phy: dict[int, int] = {}
    self.frame_id_phy2vir: dict[int, int] = {}