Skip to content

vllm.multimodal.video_decoders

Modules:

Classes:

Functions:

VideoSourceMetadata

Bases: NamedTuple

Metadata describing the encoded video source.

Source code in vllm/multimodal/video_decoders/base.py 
20
21
22
23
24
25
class VideoSourceMetadata(NamedTuple):
    """Metadata describing the encoded video source."""

    total_frames_num: int
    original_fps: float
    duration: float

VideoTargetMetadata

Bases: NamedTuple

Metadata describing the requested video sample.

Source code in vllm/multimodal/video_decoders/base.py 
12
13
14
15
16
17
class VideoTargetMetadata(NamedTuple):
    """Metadata describing the requested video sample."""

    num_frames: int
    fps: float
    max_duration: float

check_frame_pixel_limit(width, height)

Reject video frames exceeding VLLM_MAX_IMAGE_PIXELS.

Source code in vllm/multimodal/video_decoders/base.py 
28
29
30
31
32
33
34
35
36
37
def check_frame_pixel_limit(width: int, height: int) -> None:
    """Reject video frames exceeding ``VLLM_MAX_IMAGE_PIXELS``."""
    max_pixels = envs.VLLM_MAX_IMAGE_PIXELS
    if max_pixels > 0 and width * height > max_pixels:
        raise ValueError(
            f"Video frame dimensions {width}x{height} "
            f"({width * height} pixels) exceed the maximum of "
            f"{max_pixels} pixels. Set VLLM_MAX_IMAGE_PIXELS to "
            f"increase this limit."
        )

decode_video(backend, loader_cls, data, target, sampling_kwargs, backend_kwargs, *, frame_recovery)

Decode a sampled video, importing only the selected backend.

Source code in vllm/multimodal/video_decoders/__init__.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def decode_video(
    backend: str,
    loader_cls,
    data: bytes,
    target: VideoTargetMetadata,
    sampling_kwargs: dict[str, Any],
    backend_kwargs: dict[str, Any],
    *,
    frame_recovery: bool,
):
    """Decode a sampled video, importing only the selected backend."""
    _get_backend_option_defaults(backend)
    if frame_recovery and backend != "opencv":
        error = (
            ValueError if backend == PYNVVIDEOCODEC_VIDEO_BACKEND else AssertionError
        )
        raise error(f"frame_recovery is not supported by the {backend!r} backend")

    decoder_kwargs = dict(backend_kwargs)
    if backend == "opencv":
        decoder_kwargs["frame_recovery"] = frame_recovery
    module = import_module(f".{backend}", __name__)
    decoder = getattr(module, f"decode_{backend}")
    return decoder(
        loader_cls,
        data,
        target,
        sampling_kwargs,
        **decoder_kwargs,
    )

incompatible_backend_options(old_backend, new_backend)

Option names accepted by old_backend but not by new_backend.

Source code in vllm/multimodal/video_decoders/__init__.py 
45
46
47
48
49
50
51
52
def incompatible_backend_options(
    old_backend: str | None,
    new_backend: str,
) -> set[str]:
    """Option names accepted by ``old_backend`` but not by ``new_backend``."""
    old = _BACKEND_OPTION_DEFAULTS.get(old_backend or "", {})
    new = _BACKEND_OPTION_DEFAULTS.get(new_backend, {})
    return old.keys() - new.keys()

resolve_video_backend_kwargs(backend, kwargs)

Split frame-sampling kwargs from options owned by a decoder backend.

Source code in vllm/multimodal/video_decoders/__init__.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def resolve_video_backend_kwargs(
    backend: str,
    kwargs: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
    """Split frame-sampling kwargs from options owned by a decoder backend."""
    defaults = _get_backend_option_defaults(backend)
    sampling_kwargs = dict(kwargs)
    backend_kwargs = dict(defaults)
    backend_option_names = {
        name for options in _BACKEND_OPTION_DEFAULTS.values() for name in options
    }
    misplaced = (sampling_kwargs.keys() & backend_option_names) - defaults.keys()
    if misplaced:
        names = ", ".join(sorted(misplaced))
        raise ValueError(f"{names} is not supported by the {backend!r} backend")

    for name in defaults:
        if name in sampling_kwargs:
            backend_kwargs[name] = sampling_kwargs.pop(name)

    return sampling_kwargs, backend_kwargs