(self, kind: Literal["image", "audio", "video"], media_url: str)
| 10680 | return opener.open(request, timeout=timeout) |
| 10681 | |
| 10682 | def _load_media_url(self, kind: Literal["image", "audio", "video"], media_url: str) -> bytes: |
| 10683 | max_bytes = self._max_bytes_for_media(kind) |
| 10684 | if media_url.startswith("data:"): |
| 10685 | try: |
| 10686 | _, encoded = media_url.split(",", 1) |
| 10687 | except ValueError as exc: |
| 10688 | raise CompletionRequestValidationError(f"invalid data {kind} URL") from exc |
| 10689 | data = base64.b64decode(encoded, validate=False) |
| 10690 | if len(data) > max_bytes: |
| 10691 | raise CompletionRequestValidationError(f"{kind} exceeds model.mtmd.{kind}_max_bytes") |
| 10692 | return data |
| 10693 | if media_url.startswith("file:"): |
| 10694 | return self._load_media_file(kind, media_url) |
| 10695 | media_url = self._validate_remote_media_url(media_url) |
| 10696 | request = urllib.request.Request( |
| 10697 | media_url, |
| 10698 | headers={"User-Agent": "llama-cpp-python-batch-mtmd/0"}, |
| 10699 | ) |
| 10700 | try: |
| 10701 | with self._urlopen_without_redirects(request, timeout=self.image_timeout_seconds) as response: |
| 10702 | data = response.read(max_bytes + 1) |
| 10703 | except (OSError, TimeoutError, urllib.error.URLError) as exc: |
| 10704 | raise CompletionRequestValidationError( |
| 10705 | f"failed to fetch {kind} URL: {exc}" |
| 10706 | ) from exc |
| 10707 | if len(data) > max_bytes: |
| 10708 | raise CompletionRequestValidationError(f"{kind} exceeds model.mtmd.{kind}_max_bytes") |
| 10709 | return data |
| 10710 | |
| 10711 | def load_media(self, media: MediaInput) -> bytes: |
| 10712 | if media.url is not None: |
no test coverage detected