Repository: vllm-project/vllm
File: vllm/multimodal/video.py (Python)

Code before:
        }

        return frames, metadata

    @classmethod
    def load_bytes(
        cls,
        data: bytes,
        num_frames: int = -1,
        **kwargs,
    ) -> tuple[npt.NDArray, dict[str, Any]]:
        frame_sample_mode = cast(str | None, kwargs.pop("frame_sample_mode", None))
        max_fps = cast(int, kwargs.pop("max_fps", 2))
        sampling_fps = cast(int, kwargs.pop("sampling_fps", 2))
        out = cls.load_bytes_opencv(
            data,
            frame_sample_mode,
            num_frames,
            max_fps,
            sampling_fps,
            **kwargs,
        )
        return out


@VIDEO_LOADER_REGISTRY.register("opencv_dynamic_openpangu")
class OpenCVDynamicOpenPanguVideoBackend(OpenCVVideoBackend):
    @classmethod
    def load_bytes(
        cls,
        data: bytes,
        num_frames: int = 32,
        fps: int = 1,
        max_duration: int = 300,
        frame_recovery: bool = False,
        **kwargs,
    ) -> tuple[npt.NDArray, dict[str, Any]]:
        """
        Load video frames with dynamic sampling based on duration.
        Assume that total_num_frames = 10 and fps = 1.
        The timestamp of frame 0 is 0.0.
        The timestamp of frame 1 is 1.0.…
        The timestamp of frame 9 (the last frame) should be 9.0, that is,
        (total_frames_num – 1) / original_fps.

        Args:
            data: Raw video bytes
            num_frames: Not used in dynamic backend
            fps: Target FPS for sampling (default: 1)

        Returns:

Review comment (suggestion):
```suggestion
@VIDEO_LOADER_REGISTRY.register("openpangu")
```
Since this is model-specific video loader, I prefer to register model type here.

Code after:
        }

        return frames, metadata

    @classmethod
    def load_bytes(
        cls,
        data: bytes,
        num_frames: int = -1,
        **kwargs,
    ) -> tuple[npt.NDArray, dict[str, Any]]:
        frame_sample_mode = cast(str | None, kwargs.pop("frame_sample_mode", None))
        max_fps = cast(int, kwargs.pop("max_fps", 2))
        sampling_fps = cast(int, kwargs.pop("sampling_fps", 2))
        out = cls.load_bytes_opencv(
            data,
            frame_sample_mode,
            num_frames,
            max_fps,
            sampling_fps,
            **kwargs,
        )
        return out


@VIDEO_LOADER_REGISTRY.register("openpangu")
class OpenCVDynamicOpenPanguVideoBackend(OpenCVVideoBackend):
    @classmethod
    def load_bytes(
        cls,
        data: bytes,
        num_frames: int = 32,
        fps: int = 1,
        max_duration: int = 300,
        frame_recovery: bool = False,
        **kwargs,
    ) -> tuple[npt.NDArray, dict[str, Any]]:
        """
        Load video frames with dynamic sampling based on duration.
        Assume that total_num_frames = 10 and fps = 1.
        The timestamp of frame 0 is 0.0.
        The timestamp of frame 1 is 1.0.…
        The timestamp of frame 9 (the last frame) should be 9.0, that is,
        (total_frames_num – 1) / original_fps.

        Args:
            data: Raw video bytes
            num_frames: Not used in dynamic backend
            fps: Target FPS for sampling (default: 1)

        Returns: