Skip to content

vllm.assets.video

VideoAssetName module-attribute 

VideoAssetName = Literal['baby_reading']

VideoAsset dataclass

Source code in vllm/assets/video.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@dataclass(frozen=True)
class VideoAsset:
    name: VideoAssetName
    num_frames: int = -1

    _NAME_TO_FILE: ClassVar[dict[VideoAssetName, str]] = {
        "baby_reading": "sample_demo_1.mp4",
    }

    @property
    def filename(self) -> str:
        return self._NAME_TO_FILE[self.name]

    @property
    def video_path(self) -> str:
        return download_video_asset(self.filename)

    @property
    def pil_images(self) -> list[Image.Image]:
        ret = video_to_pil_images_list(self.video_path, self.num_frames)
        return ret

    @property
    def np_ndarrays(self) -> npt.NDArray:
        ret = video_to_ndarrays(self.video_path, self.num_frames)
        return ret

    @property
    def metadata(self) -> dict[str, Any]:
        ret = video_get_metadata(self.video_path, self.num_frames)
        return ret

    def get_audio(self, sampling_rate: float | None = None) -> npt.NDArray:
        """
        Read audio data from the video asset, used in Qwen2.5-Omni examples.

        See also: examples/offline_inference/qwen2_5_omni/only_thinker.py
        """
        return librosa.load(self.video_path, sr=sampling_rate)[0]

_NAME_TO_FILE class-attribute 

_NAME_TO_FILE: dict[VideoAssetName, str] = {
    "baby_reading": "sample_demo_1.mp4"
}

filename property 

filename: str

metadata property 

metadata: dict[str, Any]

name instance-attribute 

name: VideoAssetName

np_ndarrays property 

np_ndarrays: NDArray

num_frames class-attribute instance-attribute 

num_frames: int = -1

pil_images property 

pil_images: list[Image]

video_path property 

video_path: str

__init__ 

__init__(
    name: VideoAssetName, num_frames: int = -1
) -> None

get_audio 

get_audio(sampling_rate: float | None = None) -> NDArray

Read audio data from the video asset, used in Qwen2.5-Omni examples.

See also: examples/offline_inference/qwen2_5_omni/only_thinker.py

Source code in vllm/assets/video.py 
143
144
145
146
147
148
149
def get_audio(self, sampling_rate: float | None = None) -> npt.NDArray:
    """
    Read audio data from the video asset, used in Qwen2.5-Omni examples.

    See also: examples/offline_inference/qwen2_5_omni/only_thinker.py
    """
    return librosa.load(self.video_path, sr=sampling_rate)[0]

download_video_asset cached 

download_video_asset(filename: str) -> str

Download and open an image from huggingface repo: raushan-testing-hf/videos-test

Source code in vllm/assets/video.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@lru_cache
def download_video_asset(filename: str) -> str:
    """
    Download and open an image from huggingface
    repo: raushan-testing-hf/videos-test
    """
    video_directory = get_cache_dir() / "video-example-data"
    video_directory.mkdir(parents=True, exist_ok=True)

    video_path = video_directory / filename
    video_path_str = str(video_path)
    if not video_path.exists():
        video_path_str = hf_hub_download(
            repo_id="raushan-testing-hf/videos-test",
            filename=filename,
            repo_type="dataset",
            cache_dir=video_directory,
        )
    return video_path_str

video_get_metadata 

video_get_metadata(
    path: str, num_frames: int = -1
) -> dict[str, Any]
Source code in vllm/assets/video.py 
 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
def video_get_metadata(path: str, num_frames: int = -1) -> dict[str, Any]:
    import cv2

    cap = cv2.VideoCapture(path)
    if not cap.isOpened():
        raise ValueError(f"Could not open video file {path}")

    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    fps = cap.get(cv2.CAP_PROP_FPS)
    duration = total_frames / fps if fps > 0 else 0

    if num_frames == -1 or num_frames > total_frames:
        num_frames = total_frames

    metadata = {
        "total_num_frames": num_frames,
        "fps": duration / num_frames,
        "duration": duration,
        "video_backend": "opencv",
        "frames_indices": list(range(num_frames)),
        # extra field used to control hf processor's video
        # sampling behavior
        "do_sample_frames": num_frames == total_frames,
    }
    return metadata

video_to_ndarrays 

video_to_ndarrays(
    path: str, num_frames: int = -1
) -> NDArray
Source code in vllm/assets/video.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def video_to_ndarrays(path: str, num_frames: int = -1) -> npt.NDArray:
    import cv2

    cap = cv2.VideoCapture(path)
    if not cap.isOpened():
        raise ValueError(f"Could not open video file {path}")

    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    frames = []

    num_frames = num_frames if num_frames > 0 else total_frames
    frame_indices = np.linspace(0, total_frames - 1, num_frames, dtype=int)
    for idx in range(total_frames):
        ok = cap.grab()  # next img
        if not ok:
            break
        if idx in frame_indices:  # only decompress needed
            ret, frame = cap.retrieve()
            if ret:
                # OpenCV uses BGR format, we need to convert it to RGB
                # for PIL and transformers compatibility
                frames.append(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))

    frames = np.stack(frames)
    if len(frames) < num_frames:
        raise ValueError(
            f"Could not read enough frames from video file {path}"
            f" (expected {num_frames} frames, got {len(frames)})"
        )
    return frames

video_to_pil_images_list 

video_to_pil_images_list(
    path: str, num_frames: int = -1
) -> list[Image]
Source code in vllm/assets/video.py 
76
77
78
def video_to_pil_images_list(path: str, num_frames: int = -1) -> list[Image.Image]:
    frames = video_to_ndarrays(path, num_frames)
    return [Image.fromarray(frame) for frame in frames]