Skip to content

vllm.multimodal.video_decoders.torchcodec

Classes:

TorchCodecVideoBackendMixin

TorchCodec (FFmpeg-backed, PyTorch-native) codec utilities.

Builds a :class:~torchcodec.decoders.VideoDecoder over the in-memory bytes and extracts the sampled indices with a single batched get_frames_at call, while releasing the GIL during decode.

Methods:

Source code in vllm/multimodal/video_decoders/torchcodec.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class TorchCodecVideoBackendMixin:
    """TorchCodec (FFmpeg-backed, PyTorch-native) codec utilities.

    Builds a :class:`~torchcodec.decoders.VideoDecoder` over the in-memory
    bytes and extracts the sampled indices with a single batched
    ``get_frames_at`` call, while releasing the GIL during decode.
    """

    @staticmethod
    def make_torchcodec_decoder(
        data: bytes,
        *,
        num_ffmpeg_threads: int = 0,
        seek_mode: Literal["exact", "approximate"] = "exact",
    ) -> "VideoDecoder":
        # NHWC matches the (num_frames, H, W, 3) uint8 RGB layout the rest
        # of the pipeline expects, avoiding a transpose.
        return VideoDecoder(
            data,
            dimension_order="NHWC",
            num_ffmpeg_threads=num_ffmpeg_threads,
            seek_mode=seek_mode,
        )

    @staticmethod
    def get_torchcodec_metadata(decoder: "VideoDecoder") -> VideoSourceMetadata:
        md = decoder.metadata
        total_frames = md.num_frames or 0
        fps = float(md.average_fps) if md.average_fps else 0.0
        duration = float(md.duration_seconds) if md.duration_seconds else 0.0
        if total_frames == 0 and duration > 0 and fps > 0:
            total_frames = int(duration * fps)
        return VideoSourceMetadata(total_frames, fps, duration)

    @staticmethod
    def decode_torchcodec_frames(
        decoder: "VideoDecoder",
        frame_indices: list[int],
    ) -> tuple[npt.NDArray, list[int]]:
        """Decode the requested indices in one batched, index-exact call."""
        if not frame_indices:
            return np.empty((0,), dtype=np.uint8), []
        # Note: torchcodec releases the GIL for the entire call
        batch = decoder.get_frames_at(frame_indices)
        return batch.data.numpy(), list(frame_indices)

decode_torchcodec_frames(decoder, frame_indices) staticmethod

Decode the requested indices in one batched, index-exact call.

Source code in vllm/multimodal/video_decoders/torchcodec.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@staticmethod
def decode_torchcodec_frames(
    decoder: "VideoDecoder",
    frame_indices: list[int],
) -> tuple[npt.NDArray, list[int]]:
    """Decode the requested indices in one batched, index-exact call."""
    if not frame_indices:
        return np.empty((0,), dtype=np.uint8), []
    # Note: torchcodec releases the GIL for the entire call
    batch = decoder.get_frames_at(frame_indices)
    return batch.data.numpy(), list(frame_indices)