pvio.io — Video I/O

Low-level functions for reading and writing video files using imageio/FFmpeg.

pvio.io

VideoMetadata

Bases: NamedTuple

Basic video metadata returned by :func:get_video_metadata.

A lightweight typed record (tuple-compatible) with three fields:

Attributes:

Name	Type	Description
n_frames	int	Total frame count.
frame_size	tuple[int, int]	(height, width) in pixels.
fps	float | None	Frames per second, or None if unavailable (e.g. image directories).

read_frames_from_video

read_frames_from_video(
    video_path: Path | str,
    frame_indices: list[int] | None = None,
) -> tuple[list[np.ndarray], float | None]

Read specific frames from a video file.

Frames are decoded with PyAV at the source's native bit depth and channel layout: an 8-bit RGB video yields uint8 (H, W, 3) arrays, a 16-bit source (e.g. lossless FFV1) yields uint16 arrays, a source with an alpha channel yields 4-channel (H, W, 4) arrays, and a grayscale source yields 2-D (H, W) arrays. This preserves the real pixel values; decoding such sources to 8-bit RGB (as imageio's reader does) would discard the low byte of every 16-bit sample and collapse small values to near-zero.

Parameters:

Name	Type	Description	Default
video_path	Path | str	Path to the video file.	required
frame_indices	list[int] | None	Frame indices to read, in the order returned (duplicates allowed). If None, reads all frames in order.	None

Returns:

Type	Description
list[ndarray]	A 2-tuple (frames, fps). frames is a list of numpy arrays whose
float | None	dtype and channel count match the source (see above). fps is the
tuple[list[ndarray], float | None]	average frame rate reported by the container, or None if unavailable.

Raises:

Type	Description
IndexError	If a requested frame index is out of range.

Source code in src/pvio/io.py

def read_frames_from_video(
    video_path: Path | str, frame_indices: list[int] | None = None
) -> tuple[list[np.ndarray], float | None]:
    """Read specific frames from a video file.

    Frames are decoded with PyAV at the source's **native bit depth and channel
    layout**: an 8-bit RGB video yields uint8 ``(H, W, 3)`` arrays, a 16-bit
    source (e.g. lossless FFV1) yields uint16 arrays, a source with an alpha
    channel yields 4-channel ``(H, W, 4)`` arrays, and a grayscale source yields
    2-D ``(H, W)`` arrays. This preserves the real pixel values; decoding such
    sources to 8-bit RGB (as imageio's reader does) would discard the low byte of
    every 16-bit sample and collapse small values to near-zero.

    Args:
        video_path: Path to the video file.
        frame_indices: Frame indices to read, in the order returned (duplicates
            allowed). If ``None``, reads all frames in order.

    Returns:
        A 2-tuple ``(frames, fps)``. *frames* is a list of numpy arrays whose
        dtype and channel count match the source (see above). *fps* is the
        average frame rate reported by the container, or ``None`` if unavailable.

    Raises:
        IndexError: If a requested frame index is out of range.
    """
    with av.open(str(video_path)) as container:
        stream = container.streams.video[0]
        rate = stream.average_rate
        fps = float(rate) if rate else None

        if frame_indices is not None and len(frame_indices) == 0:
            return [], fps

        # Fast path: specific frames + a usable frame rate. Seek to each frame
        # (or decode forward between nearby ones) instead of decoding the whole
        # stream; the strategy adapts to the codec's seek granularity.
        indexed_attempted = frame_indices is not None and rate
        if indexed_attempted:
            result = _read_indexed_frames(
                container, stream, frame_indices, rate, stream.time_base
            )
            if result is not None:
                return result, fps

        # General path: a single forward decode, collecting wanted frames (or all
        # frames when frame_indices is None). Also the fallback when frames lack
        # timestamps. Frame-parallel decoding is only safe to request when the
        # codec hasn't already been opened by _read_indexed_frames above (a
        # thread_type change once open raises) -- see _iter_sequential_frames.
        target = None if frame_indices is None else set(frame_indices)
        collected: dict[int, np.ndarray] = {}
        sequential = _iter_sequential_frames(
            container, stream, enable_frame_threading=not indexed_attempted
        )
        for i, arr in enumerate(sequential):
            if target is None or i in target:
                collected[i] = arr
                if target is not None and len(collected) == len(target):
                    break

    if frame_indices is None:
        return [collected[i] for i in sorted(collected)], fps
    missing = [i for i in frame_indices if i not in collected]
    if missing:
        raise IndexError(f"frame indices {sorted(set(missing))} are out of range")
    return [collected[i] for i in frame_indices], fps

iter_frames_from_video

iter_frames_from_video(
    video_path: Path | str,
) -> Iterator[np.ndarray]

Yield every frame of a video in order, without buffering the whole file.

Same per-frame output contract as :func:read_frames_from_video (native bit depth and channel layout: uint8/uint16, with alpha or grayscale preserved), but frames are decoded and yielded one at a time instead of being collected into a list. Use this for large sequential-only workloads (e.g. piping straight into an encoder, or streaming feature extraction) where holding every decoded frame in memory at once is unnecessary or infeasible -- a 6000-frame 16-bit RGBA video, for example, is ~28 GB fully buffered.

Use :func:read_frames_from_video instead when random access (specific frame indices) or the full in-memory list is actually needed.

Parameters:

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

Yields:

Type	Description
ndarray	One numpy array per frame, in decode order, dtype/shape matching the
ndarray	source as described above.

Source code in src/pvio/io.py

def iter_frames_from_video(video_path: Path | str) -> Iterator[np.ndarray]:
    """Yield every frame of a video in order, without buffering the whole file.

    Same per-frame output contract as :func:`read_frames_from_video` (native bit
    depth and channel layout: uint8/uint16, with alpha or grayscale preserved),
    but frames are decoded and yielded one at a time instead of being collected
    into a list. Use this for large sequential-only workloads (e.g. piping
    straight into an encoder, or streaming feature extraction) where holding
    every decoded frame in memory at once is unnecessary or infeasible -- a
    6000-frame 16-bit RGBA video, for example, is ~28 GB fully buffered.

    Use :func:`read_frames_from_video` instead when random access (specific
    frame indices) or the full in-memory list is actually needed.

    Args:
        video_path: Path to the video file.

    Yields:
        One numpy array per frame, in decode order, dtype/shape matching the
        source as described above.
    """
    with av.open(str(video_path)) as container:
        stream = container.streams.video[0]
        yield from _iter_sequential_frames(
            container, stream, enable_frame_threading=True
        )

write_frames_to_video

write_frames_to_video(
    video_path: Path | str,
    frames: list[ndarray],
    fps: float,
    *,
    mode: str = "auto",
    quality: int = _accel.DEFAULT_QUALITY,
    preset: str | None = None,
    extra_ffmpeg_params: list[str] | None = None,
    log_interval: int | None = None,
    quiet: bool = False,
) -> None

Write a sequence of frames to an H.264 MP4 file.

The encoder is chosen by mode: "gpu" encodes on the GPU (H.264 NVENC), "cpu" on the CPU (libx264), and "auto" (default) uses the GPU when a CUDA device and NVENC-capable FFmpeg are available, else the CPU. Either way the encode always falls back to libx264 if NVENC fails (e.g. frames below NVENC's minimum size), so output is always produced.

Quality is controlled by the single quality knob, applied as libx264's CRF on the CPU path and as NVENC's constant QP on the GPU path — both on the same 0-51 H.264 quantiser scale where lower means higher quality and larger files. The default of 20 is visually lossless and conservative, suitable for scientific data.

Parameters:

Name	Type	Description	Default
video_path	Path | str	Path for the output video file.	required
frames	list[ndarray]	Frames as uint8 numpy arrays in (H, W, C) format. All frames must share the same spatial dimensions.	required
fps	float	Frames per second of the output video.	required
mode	str	Encoder selection — "auto" (default), "gpu", or "cpu". "gpu" forces the NVENC path (falling back to libx264 if NVENC is unavailable for the input); "cpu" forces libx264; "auto" picks the GPU when available.	'auto'
quality	int	Encode quality on the 0-51 H.264 quantiser scale (lower = higher quality, larger files). Applied as libx264's CRF or NVENC's QP depending on the chosen encoder, so it behaves consistently across the CPU and GPU paths.	DEFAULT_QUALITY
preset	str | None	Encoder preset. None uses a sensible per-encoder default ("slow" for libx264, "p7" for NVENC). If given, it is passed to whichever encoder runs — use encoder-appropriate values (libx264: ultrafast…placebo; NVENC: p1…p7).	None
extra_ffmpeg_params	list[str] | None	Optional raw FFmpeg parameters appended after the quality/preset flags, as an escape hatch for advanced options.	None
log_interval	int | None	If set, log progress every log_interval frames at INFO level.	None
quiet	bool	Suppress the encoder-parameters log line and the progress bar. When True, log_interval is also ignored.	False

Raises:

Type	Description
ValueError	If frames is empty, contains frames with mismatched dimensions, or mode is not one of "auto"/"gpu"/"cpu".

Source code in src/pvio/io.py

def write_frames_to_video(
    video_path: Path | str,
    frames: list[np.ndarray],
    fps: float,
    *,
    mode: str = "auto",
    quality: int = _accel.DEFAULT_QUALITY,
    preset: str | None = None,
    extra_ffmpeg_params: list[str] | None = None,
    log_interval: int | None = None,
    quiet: bool = False,
) -> None:
    """Write a sequence of frames to an H.264 MP4 file.

    The encoder is chosen by *mode*: ``"gpu"`` encodes on the GPU (H.264 NVENC),
    ``"cpu"`` on the CPU (libx264), and ``"auto"`` (default) uses the GPU when a
    CUDA device and NVENC-capable FFmpeg are available, else the CPU. Either way
    the encode always falls back to libx264 if NVENC fails (e.g. frames below
    NVENC's minimum size), so output is always produced.

    Quality is controlled by the single *quality* knob, applied as libx264's CRF
    on the CPU path and as NVENC's constant QP on the GPU path — both on the same
    0-51 H.264 quantiser scale where lower means higher quality and larger files.
    The default of 20 is visually lossless and conservative, suitable for
    scientific data.

    Args:
        video_path: Path for the output video file.
        frames: Frames as uint8 numpy arrays in ``(H, W, C)`` format. All
            frames must share the same spatial dimensions.
        fps: Frames per second of the output video.
        mode: Encoder selection — ``"auto"`` (default), ``"gpu"``, or ``"cpu"``.
            ``"gpu"`` forces the NVENC path (falling back to libx264 if NVENC is
            unavailable for the input); ``"cpu"`` forces libx264; ``"auto"``
            picks the GPU when available.
        quality: Encode quality on the 0-51 H.264 quantiser scale (lower = higher
            quality, larger files). Applied as libx264's CRF or NVENC's QP
            depending on the chosen encoder, so it behaves consistently across
            the CPU and GPU paths.
        preset: Encoder preset. ``None`` uses a sensible per-encoder default
            (``"slow"`` for libx264, ``"p7"`` for NVENC). If given, it is passed
            to whichever encoder runs — use encoder-appropriate values
            (libx264: ``ultrafast``…``placebo``; NVENC: ``p1``…``p7``).
        extra_ffmpeg_params: Optional raw FFmpeg parameters appended after the
            quality/preset flags, as an escape hatch for advanced options.
        log_interval: If set, log progress every *log_interval* frames at
            ``INFO`` level.
        quiet: Suppress the encoder-parameters log line and the progress bar.
            When ``True``, *log_interval* is also ignored.

    Raises:
        ValueError: If *frames* is empty, contains frames with mismatched
            dimensions, or *mode* is not one of ``"auto"``/``"gpu"``/``"cpu"``.
    """
    if mode not in ("auto", "gpu", "cpu"):
        raise ValueError(f"mode must be 'auto', 'gpu', or 'cpu', got {mode!r}.")

    # Check frame size consistency
    if len(frames) == 0:
        raise ValueError("No frames provided to write_frames_to_video")
    frame_size = frames[0].shape[:2]
    for frame in frames:
        if frame.shape[:2] != frame_size:
            raise ValueError(
                "All frames must have the same dimensions. The 0th frame has size "
                f"{frame_size}, but at least one frame has size {frame.shape[:2]}."
            )
    height, width = frame_size[0], frame_size[1]

    _encode_with_fallback(
        video_path,
        frames,
        len(frames),
        height,
        width,
        fps,
        mode=mode,
        quality=quality,
        preset=preset,
        extra_ffmpeg_params=extra_ffmpeg_params,
        log_interval=log_interval,
        quiet=quiet,
    )

write_image_paths_to_video

write_image_paths_to_video(
    video_path: Path | str,
    image_paths: list[Path | str],
    fps: float,
    *,
    mode: str = "auto",
    quality: int = _accel.DEFAULT_QUALITY,
    preset: str | None = None,
    extra_ffmpeg_params: list[str] | None = None,
    log_interval: int | None = None,
    quiet: bool = False,
) -> None

Combine on-disk image files into an H.264 MP4 file.

Like :func:write_frames_to_video, but the frames are given as paths to image files (PNG, JPEG, TIFF, …) instead of in-memory numpy arrays. Images are read lazily, one at a time, so an arbitrarily long sequence can be encoded without holding every frame in memory at once. The encoder selection, quality semantics, and NVENC→libx264 fallback are identical to :func:write_frames_to_video.

Frames are encoded in the order given by image_paths; sort the paths beforehand if a particular ordering is required. The first image's spatial dimensions define the output size, and every subsequent image must match it.

Parameters:

Name	Type	Description	Default
video_path	Path | str	Path for the output video file.	required
image_paths	list[Path | str]	Paths to the image files to combine, in output order. Each file is read with imageio and must decode to an (H, W, C) array; all images must share the same spatial dimensions.	required
fps	float	Frames per second of the output video.	required
mode	str	Encoder selection — "auto" (default), "gpu", or "cpu". See :func:write_frames_to_video for details.	'auto'
quality	int	Encode quality on the 0-51 H.264 quantiser scale (lower = higher quality, larger files). See :func:write_frames_to_video.	DEFAULT_QUALITY
preset	str | None	Encoder preset. None uses a sensible per-encoder default. See :func:write_frames_to_video.	None
extra_ffmpeg_params	list[str] | None	Optional raw FFmpeg parameters appended after the quality/preset flags.	None
log_interval	int | None	If set, log progress every log_interval frames at INFO level.	None
quiet	bool	Suppress the encoder-parameters log line and the progress bar. When True, log_interval is also ignored.	False

Raises:

Type	Description
ValueError	If image_paths is empty, an image's dimensions differ from the first image's, or mode is not one of "auto"/"gpu"/"cpu".

A missing or unreadable image path surfaces whatever error imageio raises while reading it (typically :class:FileNotFoundError).

Source code in src/pvio/io.py

def write_image_paths_to_video(
    video_path: Path | str,
    image_paths: list[Path | str],
    fps: float,
    *,
    mode: str = "auto",
    quality: int = _accel.DEFAULT_QUALITY,
    preset: str | None = None,
    extra_ffmpeg_params: list[str] | None = None,
    log_interval: int | None = None,
    quiet: bool = False,
) -> None:
    """Combine on-disk image files into an H.264 MP4 file.

    Like :func:`write_frames_to_video`, but the frames are given as paths to
    image files (PNG, JPEG, TIFF, …) instead of in-memory numpy arrays. Images
    are read lazily, one at a time, so an arbitrarily long sequence can be
    encoded without holding every frame in memory at once. The encoder selection,
    quality semantics, and NVENC→libx264 fallback are identical to
    :func:`write_frames_to_video`.

    Frames are encoded in the order given by *image_paths*; sort the paths
    beforehand if a particular ordering is required. The first image's spatial
    dimensions define the output size, and every subsequent image must match it.

    Args:
        video_path: Path for the output video file.
        image_paths: Paths to the image files to combine, in output order. Each
            file is read with imageio and must decode to an ``(H, W, C)`` array;
            all images must share the same spatial dimensions.
        fps: Frames per second of the output video.
        mode: Encoder selection — ``"auto"`` (default), ``"gpu"``, or ``"cpu"``.
            See :func:`write_frames_to_video` for details.
        quality: Encode quality on the 0-51 H.264 quantiser scale (lower = higher
            quality, larger files). See :func:`write_frames_to_video`.
        preset: Encoder preset. ``None`` uses a sensible per-encoder default. See
            :func:`write_frames_to_video`.
        extra_ffmpeg_params: Optional raw FFmpeg parameters appended after the
            quality/preset flags.
        log_interval: If set, log progress every *log_interval* frames at
            ``INFO`` level.
        quiet: Suppress the encoder-parameters log line and the progress bar.
            When ``True``, *log_interval* is also ignored.

    Raises:
        ValueError: If *image_paths* is empty, an image's dimensions differ from
            the first image's, or *mode* is not one of
            ``"auto"``/``"gpu"``/``"cpu"``.

    A missing or unreadable image path surfaces whatever error imageio raises
    while reading it (typically :class:`FileNotFoundError`).
    """
    if mode not in ("auto", "gpu", "cpu"):
        raise ValueError(f"mode must be 'auto', 'gpu', or 'cpu', got {mode!r}.")

    paths = [Path(p) for p in image_paths]
    if len(paths) == 0:
        raise ValueError("No image paths provided to write_image_paths_to_video")

    # Read the first image to determine the output frame size. The remaining
    # images are validated lazily as they are read during encoding.
    first_frame = imageio.imread(paths[0])
    frame_size = first_frame.shape[:2]
    height, width = frame_size[0], frame_size[1]

    frames = _ImagePathFrames(paths, frame_size)
    _encode_with_fallback(
        video_path,
        frames,
        len(paths),
        height,
        width,
        fps,
        mode=mode,
        quality=quality,
        preset=preset,
        extra_ffmpeg_params=extra_ffmpeg_params,
        log_interval=log_interval,
        quiet=quiet,
    )

check_num_frames

check_num_frames(video_path: Path | str) -> int

Return the number of frames in a video file.

Parameters:

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

Returns:

Type	Description
int	Total frame count.

Raises:

Type	Description
RuntimeError	If the file cannot be opened.

Source code in src/pvio/io.py

def check_num_frames(video_path: Path | str) -> int:
    """Return the number of frames in a video file.

    Args:
        video_path: Path to the video file.

    Returns:
        Total frame count.

    Raises:
        RuntimeError: If the file cannot be opened.
    """
    try:
        num_frames, _, _ = _probe_video(video_path)
    except Exception as e:
        raise RuntimeError(f"Failed to open video file: {video_path}") from e
    return num_frames

get_video_metadata

get_video_metadata(
    video_path: Path | str,
    cache_metadata: bool = True,
    use_cached_metadata: bool = True,
    metadata_suffix: str = ".metadata.json",
) -> VideoMetadata

Return frame count, frame size, and FPS for a video file.

Results are cached to a sidecar JSON file alongside the video to avoid re-reading on subsequent calls. The cache stores the video's (size, mtime) signature and is automatically invalidated (the metadata is re-read and the cache rewritten) whenever that signature changes, i.e. whenever the video has been modified, so using the cache is safe in practice.

Parameters:

Name	Type	Description	Default
video_path	Path | str	Path to the video file.	required
cache_metadata	bool	Write metadata to a cache file after reading.	True
use_cached_metadata	bool	Return cached metadata when the sidecar file exists and its (size, mtime) signature still matches the video. Set to False to force a fresh read regardless of the cache.	True
metadata_suffix	str	Suffix appended to the video filename to form the cache path. Default: ".metadata.json".	'.metadata.json'

Returns:

Name	Type	Description
A	VideoMetadata	class:VideoMetadata named tuple with fields n_frames (int),
	VideoMetadata	frame_size ((height, width) tuple), and fps (float or
	VideoMetadata	None if unavailable).

Source code in src/pvio/io.py

def get_video_metadata(
    video_path: Path | str,
    cache_metadata: bool = True,
    use_cached_metadata: bool = True,
    metadata_suffix: str = ".metadata.json",
) -> VideoMetadata:
    """Return frame count, frame size, and FPS for a video file.

    Results are cached to a sidecar JSON file alongside the video to avoid
    re-reading on subsequent calls. The cache stores the video's ``(size,
    mtime)`` signature and is automatically invalidated (the metadata is re-read
    and the cache rewritten) whenever that signature changes, i.e. whenever the
    video has been modified, so using the cache is safe in practice.

    Args:
        video_path: Path to the video file.
        cache_metadata: Write metadata to a cache file after reading.
        use_cached_metadata: Return cached metadata when the sidecar file exists
            and its ``(size, mtime)`` signature still matches the video. Set to
            ``False`` to force a fresh read regardless of the cache.
        metadata_suffix: Suffix appended to the video filename to form the
            cache path. Default: ``".metadata.json"``.

    Returns:
        A :class:`VideoMetadata` named tuple with fields ``n_frames`` (int),
        ``frame_size`` (``(height, width)`` tuple), and ``fps`` (float or
        ``None`` if unavailable).
    """
    video_path = Path(video_path)
    cache_path = video_path.parent / (video_path.name + metadata_suffix)

    if use_cached_metadata:
        if cache_path.is_file():
            cached = _read_metadata_cache(cache_path, video_path)
            if cached is not None:
                return cached
        else:
            logger.info(
                "No metadata cache found at %s; reading metadata from video.",
                cache_path,
            )

    n_frames, frame_size, fps = _probe_video(video_path)

    if cache_metadata:
        _write_metadata_cache(cache_path, video_path, n_frames, frame_size, fps)
        logger.info("Wrote metadata cache to %s.", cache_path)

    return VideoMetadata(n_frames=n_frames, frame_size=tuple(frame_size), fps=fps)