| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | from abc import ABC, abstractmethod |
| 5 | from collections.abc import Iterable, Iterator |
| 6 | from dataclasses import dataclass, field |
| 7 | from pathlib import Path |
| 8 | from typing import Any, Generic, TypeVar, cast |
| 9 | |
| 10 | import numpy as np |
| 11 | |
| 12 | _T = TypeVar("_T") |
| 13 | |
| 14 | |
| 15 | @dataclass |
| 16 | class MediaWithBytes(Generic[_T]): |
| 17 | """Wrapper that couples a media object with its original encoded bytes. |
| 18 | |
| 19 | This ensures the raw bytes and media object remain synchronized, |
| 20 | preventing cache corruption from in-place modifications. |
| 21 | |
| 22 | The wrapper delegates attribute access to the underlying media object, |
| 23 | making it behave transparently like the wrapped type (e.g., PIL.Image). |
| 24 | |
| 25 | NOTE: Currently, this wrapper is used only for the image and video |
| 26 | modalities. |
| 27 | """ |
| 28 | |
| 29 | media: _T |
| 30 | original_bytes: bytes = field(repr=False) |
| 31 | io_config: dict[str, Any] | None = None |
| 32 | """Decode settings that altered the media relative to `original_bytes` |
| 33 | (e.g. `image_mode` conversion), so they participate in cache hashing.""" |
| 34 | |
| 35 | def __array__(self, *args, **kwargs) -> np.ndarray: |
| 36 | """Allow np.array(obj) to return np.array(obj.media).""" |
| 37 | return np.array(self.media, *args, **kwargs) |
| 38 | |
| 39 | def __iter__(self) -> Iterator[Any]: |
| 40 | """Allow unpacking obj to unpack obj.media (e.g. video tuples).""" |
| 41 | return iter(cast(Iterable[Any], self.media)) |
| 42 | |
| 43 | def __getitem__(self, index: Any) -> Any: |
| 44 | """Allow obj[i] to index obj.media (e.g. video tuples).""" |
| 45 | return cast(Any, self.media)[index] |
| 46 | |
| 47 | def __getstate__(self): |
| 48 | return self.__dict__.copy() |
| 49 | |
| 50 | def __setstate__(self, state: dict[str, Any]): |
| 51 | self.__dict__.update(state) |
| 52 | |
| 53 | def __getattr__(self, name: str): |
| 54 | """Delegate attribute access to the underlying media object.""" |
| 55 | return getattr(self.media, name) |
| 56 | |
| 57 | |
| 58 | class MediaIO(ABC, Generic[_T]): |
| 59 | """Configuration values can be user-provided either by --media-io-kwargs or |
| 60 | by the runtime API field "media_io_kwargs". Ensure proper validation and |
| 61 | error handling. |
| 62 | """ |
| 63 | |
| 64 | @classmethod |
| 65 | def merge_kwargs( |
| 66 | cls, |
| 67 | default_kwargs: dict[str, Any] | None, |
| 68 | runtime_kwargs: dict[str, Any] | None, |
| 69 | ) -> dict[str, Any]: |
| 70 | """Merge config-level kwargs and request-level kwargs. |
| 71 | |
| 72 | By default this performs a shallow merge where runtime kwargs override |
| 73 | keys in default kwargs. Subclasses may override to apply modality- |
| 74 | specific behavior. |
| 75 | """ |
| 76 | merged = dict(default_kwargs or {}) |
| 77 | if runtime_kwargs: |
| 78 | merged.update(runtime_kwargs) |
| 79 | return merged |
| 80 | |
| 81 | def get_max_bytes(self) -> int | None: |
| 82 | """Return the maximum encoded payload size accepted by this media IO.""" |
| 83 | return None |
| 84 | |
| 85 | @abstractmethod |
| 86 | def load_bytes(self, data: bytes) -> _T: |
| 87 | raise NotImplementedError |
| 88 | |
| 89 | @abstractmethod |
| 90 | def load_base64(self, media_type: str, data: str) -> _T: |
| 91 | """List of media types: |
| 92 | https://www.iana.org/assignments/media-types/media-types.xhtml |
| 93 | """ |
| 94 | raise NotImplementedError |
| 95 | |
| 96 | @abstractmethod |
| 97 | def load_file(self, filepath: Path) -> _T: |
| 98 | raise NotImplementedError |