diff --git a/docs/changelog.md b/docs/changelog.md index adedcb3fd..ee3adf5c0 100644 --- a/docs/changelog.md +++ b/docs/changelog.md @@ -1,6 +1,6 @@ --- description: "Full version history of the supervision Python library — release notes, breaking changes, new features, and deprecations for every version." -date_modified: 2026-07-08 +date_modified: 2026-07-09 --- # Changelog @@ -46,6 +46,7 @@ date_modified: 2026-07-08 - Fixed: dataset IO/export edge cases now avoid mutating caller-owned `Detections` during `DetectionDataset` construction, reject non-integer and out-of-range class ids with a clear `ValueError`, load COCO annotations that omit optional `iscrowd`/`area` fields, expose `DetectionDataset.from_coco(use_iscrowd=...)` without changing the existing positional `show_progress` argument, export mask pixel area to COCO when no stored area is present, ignore folder-structure root clutter and non-image files inside class folders, and accept PIL-readable YOLO images such as RGBA or palette PNGs. ### 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)) - `KeyPoints.merge` — combine a list of `KeyPoints` objects into one, mirroring `Detections.merge`. Empty inputs are ignored; all non-empty inputs must share the same number of keypoints per skeleton. Completes the merge-then-suppress workflow introduced by `KeyPoints.with_nms` ([#2412](https://github.com/roboflow/supervision/pull/2412)) - `BaseAnnotator.requires_mask` — class-level `bool` flag on all annotators; `True` for `MaskAnnotator`, `PolygonAnnotator`, and `HaloAnnotator`; `False` for all others. Integrations can inspect this before materializing expensive mask payloads ([#2370](https://github.com/roboflow/supervision/pull/2370)) - `CompactMask.from_coco_rle` — efficient COCO RLE ingestion into crop-scoped compact mask format without materializing dense `(N, H, W)` arrays ([#2367](https://github.com/roboflow/supervision/pull/2367)) diff --git a/docs/utils/image.md b/docs/utils/image.md index 9d1c1895c..0b61abb3c 100644 --- a/docs/utils/image.md +++ b/docs/utils/image.md @@ -11,6 +11,12 @@ status: new :::supervision.utils.image.crop_image +
+

load_image_from_url

+
+ +:::supervision.utils.image.load_image_from_url +

scale_image

diff --git a/src/supervision/__init__.py b/src/supervision/__init__.py index a3f299991..8e1c72b75 100644 --- a/src/supervision/__init__.py +++ b/src/supervision/__init__.py @@ -145,6 +145,7 @@ get_image_resolution_wh, grayscale_image, letterbox_image, + load_image_from_url, overlay_image, resize_image, scale_image, @@ -263,6 +264,7 @@ "is_valid_hex", "letterbox_image", "list_files_with_extensions", + "load_image_from_url", "mask_iou_batch", "mask_non_max_merge", "mask_non_max_suppression", diff --git a/src/supervision/assets/downloader.py b/src/supervision/assets/downloader.py index 99387df6c..bd0ab9653 100644 --- a/src/supervision/assets/downloader.py +++ b/src/supervision/assets/downloader.py @@ -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( diff --git a/src/supervision/utils/file.py b/src/supervision/utils/file.py index 00466e530..998218204 100644 --- a/src/supervision/utils/file.py +++ b/src/supervision/utils/file.py @@ -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: diff --git a/src/supervision/utils/image.py b/src/supervision/utils/image.py index 2fa1372c2..d03159b63 100644 --- a/src/supervision/utils/image.py +++ b/src/supervision/utils/image.py @@ -4,8 +4,12 @@ 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 @@ 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 = ( + 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( diff --git a/tests/assets/test_downloader.py b/tests/assets/test_downloader.py index f3c03cd44..9c421c67e 100644 --- a/tests/assets/test_downloader.py +++ b/tests/assets/test_downloader.py @@ -1,3 +1,5 @@ +import io +from pathlib import Path from unittest.mock import MagicMock, mock_open, patch import pytest @@ -7,7 +9,15 @@ 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 @@ def test_file_not_exists(self) -> None: 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 @@ def test_download_new_file_to_custom_directory( ) @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 @@ def test_invalid_asset_when_file_exists(self, mock_exists) -> None: 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_download.assert_called_once_with( + asset.filename, Path.cwd() / asset.filename + ) mock_md5.assert_called_once_with( - asset.filename, "0f5a4b98abf3e3973faf9e9260a7d876" + asset.filename, MEDIA_ASSETS[asset.filename][1] ) diff --git a/tests/utils/test_file.py b/tests/utils/test_file.py index 62ca0addc..b557105b9 100644 --- a/tests/utils/test_file.py +++ b/tests/utils/test_file.py @@ -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 diff --git a/tests/utils/test_image.py b/tests/utils/test_image.py index 9b3ea5308..600f126f5 100644 --- a/tests/utils/test_image.py +++ b/tests/utils/test_image.py @@ -1,8 +1,12 @@ +from __future__ import annotations + import warnings +from unittest.mock import Mock, patch import cv2 import numpy as np import pytest +import requests from PIL import Image, ImageChops from supervision.utils.image import ( @@ -11,6 +15,7 @@ crop_image, get_image_resolution_wh, letterbox_image, + load_image_from_url, overlay_image, resize_image, scale_image, @@ -18,6 +23,200 @@ ) +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)