vllm.multimodal.video_decoders.torchcodec ¶

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",
        device: str = "cpu",
    ) -> "VideoDecoder":
        torch_device = torch.device(device)
        if torch_device.type == "cuda" and not current_platform.is_cuda():
            raise ValueError(
                f"torchcodec video decoding on device {device!r} requires "
                "a CUDA-capable platform."
            )
        elif torch_device.type not in ("cpu", "cuda"):
            raise ValueError(
                f"torchcodec video decoding only supports 'cpu' and 'cuda' "
                f"devices, got {device!r}."
            )
        # 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,
            device=device,
        )

    @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],
        *,
        device: str = "cpu",
    ) -> tuple[npt.NDArray | torch.Tensor, 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)
        frames = batch.data
        if (
            torch.device(device).type != "cpu"
            and (status := getattr(decoder, "cpu_fallback", None)) is not None
            and bool(status)
        ):
            logger.warning_once(
                "torchcodec could not use NVDEC for this video and "
                "decoded on CPU instead; check codec support and "
                "libnvcuvid."
            )
        return frames, list(frame_indices)