xiaoxue_admin
/
supervision
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add image URL loader (#2372) 

- Added image loading from HTTP and HTTPS URLs with descriptive URL validation errors
- Added optional caching for image URL loads using the shared Supervision cache
- Improved URL downloads with atomic file replacement and shared download behavior across image loading and asset downloads
- Updated image decoding compatibility with Pillow fallbacks when OpenCV decoding or encoding is unavailable

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Jirka Borovec <6035284+Borda@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Erik 2026-07-28 21:24:55 +02:00 committed by GitHub
parent 5212af7b70
commit 541b0226cd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 672 additions and 264 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/changelog.md
View File

@ -67,6 +67,7 @@ date_modified: 2026-07-27
### Added
- `sv.load_image_from_url` — load an image from an HTTP(S) URL as an OpenCV image, with optional on-disk caching under the shared supervision cache directory (`{tmpdir}/supervision/image-url/` by default, configurable via `cache_dir`) ([#2372](https://github.com/roboflow/supervision/pull/2372))
- `sv.VLM.GOOGLE_GEMINI_3_5` — `sv.Detections.from_vlm` now parses Google Gemini 3.5 output (detection and segmentation), reusing the Gemini 2.5 JSON format (`box_2d` + `label`, optional `mask`/`confidence`).
- `sv.get_video_frames_generator` now accepts `prefetch: int = 0` ([#2273](https://github.com/roboflow/supervision/pull/2273)). When `> 0`, frames are decoded on a background daemon thread and buffered in a bounded queue, overlapping I/O with consumer processing. Default `0` preserves the existing synchronous behaviour.
- Added a cv2-free PyAV fallback for file-video capture, writing, frame seeking, metadata, and `process_video(preserve_audio=True)` audio remuxing. OpenCV remains the primary backend when available; `av>=14.2.0` is now required alongside OpenCV during the transition, with the later OpenCV-removal integration removing the OpenCV dependency.

6
docs/utils/image.md
View File

@ -11,6 +11,12 @@ status: new
:::supervision.utils.image.crop_image
<div class="md-typeset">
<h2><a href="#supervision.utils.image.load_image_from_url">load_image_from_url</a></h2>
</div>
:::supervision.utils.image.load_image_from_url
<div class="md-typeset">
<h2><a href="#supervision.utils.image.scale_image">scale_image</a></h2>
</div>

2
src/supervision/__init__.py
View File

@ -147,6 +147,7 @@ from supervision.utils.image import (
get_image_resolution_wh,
grayscale_image,
letterbox_image,
load_image_from_url,
overlay_image,
resize_image,
scale_image,
@ -268,6 +269,7 @@ __all__ = [
"is_valid_hex",
"letterbox_image",
"list_files_with_extensions",
"load_image_from_url",
"mask_iou_batch",
"mask_non_max_merge",
"mask_non_max_suppression",

8
src/supervision/_cv2/__init__.py
View File

@ -32,6 +32,8 @@ from supervision._cv2._image import (
_convert_scale_abs,
_copy_make_border,
_flip,
_imdecode,
_imencode,
_imread,
_imwrite,
_mean,
@ -132,6 +134,8 @@ if _IS_CV2_AVAILABLE:
fillPoly,
flip,
getTextSize,
imdecode,
imencode,
imread,
imwrite,
intersectConvexConvex,
@ -202,6 +206,8 @@ else:
_find_contours_impl = _find_contours
flip = _flip # type: ignore[assignment]
getTextSize = _get_text_size # type: ignore[assignment]
imdecode = _imdecode # type: ignore[assignment]
imencode = _imencode # type: ignore[assignment]
imread = _imread # type: ignore[assignment]
imwrite = _imwrite # type: ignore[assignment]
intersectConvexConvex = _intersect_convex_convex # type: ignore[assignment]
@ -271,6 +277,8 @@ __all__ = [
"find_contours",
"flip",
"getTextSize",
"imdecode",
"imencode",
"imread",
"imwrite",
"intersectConvexConvex",

61
src/supervision/_cv2/_image.py
View File

@ -207,12 +207,12 @@ def _resize(
return np.ascontiguousarray(_cast_array_like_opencv(resized, src.dtype))
def _imread(filename: str, flags: int = _IMREAD_COLOR) -> npt.NDArray[Any] | None:
"""Read an image with Pillow while returning BGR or BGRA arrays."""
def _read_pil_source(source: Any, flags: int) -> npt.NDArray[Any] | None:
"""Decode any source Pillow can open into BGR or BGRA arrays."""
from PIL import Image
try:
with Image.open(filename) as image:
with Image.open(source) as image:
if flags == _IMREAD_UNCHANGED:
if image.mode == "P":
converted = image.convert(
@ -229,7 +229,7 @@ def _imread(filename: str, flags: int = _IMREAD_COLOR) -> npt.NDArray[Any] | Non
values = np.repeat(values[..., np.newaxis], 3, axis=2)
else:
values = np.asarray(image.convert("RGB"))
except (FileNotFoundError, OSError):
except (FileNotFoundError, OSError, ValueError):
return None
if values.ndim == 3 and values.shape[2] == 3:
@ -239,6 +239,31 @@ def _imread(filename: str, flags: int = _IMREAD_COLOR) -> npt.NDArray[Any] | Non
return np.ascontiguousarray(values)
def _imread(filename: str, flags: int = _IMREAD_COLOR) -> npt.NDArray[Any] | None:
"""Read an image with Pillow while returning BGR or BGRA arrays."""
return _read_pil_source(filename, flags)
def _imdecode(
buf: npt.NDArray[Any], flags: int = _IMREAD_COLOR
) -> npt.NDArray[Any] | None:
"""Decode in-memory encoded image bytes, returning BGR or BGRA arrays."""
import io
data = np.asarray(buf, dtype=np.uint8).tobytes()
return _read_pil_source(io.BytesIO(data), flags)
def _bgr_to_pil_values(image: npt.NDArray[Any]) -> npt.NDArray[Any]:
"""Reorder BGR or BGRA channels into the RGB order Pillow expects."""
values = np.asarray(image)
if values.ndim == 3 and values.shape[2] == 3:
values = values[..., ::-1]
elif values.ndim == 3 and values.shape[2] == 4:
values = values[..., [2, 1, 0, 3]]
return np.ascontiguousarray(values)
def _imwrite(
filename: str, image: npt.NDArray[Any], params: Sequence[int] | None = None
) -> bool:
@ -246,13 +271,29 @@ def _imwrite(
from PIL import Image
del params
values = np.asarray(image)
if values.ndim == 3 and values.shape[2] == 3:
values = values[..., ::-1]
elif values.ndim == 3 and values.shape[2] == 4:
values = values[..., [2, 1, 0, 3]]
try:
Image.fromarray(np.ascontiguousarray(values)).save(filename)
Image.fromarray(_bgr_to_pil_values(image)).save(filename)
except (OSError, ValueError):
return False
return True
def _imencode(
ext: str, image: npt.NDArray[Any], params: Sequence[int] | None = None
) -> tuple[bool, npt.NDArray[np.uint8] | None]:
"""Encode a BGR or BGRA array in memory, mirroring `cv2.imencode`'s return."""
import io
from PIL import Image
del params
# Pillow registers the JPEG codec as "JPEG", not the "jpg" file extension.
image_format = ext.lstrip(".").upper()
if image_format == "JPG":
image_format = "JPEG"
buffer = io.BytesIO()
try:
Image.fromarray(_bgr_to_pil_values(image)).save(buffer, format=image_format)
except (KeyError, OSError, ValueError):
return False, None
return True, np.frombuffer(buffer.getvalue(), dtype=np.uint8)

29
src/supervision/assets/downloader.py
View File

@ -1,12 +1,9 @@
import os
from hashlib import md5
from pathlib import Path
from shutil import copyfileobj
from requests import get
from tqdm.auto import tqdm
from supervision.assets.list import MEDIA_ASSETS, Assets
from supervision.utils.file import _download_to_file
from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
@ -40,29 +37,7 @@ def _download_asset(filename: str, destination: Path) -> None:
"""
Download asset bytes to the target destination via a temporary file.
"""
response = get(
MEDIA_ASSETS[filename][0], stream=True, allow_redirects=True, timeout=30
)
response.raise_for_status()
file_size = int(response.headers.get("Content-Length", 0))
destination.parent.mkdir(parents=True, exist_ok=True)
temp_path = destination.with_name(f"{destination.name}.part")
try:
with tqdm.wrapattr(
response.raw, "read", total=file_size, desc="", colour="#a351fb"
) as raw_resp:
with temp_path.open("wb") as file:
copyfileobj(raw_resp, file)
except Exception:
temp_path.unlink(missing_ok=True)
raise
try:
os.replace(temp_path, destination)
finally:
temp_path.unlink(missing_ok=True)
_download_to_file(MEDIA_ASSETS[filename][0], destination, timeout=30.0, stream=True)
def _download_verified_asset(

94
src/supervision/utils/file.py
View File

@ -1,10 +1,104 @@
import json
import os
import tempfile
import urllib.parse
from pathlib import Path
from shutil import copyfileobj
from typing import Any
import numpy as np
import requests
import yaml
SUPERVISION_CACHE_DIR = Path(tempfile.gettempdir()) / "supervision"
def _normalize_http_url(url: str) -> str:
"""
Validate and normalize an HTTP(S) URL.
Args:
url: URL to validate.
Returns:
Normalized URL string.
Raises:
ValueError: If the URL is invalid or uses an unsupported scheme.
"""
try:
original_parsed_url = urllib.parse.urlparse(url)
if "\\" in original_parsed_url.netloc:
raise ValueError("URL authority contains a backslash")
prepared_request = requests.Request(method="GET", url=url).prepare()
prepared_url = prepared_request.url
if prepared_url is None:
raise ValueError("prepared URL is empty")
parsed_url = urllib.parse.urlparse(prepared_url)
except (requests.RequestException, ValueError) as error:
raise ValueError(f"Invalid URL {url!r}: {error}") from error
if parsed_url.scheme not in {"http", "https"}:
raise ValueError(
f"Unsupported URL scheme {parsed_url.scheme!r} in {url!r}. "
"Only HTTP and HTTPS URLs are supported."
)
if parsed_url.hostname is None:
raise ValueError(f"Invalid URL {url!r}: no host supplied.")
return prepared_url
def _download_to_file(
url: str, target: Path, *, timeout: float = 30.0, stream: bool = False
) -> None:
"""
Download `url` to `target` atomically using a temporary file and `os.replace`.
Args:
url: HTTP(S) URL to download.
target: Destination file path. Parent directories are created as needed.
timeout: Request timeout in seconds. Defaults to `30.0`.
stream: If `True`, stream the response to disk in chunks and display a
progress bar. If `False`, buffer the response in memory before
writing. Defaults to `False`.
Raises:
requests.RequestException: If the request fails or returns an error status.
"""
response = requests.get(url, stream=stream, allow_redirects=True, timeout=timeout)
try:
response.raise_for_status()
target.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(
delete=False, dir=target.parent, suffix=target.suffix
) as file:
temp_path = Path(file.name)
try:
if stream:
from tqdm.auto import tqdm
file_size = int(response.headers.get("Content-Length", 0))
with tqdm.wrapattr(
response.raw, "read", total=file_size, desc="", colour="#a351fb"
) as raw_response:
copyfileobj(raw_response, file)
else:
file.write(response.content)
except BaseException:
file.close()
temp_path.unlink(missing_ok=True)
raise
finally:
response.close()
try:
os.replace(temp_path, target)
finally:
temp_path.unlink(missing_ok=True)
class NumpyJsonEncoder(json.JSONEncoder):
def default(self, obj: Any) -> Any:

118
src/supervision/utils/image.py
View File

@ -4,8 +4,12 @@ import itertools
import math
import os
import shutil
import tempfile
import urllib.parse
from collections.abc import Callable
from functools import partial
from hashlib import md5
from pathlib import Path
from types import TracebackType
from typing import Literal, cast
@ -27,12 +31,126 @@ from supervision.utils.conversion import (
ensure_cv2_image_for_standalone_function,
images_to_cv2,
)
from supervision.utils.file import (
SUPERVISION_CACHE_DIR,
_download_to_file,
_normalize_http_url,
)
from supervision.utils.iterables import create_batches, fill
RelativePosition = Literal["top", "bottom"]
MAX_COLUMNS_FOR_SINGLE_ROW_GRID = 3
DEFAULT_IMAGE_URL_CACHE_DIR = SUPERVISION_CACHE_DIR / "image-url"
def _get_image_url_cache_path(value: str, cache_dir: str | Path | None) -> Path:
"""
Build the cache file path for a URL: `<cache root>/<md5(url)><suffix>`.
"""
cache_root = (
DEFAULT_IMAGE_URL_CACHE_DIR
if cache_dir is None
else Path(cache_dir).expanduser().resolve()
)
url_path = urllib.parse.urlparse(value).path
suffix = Path(url_path).suffix or ".image"
url_hash = md5(value.encode("utf-8"), usedforsecurity=False).hexdigest()
return cache_root / f"{url_hash}{suffix}"
def _decode_image_from_bytes(
value: bytes,
cv_imread_flags: int,
) -> npt.NDArray[np.uint8]:
"""
Decode raw image bytes into an OpenCV image, raising on undecodable data.
"""
image = cv2.imdecode(
np.frombuffer(value, dtype=np.uint8),
cv_imread_flags,
)
if image is None:
raise ValueError("Data pointed by URL could not be decoded into image.")
return cast(npt.NDArray[np.uint8], image)
def load_image_from_url(
value: str,
cv_imread_flags: int = cv2.IMREAD_COLOR,
timeout: float = 30.0,
use_cache: bool = True,
cache_dir: str | Path | None = None,
force_reload: bool = False,
) -> npt.NDArray[np.uint8]:
"""
Load an image from a URL as an OpenCV image.
Args:
value: HTTP(S) URL of the image.
cv_imread_flags: OpenCV image read flag passed to `cv2.imdecode`.
Defaults to `cv2.IMREAD_COLOR`.
timeout: Request timeout in seconds. Defaults to `30.0`.
use_cache: If `True`, cache downloaded image bytes locally and reuse them
on repeated calls. Defaults to `True`.
cache_dir: Directory where downloaded image bytes are cached. If `None`,
uses the system temporary directory. Defaults to `None`.
force_reload: If `True`, re-download the image and refresh the cache.
Defaults to `False`.
Returns:
Image as a NumPy array in the format selected by `cv_imread_flags`.
Raises:
ValueError: If the URL is invalid or the downloaded bytes cannot be decoded.
requests.RequestException: If the request fails or returns an error status.
Examples:
```python
import supervision as sv
image = sv.load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg"
)
image.shape
```
"""
prepared_url = _normalize_http_url(url=value)
if not use_cache:
with tempfile.TemporaryDirectory() as temp_dir:
temp_target = Path(temp_dir) / "image"
_download_to_file(prepared_url, temp_target, timeout=timeout)
return _decode_image_from_bytes(
value=temp_target.read_bytes(),
cv_imread_flags=cv_imread_flags,
)
cache_path = _get_image_url_cache_path(
value=prepared_url,
cache_dir=cache_dir,
)
if cache_path.exists() and not force_reload:
try:
return _decode_image_from_bytes(
value=cache_path.read_bytes(),
cv_imread_flags=cv_imread_flags,
)
except ValueError:
cache_path.unlink(missing_ok=True)
_download_to_file(prepared_url, cache_path, timeout=timeout)
try:
return _decode_image_from_bytes(
value=cache_path.read_bytes(),
cv_imread_flags=cv_imread_flags,
)
except ValueError:
cache_path.unlink(missing_ok=True)
raise
@ensure_cv2_image_for_standalone_function
def crop_image(

277
tests/assets/test_downloader.py
View File

@ -1,3 +1,5 @@
import io
from pathlib import Path
from unittest.mock import MagicMock, mock_open, patch
import pytest
@ -7,7 +9,15 @@ from supervision.assets.downloader import (
download_assets,
is_md5_hash_matching,
)
from supervision.assets.list import ImageAssets, VideoAssets
from supervision.assets.list import MEDIA_ASSETS, ImageAssets, VideoAssets
def _mock_streaming_response(payload: bytes = b"asset-bytes") -> MagicMock:
response = MagicMock()
response.headers = {"Content-Length": str(len(payload))}
response.raw = io.BytesIO(payload)
response.raise_for_status = MagicMock()
return response
class TestMD5HashMatching:
@ -40,237 +50,108 @@ class TestMD5HashMatching:
class TestDownloadAssets:
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.exists", return_value=True)
def test_already_exists_and_valid(
self, mock_exists, mock_md5, mock_logger, mock_replace
) -> None:
def test_already_exists_and_valid(self, mock_exists, mock_md5, mock_logger) -> None:
"""Test download_assets when file already exists and is valid."""
filename = "vehicles.mp4"
result = download_assets(filename)
assert result == filename
mock_logger.info.assert_called_with("%s asset download complete.", filename)
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("os.remove")
@patch("supervision.assets.downloader._download_asset")
@patch(
"supervision.assets.downloader.is_md5_hash_matching",
side_effect=[False, True],
)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("pathlib.Path.exists", return_value=True)
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
def test_already_exists_but_corrupted(
self,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_exists,
mock_mkdir,
mock_open_file,
mock_md5,
mock_remove,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_remove, mock_logger
) -> None:
"""Test download_assets when file exists but is corrupted (re-downloads)."""
filename = "vehicles.mp4"
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
assert result == filename
mock_get.assert_called_once()
mock_copyfileobj.assert_called_once()
mock_download.assert_called_once()
mock_logger.warning.assert_called_once_with("File corrupted. Re-downloading...")
mock_remove.assert_called_once_with(filename)
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader._download_asset")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("pathlib.Path.exists", return_value=False)
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
def test_download_new_file(
self,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_exists,
mock_mkdir,
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_logger
) -> None:
"""Test download_assets verifies a freshly downloaded file."""
filename = "vehicles.mp4"
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
assert result == filename
mock_logger.info.assert_called_with("Downloading %s assets", filename)
mock_get.assert_called_once()
mock_response.raise_for_status.assert_called_once_with()
mock_copyfileobj.assert_called_once()
mock_download.assert_called_once_with(filename, Path.cwd() / filename)
mock_md5.assert_called_once_with(filename, "8155ff4e4de08cfa25f39de96483f918")
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("os.remove")
@patch("supervision.assets.downloader._download_asset")
@patch(
"supervision.assets.downloader.is_md5_hash_matching",
side_effect=[False, True],
)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("pathlib.Path.exists", return_value=False)
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
def test_download_new_file_retries_corrupted_payload(
self,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_exists,
mock_mkdir,
mock_open_file,
mock_md5,
mock_remove,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_remove, mock_logger
) -> None:
"""Test download_assets retries once when a fresh payload fails MD5."""
filename = "vehicles.mp4"
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
assert result == filename
assert mock_get.call_count == 2
assert mock_copyfileobj.call_count == 2
assert mock_download.call_count == 2
mock_remove.assert_called_once_with(filename)
mock_logger.warning.assert_called_once_with("File corrupted. Re-downloading...")
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("os.remove")
@patch("supervision.assets.downloader._download_asset")
@patch(
"supervision.assets.downloader.is_md5_hash_matching",
side_effect=[False, False],
)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("pathlib.Path.exists", return_value=False)
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
def test_download_new_file_raises_after_second_md5_mismatch(
self,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_exists,
mock_mkdir,
mock_open_file,
mock_md5,
mock_remove,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_remove, mock_logger
) -> None:
"""Test download_assets fails after the verified retry is also corrupted."""
filename = "vehicles.mp4"
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
with pytest.raises(ValueError, match="failed MD5 verification"):
download_assets(filename)
assert mock_get.call_count == 2
assert mock_copyfileobj.call_count == 2
assert mock_download.call_count == 2
assert mock_remove.call_count == 2
assert mock_logger.warning.call_count == 2
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
def test_download_new_file_to_custom_directory(
self,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_md5,
mock_logger,
tmp_path,
self, mock_md5, mock_logger, tmp_path
) -> None:
"""Test download_assets writes into an explicit output directory."""
filename = "vehicles.mp4"
target_directory = tmp_path / "nested" / "assets"
response = _mock_streaming_response(b"asset-bytes")
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
mock_copyfileobj.side_effect = lambda _src, file: file.write(b"asset-bytes")
result = download_assets(filename, directory=target_directory)
with patch("supervision.utils.file.requests.get", return_value=response):
result = download_assets(filename, directory=target_directory)
assert result == str(target_directory / filename)
assert (target_directory / filename).exists()
@ -280,41 +161,23 @@ class TestDownloadAssets:
)
@patch("os.replace")
@patch("supervision.assets.downloader.get")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.copyfileobj")
def test_partial_download_does_not_leave_final_file(
self,
mock_copyfileobj,
mock_tqdm,
mock_get,
mock_replace,
tmp_path,
self, mock_replace, tmp_path
) -> None:
"""Test _download_asset stages downloads so failed replaces do not leak."""
filename = "vehicles.mp4"
destination = tmp_path / filename
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
mock_copyfileobj.side_effect = lambda _src, file: file.write(b"partial")
response = _mock_streaming_response(b"partial")
mock_replace.side_effect = OSError("boom")
with pytest.raises(OSError, match="boom"):
with (
patch("supervision.utils.file.requests.get", return_value=response),
pytest.raises(OSError, match="boom"),
):
_download_asset(filename, destination)
mock_copyfileobj.assert_called_once()
assert not destination.exists()
assert not destination.with_name(f"{filename}.part").exists()
assert list(tmp_path.iterdir()) == []
@patch("pathlib.Path.exists", return_value=False)
def test_invalid_asset(self, mock_exists) -> None:
@ -338,82 +201,44 @@ class TestDownloadAssets:
assert "Invalid asset" in str(exc_info.value)
assert "vehicles.mp4" in str(exc_info.value)
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader._download_asset")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
@patch("pathlib.Path.exists", return_value=False)
def test_with_video_enum(
self,
mock_exists,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_mkdir,
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_logger
) -> None:
"""Test download_assets with VideoAssets enum."""
asset = VideoAssets.VEHICLES
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
result = download_assets(asset)
assert result == asset.filename
mock_logger.info.assert_called_with("Downloading %s assets", asset.filename)
mock_download.assert_called_once_with(
asset.filename, Path.cwd() / asset.filename
)
mock_md5.assert_called_once_with(
asset.filename, "8155ff4e4de08cfa25f39de96483f918"
asset.filename, MEDIA_ASSETS[asset.filename][1]
)
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader._download_asset")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@patch("pathlib.Path.mkdir")
@patch("supervision.assets.downloader.copyfileobj")
@patch("supervision.assets.downloader.tqdm")
@patch("supervision.assets.downloader.get")
@patch("pathlib.Path.exists", return_value=False)
def test_with_image_enum(
self,
mock_exists,
mock_get,
mock_tqdm,
mock_copyfileobj,
mock_mkdir,
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
self, mock_exists, mock_md5, mock_download, mock_logger
) -> None:
"""Test download_assets with ImageAssets enum."""
asset = ImageAssets.SOCCER
mock_response = MagicMock()
mock_response.headers = {"Content-Length": "100"}
mock_response.raw = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
result = download_assets(asset)
assert result == asset.filename
mock_logger.info.assert_called_with("Downloading %s assets", asset.filename)
mock_md5.assert_called_once_with(
asset.filename, "0f5a4b98abf3e3973faf9e9260a7d876"
mock_download.assert_called_once_with(
asset.filename, Path.cwd() / asset.filename
)
mock_md5.assert_called_once_with(
asset.filename, MEDIA_ASSETS[asset.filename][1]
)

43
tests/cv2/test_image.py
View File

@ -13,6 +13,8 @@ from supervision._cv2._image import (
_convert_scale_abs,
_copy_make_border,
_flip,
_imdecode,
_imencode,
_imread,
_imwrite,
_mean,
@ -276,6 +278,47 @@ def test_fallback_image_io_preserves_sixteen_bit_unchanged(tmp_path: Path) -> No
)
def test_fallback_in_memory_codec_preserves_bgr() -> None:
"""Preserve BGR channel order across an encode and decode round trip."""
image = np.array([[[10, 20, 30], [40, 50, 60]]], dtype=np.uint8)
success, encoded = _imencode(".png", image)
assert success
assert encoded is not None
decoded = _imdecode(encoded, _IMREAD_COLOR)
assert decoded is not None
np.testing.assert_array_equal(decoded, image)
def test_fallback_imdecode_matches_opencv_for_jpeg() -> None:
"""Decode OpenCV-encoded JPEG bytes identically to cv2.imdecode."""
image = np.full((4, 4, 3), (10, 20, 30), dtype=np.uint8)
encoded = cv2.imencode(".jpg", image)[1]
np.testing.assert_array_equal(
_imdecode(encoded, _IMREAD_COLOR),
cv2.imdecode(encoded, cv2.IMREAD_COLOR),
)
def test_fallback_imdecode_returns_none_for_invalid_bytes() -> None:
"""Return None when decoding bytes that are not an image."""
invalid = np.frombuffer(b"not an image", dtype=np.uint8)
assert _imdecode(invalid, _IMREAD_COLOR) is None
def test_fallback_imencode_reports_failure_for_unknown_extension() -> None:
"""Report failure when encoding to an extension Pillow cannot handle."""
image = np.zeros((2, 2, 3), dtype=np.uint8)
success, encoded = _imencode(".unknown", image)
assert not success
assert encoded is None
def test_fallback_image_io_matches_opencv_color_conversion_for_sixteen_bit(
tmp_path: Path,
) -> None:

98
tests/utils/test_file.py
View File

@ -1,10 +1,106 @@
import os
from contextlib import ExitStack as DoesNotRaise
from pathlib import Path
from unittest.mock import Mock, patch
import pytest
import requests
from supervision.utils.file import (
_download_to_file,
_normalize_http_url,
list_files_with_extensions,
read_txt_file,
)
class TestNormalizeHttpUrl:
def test_returns_normalized_http_url(self) -> None:
"""Valid HTTPS URL is returned normalized."""
# given
url = "https://media.roboflow.com/quickstart/dog.jpeg"
# when
result = _normalize_http_url(url=url)
# then
assert result == url
@pytest.mark.parametrize(
("url", "match"),
[
pytest.param(
"file:///tmp/image.jpg", "Unsupported URL scheme", id="file-scheme"
),
pytest.param(
"ftp://example.com/image.jpg",
"Unsupported URL scheme",
id="ftp-scheme",
),
pytest.param(
"javascript:alert(1)",
"Unsupported URL scheme",
id="javascript-scheme",
),
pytest.param(
"data:text/plain;base64,aGk=",
"Unsupported URL scheme",
id="data-scheme",
),
pytest.param("not a url", "Invalid URL", id="not-a-url"),
pytest.param("http://", "Invalid URL", id="missing-host"),
pytest.param(
"https://foo\\bar/image.jpg", "Invalid URL", id="backslash-authority"
),
],
)
def test_rejects_invalid_url(self, url: str, match: str) -> None:
"""Invalid or non-HTTP(S) URLs raise ValueError."""
with pytest.raises(ValueError, match=match):
_normalize_http_url(url=url)
class TestDownloadToFile:
def test_writes_response_content_to_target(self, tmp_path) -> None:
"""Non-streaming download writes response bytes to the target path."""
# given
target = tmp_path / "subdir" / "file.bin"
response = Mock()
response.content = b"payload"
response.raise_for_status.return_value = None
# when
with patch("supervision.utils.file.requests.get", return_value=response) as get:
_download_to_file("https://example.com/file.bin", target)
# then
get.assert_called_once_with(
"https://example.com/file.bin",
stream=False,
allow_redirects=True,
timeout=30.0,
)
assert target.read_bytes() == b"payload"
assert list(target.parent.iterdir()) == [target]
response.close.assert_called_once()
def test_raises_and_leaves_no_file_on_http_error(self, tmp_path) -> None:
"""HTTP error status raises and leaves no file behind."""
# given
target = tmp_path / "file.bin"
response = Mock()
response.raise_for_status.side_effect = requests.HTTPError("404")
# when / then
with (
patch("supervision.utils.file.requests.get", return_value=response),
pytest.raises(requests.HTTPError, match="404"),
):
_download_to_file("https://example.com/file.bin", target)
assert not target.exists()
response.close.assert_called_once()
from supervision.utils.file import list_files_with_extensions, read_txt_file
FILE_1_CONTENT = """Line 1
Line 2

199
tests/utils/test_image.py
View File

@ -1,7 +1,11 @@
from __future__ import annotations
import warnings
from unittest.mock import Mock, patch
import numpy as np
import pytest
import requests
from PIL import Image, ImageChops
from supervision import _cv2 as cv2
@ -11,6 +15,7 @@ from supervision.utils.image import (
crop_image,
get_image_resolution_wh,
letterbox_image,
load_image_from_url,
overlay_image,
resize_image,
scale_image,
@ -18,6 +23,200 @@ from supervision.utils.image import (
)
class TestLoadImageFromUrl:
def test_returns_decoded_image(self, tmp_path) -> None:
"""Valid image URL returns an OpenCV image."""
# given
image = np.full((10, 20, 3), 127, dtype=np.uint8)
encoded = cv2.imencode(".jpg", image)[1]
response = Mock()
response.content = encoded.tobytes()
response.raise_for_status.return_value = None
# when
with patch("supervision.utils.file.requests.get", return_value=response) as get:
result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
# then
get.assert_called_once_with(
"https://media.roboflow.com/quickstart/dog.jpeg",
stream=False,
allow_redirects=True,
timeout=30.0,
)
assert result.shape == image.shape
assert result.dtype == np.uint8
response.close.assert_called_once()
def test_uses_cached_image_on_repeated_calls(self, tmp_path) -> None:
"""Repeated image URL loads use the local cache."""
# given
image = np.full((10, 20, 3), 127, dtype=np.uint8)
encoded = cv2.imencode(".jpg", image)[1]
response = Mock()
response.content = encoded.tobytes()
response.raise_for_status.return_value = None
# when
with patch("supervision.utils.file.requests.get", return_value=response) as get:
first_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
second_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
# then
get.assert_called_once()
assert first_result.shape == image.shape
assert second_result.shape == image.shape
def test_downloads_each_time_when_cache_is_disabled(self, tmp_path) -> None:
"""Disabled cache skips cache reads and writes."""
# given
image = np.full((10, 20, 3), 127, dtype=np.uint8)
encoded = cv2.imencode(".jpg", image)[1]
first_response = Mock()
first_response.content = encoded.tobytes()
first_response.raise_for_status.return_value = None
second_response = Mock()
second_response.content = encoded.tobytes()
second_response.raise_for_status.return_value = None
# when
with patch(
"supervision.utils.file.requests.get",
side_effect=[first_response, second_response],
) as get:
first_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
use_cache=False,
)
second_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
use_cache=False,
)
# then
assert get.call_count == 2
assert first_result.shape == image.shape
assert second_result.shape == image.shape
assert list(tmp_path.iterdir()) == []
def test_force_reload_refreshes_cached_image(self, tmp_path) -> None:
"""Force reload bypasses the cached image and refreshes it."""
# given
first_image = np.zeros((10, 20, 3), dtype=np.uint8)
second_image = np.full((12, 22, 3), 127, dtype=np.uint8)
first_response = Mock()
first_response.content = cv2.imencode(".jpg", first_image)[1].tobytes()
first_response.raise_for_status.return_value = None
second_response = Mock()
second_response.content = cv2.imencode(".jpg", second_image)[1].tobytes()
second_response.raise_for_status.return_value = None
# when
with patch(
"supervision.utils.file.requests.get",
side_effect=[first_response, second_response],
) as get:
cached_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
refreshed_result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
force_reload=True,
)
# then
assert get.call_count == 2
assert cached_result.shape == first_image.shape
assert refreshed_result.shape == second_image.shape
def test_redownloads_when_cached_image_is_invalid(self, tmp_path) -> None:
"""Invalid cached image bytes are discarded and downloaded again."""
# given
image = np.full((10, 20, 3), 127, dtype=np.uint8)
first_response = Mock()
first_response.content = cv2.imencode(".jpg", image)[1].tobytes()
first_response.raise_for_status.return_value = None
second_response = Mock()
second_response.content = cv2.imencode(".jpg", image)[1].tobytes()
second_response.raise_for_status.return_value = None
# when
with patch(
"supervision.utils.file.requests.get",
side_effect=[first_response, second_response],
) as get:
load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
for cache_file in tmp_path.iterdir():
cache_file.write_bytes(b"not an image")
result = load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
# then
assert get.call_count == 2
assert result.shape == image.shape
def test_raises_when_bytes_are_not_image(self, tmp_path) -> None:
"""Invalid image bytes raise ValueError."""
# given
response = Mock()
response.content = b"not an image"
response.raise_for_status.return_value = None
# when / then
with (
patch("supervision.utils.file.requests.get", return_value=response),
pytest.raises(ValueError, match="could not be decoded into image"),
):
load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
response.close.assert_called_once()
def test_raises_for_request_error(self, tmp_path) -> None:
"""Request failures are propagated."""
# given
request_error = requests.RequestException("boom")
# when / then
with (
patch("supervision.utils.file.requests.get", side_effect=request_error),
pytest.raises(requests.RequestException, match="boom"),
):
load_image_from_url(
"https://media.roboflow.com/quickstart/dog.jpeg",
cache_dir=tmp_path,
)
def test_rejects_non_http_url(self) -> None:
"""Non-HTTP URLs are rejected before making a request."""
# given
with patch("supervision.utils.file.requests.get") as get:
# when / then
with pytest.raises(ValueError, match="HTTP"):
load_image_from_url("file:///tmp/image.jpg")
get.assert_not_called()
def test_resize_image_for_opencv_image() -> None:
# given
image = np.zeros((480, 640, 3), dtype=np.uint8)