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

fix: remaining review findings in dataset, docs, and tests (#2416) 

- Added `sv.mask_to_roi` as an explicit migration path for exclusive mask bounds
- Fixed COCO, CreateML, and Pascal VOC export validation to reject ambiguous or colliding dataset paths before writing
- Fixed in-memory `DetectionDataset` split and merge behavior
- Fixed `supervision` imports to avoid loading ByteTrack until it is used
- Fixed detection conversion helpers to support coordinate-convention migration while preserving legacy inclusive defaults
- Fixed Azure tag mapping, anchor rounding, and line-zone smoothing to avoid incorrect or ghost detections
- Fixed video processing shutdown handling for timeout and full-queue cases
- Improved downloader, validator, documentation, and regression coverage for the shipped dataset, detection, annotator, image, and video behavior

---------

Co-authored-by: Codex <codex@openai.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Jirka Borovec 2026-07-08 23:00:32 +02:00 committed by GitHub
parent 23a2227ae7
commit 75023c5f2f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
64 changed files with 1978 additions and 266 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
docs/changelog.md
View File

@ -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-06
date_modified: 2026-07-08
---
# Changelog
@ -18,6 +18,11 @@ date_modified: 2026-07-06
- `sv.mask_non_max_merge` now computes exact mask overlap at the original mask resolution and ignores the deprecated `mask_dimension` parameter. Code that relied on downscaled mask overlap should recalibrate thresholds; passing `mask_dimension` positionally now emits a deprecation warning, and the parameter is scheduled for removal in `0.33.0` ([#2400](https://github.com/roboflow/supervision/pull/2400)).
### Fixed
- Fixed [#2416](https://github.com/roboflow/supervision/pull/2416): `sv.process_video` no longer risks hanging during shutdown; the sentinel enqueue is best-effort and worker joins are bounded.
- Fixed [#2416](https://github.com/roboflow/supervision/pull/2416): COCO and CreateML dataset loaders now canonicalize resolved image paths and reject duplicate aliases for the same file.
- Fixed [#2416](https://github.com/roboflow/supervision/pull/2416): `DetectionDataset.as_pascal_voc()` now preflights image and annotation basename collisions before writing, so exports fail fast instead of producing partial output.
- `import supervision` no longer surfaces the deprecated `ByteTrack` warning; the top-level tracker alias now resolves lazily when accessed explicitly.
- Fixed dataset export edge cases: `DetectionDataset.split()` and `DetectionDataset.merge()` now preserve in-memory image payloads without re-emitting the deprecation warning, and COCO/CreateML exports now reject duplicate image basenames instead of silently collapsing distinct paths into the same output key.
- Fixed: `sv.Color(...)` now validates direct RGBA channel values and raises `ValueError` when any channel falls outside the 0-255 byte range.
- Fixed: `approximate_mask_with_polygons` now defaults to no polygon simplification, matching the public dataset export methods.
- Fixed: `ImageSink.save_image()` now raises `OSError` when `cv2.imwrite()` fails, and deprecation-warning control accepts the correct `SUPERVISION_DEPRECATION_WARNING` environment variable while still honoring the legacy misspelled alias.
@ -41,6 +46,7 @@ date_modified: 2026-07-06
- `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))
- `Detections.from_inference(compact_masks=True)` — opt-in compact mask representation for Roboflow/Inference segmentation results; masks are cropped to detector bounding boxes ([#2367](https://github.com/roboflow/supervision/pull/2367))
- `CompactMask.image_shape` — new public property returning `(H, W)` of the full image the mask is scoped to ([#2383](https://github.com/roboflow/supervision/pull/2383))
- `sv.mask_to_roi` — explicit exclusive mask-bound helper for NumPy slicing and crop extraction. `sv.mask_to_xyxy` stays inclusive for compatibility with CompactMask and current box-based adapters, so the coordinate-convention migration path is now explicit instead of implicit.
### Changed
- Performance [#2383](https://github.com/roboflow/supervision/pull/2383): `sv.Detections.merge()` on mixed dense `ndarray` + `CompactMask` inputs now returns a `CompactMask` instead of a dense `ndarray`. Previously (0.29.0/0.29.1) the mixed path fell back to `np.vstack`, allocating a full `(N, H, W)` array; the new path converts dense inputs to `CompactMask` without materialising the full stack (~2 500× less peak memory, ~13× faster on 1080p / 40 detections). **Behavior change**: code that checks `isinstance(merged.mask, np.ndarray)` or calls bare ndarray methods (`.astype`, `.reshape`, `.ravel`) on a mixed-merge result will need to be updated. The all-dense path is unchanged and still returns `ndarray`.

6
docs/detection/utils/masks.md
View File

@ -5,6 +5,12 @@ status: new
# Masks Utils
<div class="md-typeset">
<h2><a href="#supervision.detection.utils.masks.mask_to_roi">mask_to_roi</a></h2>
</div>
:::supervision.detection.utils.masks.mask_to_roi
<div class="md-typeset">
<h2><a href="#supervision.detection.utils.masks.move_masks">move_masks</a></h2>
</div>

4
docs/how_to/use_compact_masks.md
View File

@ -12,6 +12,10 @@ date_modified: 2026-07-01
[CompactMask][supervision.detection.compact_mask.CompactMask] stores each instance mask as a run-length encoding of its bounding-box **crop** rather than a full `(H, W)` boolean frame. For high-resolution images with many sparse masks this can reduce memory from tens of gigabytes to tens of megabytes, and eliminates full-frame decode work in annotators that only need the cropped region.
!!! Note
`sv.mask_to_xyxy` keeps supervision's inclusive max-coordinate convention for compatibility with `CompactMask` and current box-based adapters. Use `sv.mask_to_roi` when you need exclusive slice bounds for NumPy indexing or crop extraction.
This guide covers the four main integration points:
1. [Ingesting COCO RLE payloads directly as CompactMask](#ingest-coco-rle-payloads)

2
docs/keypoint/annotators.md
View File

@ -95,6 +95,8 @@ comments: true
)
```
`sv.VertexEllipseAnnotator` is a compatibility alias for `sv.VertexEllipseAreaAnnotator`.
=== "VertexEllipseOutlineAnnotator"
```python

1
pyproject.toml
View File

@ -195,6 +195,7 @@ strict = true
overrides = [ { module = [ "examples.*", "tests.*" ], ignore_errors = true } ]
[tool.pytest]
ini_options.testpaths = [ "src", "tests" ]
ini_options.norecursedirs = [ ".git", ".venv", "build", "dist", "docs", "examples", "notebooks" ]
ini_options.addopts = [
"--doctest-modules",

17
src/supervision/__init__.py
View File

@ -1,4 +1,5 @@
import importlib.metadata as importlib_metadata
from typing import TYPE_CHECKING, Any
try:
# This will read version from pyproject.toml
@ -102,6 +103,7 @@ from supervision.detection.utils.masks import (
contains_holes,
contains_multiple_segments,
filter_segments_by_distance,
mask_to_roi,
move_masks,
)
from supervision.detection.utils.polygons import (
@ -135,7 +137,6 @@ from supervision.key_points.annotators import (
)
from supervision.key_points.core import KeyPoints
from supervision.metrics.detection import ConfusionMatrix, MeanAveragePrecision
from supervision.tracker.byte_tracker.core import ByteTrack
from supervision.utils.conversion import cv2_to_pillow, pillow_to_cv2
from supervision.utils.file import list_files_with_extensions
from supervision.utils.image import (
@ -158,6 +159,9 @@ from supervision.utils.video import (
process_video,
)
if TYPE_CHECKING:
from supervision.tracker.byte_tracker.core import ByteTrack
__all__ = [
"LMM",
"VLM",
@ -264,6 +268,7 @@ __all__ = [
"mask_non_max_suppression",
"mask_to_polygons",
"mask_to_rle",
"mask_to_roi",
"mask_to_xyxy",
"move_boxes",
"move_masks",
@ -292,3 +297,13 @@ __all__ = [
"xyxy_to_xywh",
"xyxyxyxy_to_xyxy",
]
def __getattr__(name: str) -> Any:
"""Lazily resolve deprecated compatibility exports."""
if name == "ByteTrack":
from supervision.tracker.byte_tracker.core import ByteTrack as byte_track
globals()[name] = byte_track
return byte_track
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

25
src/supervision/annotators/core.py
View File

@ -1367,11 +1367,11 @@ class LabelAnnotator(_BaseLabelAnnotator):
@ensure_cv2_image_for_class_method
def annotate(
self,
scene: Image.Image,
scene: ImageType,
detections: Detections,
labels: list[str] | None = None,
custom_color_lookup: npt.NDArray[np.int_] | None = None,
) -> Image.Image:
) -> ImageType:
"""
Annotates the given scene with labels based on the provided detections.
@ -1426,10 +1426,6 @@ class LabelAnnotator(_BaseLabelAnnotator):
)
if self.smart_position:
xyxy = label_properties[:, :4]
xyxy = cast(npt.NDArray[np.float32], spread_out_boxes(xyxy))
label_properties[:, :4] = xyxy
label_properties = self._adjust_labels_in_frame(
(scene.shape[1], scene.shape[0]),
labels,
@ -1723,11 +1719,11 @@ class RichLabelAnnotator(_BaseLabelAnnotator):
@ensure_pil_image_for_class_method
def annotate(
self,
scene: Image.Image,
scene: ImageType,
detections: Detections,
labels: list[str] | None = None,
custom_color_lookup: npt.NDArray[np.int_] | None = None,
) -> Image.Image:
) -> ImageType:
"""
Annotates the given scene with labels based on the provided
detections, with support for Unicode characters.
@ -1780,12 +1776,9 @@ class RichLabelAnnotator(_BaseLabelAnnotator):
)
if self.smart_position:
xyxy = label_properties[:, :4]
xyxy = cast(npt.NDArray[np.float32], spread_out_boxes(xyxy))
label_properties[:, :4] = xyxy
scene_pil = cast(Image.Image, scene)
label_properties = self._adjust_labels_in_frame(
(scene.width, scene.height),
(scene_pil.width, scene_pil.height),
labels,
label_properties,
)
@ -2357,7 +2350,7 @@ class HeatMapAnnotator(BaseAnnotator):
"""
if not isinstance(scene, np.ndarray):
return scene
if self.heat_mask is None:
if self.heat_mask is None or self.heat_mask.shape != scene.shape[:2]:
self.heat_mask = np.zeros(scene.shape[:2], dtype=np.float32)
mask: npt.NDArray[np.float32] = np.zeros(scene.shape[:2], dtype=np.float32)
@ -2888,6 +2881,7 @@ class PercentageBarAnnotator(BaseAnnotator):
def calculate_border_coordinates(
anchor_xy: tuple[int, int], border_wh: tuple[int, int], position: Position
) -> tuple[tuple[int, int], tuple[int, int]]:
"""Compute the border corner coordinates for a given anchor position."""
cx, cy = anchor_xy
width, height = border_wh
@ -2912,6 +2906,7 @@ class PercentageBarAnnotator(BaseAnnotator):
return (cx - width // 2, cy), (cx + width // 2, cy + height)
elif position == Position.BOTTOM_RIGHT:
return (cx, cy), (cx + width, cy + height)
raise ValueError(f"Unsupported position: {position}")
@staticmethod
def _validate_custom_values(
@ -3083,6 +3078,7 @@ class CropAnnotator(BaseAnnotator):
def calculate_crop_coordinates(
anchor: tuple[int, int], crop_wh: tuple[int, int], position: Position
) -> tuple[tuple[int, int], tuple[int, int]]:
"""Compute the crop coordinates for a given anchor position."""
anchor_x, anchor_y = anchor
width, height = crop_wh
@ -3119,6 +3115,7 @@ class CropAnnotator(BaseAnnotator):
)
elif position == Position.BOTTOM_RIGHT:
return (anchor_x, anchor_y), (anchor_x + width, anchor_y + height)
raise ValueError(f"Unsupported position: {position}")
class BackgroundOverlayAnnotator(BaseAnnotator):

15
src/supervision/annotators/utils.py
View File

@ -80,6 +80,7 @@ def resolve_text_background_xyxy(
text_wh: tuple[int, int],
position: Position,
) -> tuple[int, int, int, int]:
"""Compute the background box for text anchored at `position`."""
center_x, center_y = center_coordinates
text_w, text_h = text_wh
@ -126,6 +127,7 @@ def resolve_text_background_xyxy(
center_x + text_w,
center_y + text_h // 2,
)
raise ValueError(f"Unsupported position: {position}")
def get_color_by_index(color: Color | ColorPalette, idx: int) -> Color:
@ -353,6 +355,13 @@ class Trace:
self.tracker_id: npt.NDArray[np.int_] = np.array([], dtype=int)
def put(self, detections: Detections) -> None:
"""Append a frame of detections to the trace history."""
if detections.tracker_id is None:
raise ValueError(
"Could not put detections into Trace because "
"Detections do not have tracker_id."
)
frame_id: npt.NDArray[np.int_] = np.full(
len(detections), self.current_frame_id, dtype=int
)
@ -363,12 +372,6 @@ class Trace:
detections.get_anchors_coordinates(self.anchor),
]
)
if detections.tracker_id is None:
raise ValueError(
"Could not put detections into Trace because "
"Detections do not have tracker_id."
)
self.tracker_id = np.concatenate([self.tracker_id, detections.tracker_id])
unique_frame_id = np.unique(self.frame_id)

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

@ -12,7 +12,7 @@ from supervision.utils.logger import _get_logger
logger = _get_logger(__name__)
def is_md5_hash_matching(filename: str, original_md5_hash: str) -> bool:
def is_md5_hash_matching(filename: str | Path, original_md5_hash: str) -> bool:
"""
Check if the MD5 hash of a file matches the original hash.
@ -36,9 +36,9 @@ def is_md5_hash_matching(filename: str, original_md5_hash: str) -> bool:
return computed_md5_hash.hexdigest() == original_md5_hash
def _download_asset(filename: str) -> None:
def _download_asset(filename: str, destination: Path) -> None:
"""
Download asset bytes to the target filename.
Download asset bytes to the target destination via a temporary file.
"""
response = get(
MEDIA_ASSETS[filename][0], stream=True, allow_redirects=True, timeout=30
@ -46,34 +46,49 @@ def _download_asset(filename: str) -> None:
response.raise_for_status()
file_size = int(response.headers.get("Content-Length", 0))
folder_path = Path(filename).expanduser().resolve()
folder_path.parent.mkdir(parents=True, exist_ok=True)
destination.parent.mkdir(parents=True, exist_ok=True)
temp_path = destination.with_name(f"{destination.name}.part")
with tqdm.wrapattr(
response.raw, "read", total=file_size, desc="", colour="#a351fb"
) as raw_resp:
with folder_path.open("wb") as file:
copyfileobj(raw_resp, file)
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)
def _download_verified_asset(
filename: str, original_md5_hash: str, retry_on_mismatch: bool = True
filename: str,
original_md5_hash: str,
destination: Path,
check_target: str | Path,
retry_on_mismatch: bool = True,
) -> None:
"""
Download an asset and reject payloads whose MD5 does not match the catalog.
"""
_download_asset(filename)
_download_asset(filename, destination)
if is_md5_hash_matching(filename, original_md5_hash):
if is_md5_hash_matching(check_target, original_md5_hash):
return
logger.warning("File corrupted. Re-downloading...")
os.remove(filename)
os.remove(check_target)
if retry_on_mismatch:
_download_verified_asset(
filename=filename,
original_md5_hash=original_md5_hash,
destination=destination,
check_target=check_target,
retry_on_mismatch=False,
)
return
@ -81,15 +96,21 @@ def _download_verified_asset(
raise ValueError(f"Downloaded asset {filename!r} failed MD5 verification.")
def download_assets(asset_name: Assets | str) -> str:
def download_assets(
asset_name: Assets | str,
directory: str | Path | None = None,
) -> str:
"""
Download a specified asset if it doesn't already exist or is corrupted.
Args:
asset_name: The name or type of the asset to be downloaded.
directory: Optional output directory. Defaults to the current working
directory for backward compatibility.
Returns:
The filename of the downloaded asset.
The downloaded asset path. When `directory` is omitted, this preserves
the historical filename-only return value.
Example:
```pycon
@ -104,17 +125,36 @@ def download_assets(asset_name: Assets | str) -> str:
"""
filename = asset_name.filename if isinstance(asset_name, Assets) else asset_name
if directory is None:
destination = Path.cwd() / filename
check_target: str | Path = filename
return_value = filename
else:
destination_directory = Path(directory).expanduser().resolve()
destination = destination_directory / filename
check_target = str(destination)
return_value = str(destination)
if filename in MEDIA_ASSETS:
original_md5_hash = MEDIA_ASSETS[filename][1]
if not Path(filename).exists():
if not Path(check_target).exists():
logger.info("Downloading %s assets", filename)
_download_verified_asset(filename, original_md5_hash)
_download_verified_asset(
filename=filename,
original_md5_hash=original_md5_hash,
destination=destination,
check_target=check_target,
)
else:
if not is_md5_hash_matching(filename, original_md5_hash):
if not is_md5_hash_matching(check_target, original_md5_hash):
logger.warning("File corrupted. Re-downloading...")
os.remove(filename)
_download_verified_asset(filename, original_md5_hash)
os.remove(check_target)
_download_verified_asset(
filename=filename,
original_md5_hash=original_md5_hash,
destination=destination,
check_target=check_target,
)
logger.info("%s asset download complete.", filename)
else:
@ -123,4 +163,4 @@ def download_assets(asset_name: Assets | str) -> str:
f"Invalid asset. It should be one of the following: {valid_assets}."
)
return filename
return return_value

36
src/supervision/dataset/core.py
View File

@ -252,27 +252,26 @@ class DetectionDataset(BaseDataset):
shuffle=shuffle,
)
train_input: list[str] | dict[str, npt.NDArray[np.uint8]]
test_input: list[str] | dict[str, npt.NDArray[np.uint8]]
if self._images_in_memory:
train_input = {path: self._images_in_memory[path] for path in train_paths}
test_input = {path: self._images_in_memory[path] for path in test_paths}
else:
train_input = train_paths
test_input = test_paths
train_annotations = {path: self.annotations[path] for path in train_paths}
test_annotations = {path: self.annotations[path] for path in test_paths}
train_dataset = DetectionDataset(
classes=self.classes,
images=train_input,
images=train_paths,
annotations=train_annotations,
)
test_dataset = DetectionDataset(
classes=self.classes,
images=test_input,
images=test_paths,
annotations=test_annotations,
)
if self._images_in_memory:
train_dataset._images_in_memory = {
path: self._images_in_memory[path] for path in train_paths
}
test_dataset._images_in_memory = {
path: self._images_in_memory[path] for path in test_paths
}
return train_dataset, test_dataset
@classmethod
@ -369,11 +368,14 @@ class DetectionDataset(BaseDataset):
detections=annotations[image_path],
)
return cls(
merged_dataset = cls(
classes=classes,
images=images_in_memory or image_paths,
images=image_paths,
annotations=annotations,
)
if all_in_memory:
merged_dataset._images_in_memory = images_in_memory
return merged_dataset
def as_pascal_voc(
self,
@ -386,7 +388,9 @@ class DetectionDataset(BaseDataset):
) -> None:
"""
Exports the dataset to PASCAL VOC format. This method saves the images
and their corresponding annotations in PASCAL VOC format.
and their corresponding annotations in PASCAL VOC format. Both output
layouts are preflighted before any files are written so a collision in
either target fails without partial output.
Args:
images_directory_path: The path to the directory
@ -421,6 +425,12 @@ class DetectionDataset(BaseDataset):
key=lambda image_path: Path(image_path).name,
output_kind="image",
)
if annotations_directory_path:
check_no_basename_collisions(
image_paths=self.image_paths,
key=lambda image_path: f"{Path(image_path).stem}.xml",
output_kind="Pascal VOC annotation",
)
if images_directory_path:
save_dataset_images(

30
src/supervision/dataset/formats/coco.py
View File

@ -12,6 +12,7 @@ from tqdm.auto import tqdm
from supervision.config import COCO_RAW_SEGMENTATION
from supervision.dataset.utils import (
approximate_mask_with_polygons,
check_no_basename_collisions,
map_detections_class_id,
)
from supervision.detection.core import Detections
@ -443,13 +444,15 @@ def load_coco_annotations(
show_progress: If `True`, display a progress bar during loading.
Returns:
A tuple of `(classes, image_paths, annotations)`.
A tuple of `(classes, image_paths, annotations)` where image paths are
canonicalized resolved paths inside ``images_directory_path``.
Raises:
ValueError: If any annotation's ``file_name`` resolves to the images
directory itself, to a path outside the images directory (e.g. via
``../`` traversal or an absolute path), or to a subdirectory instead
of a regular image file.
ValueError: If two image entries resolve to the same canonical path.
Note:
Each annotation's ``file_name`` is validated against
@ -513,6 +516,12 @@ def load_coco_annotations(
f"resolves to directory {resolved_image_path}. Expected a "
"path to an image file."
)
image_path = str(resolved_image_path)
if image_path in annotations:
raise ValueError(
f"COCO annotation file contains duplicate entries for image "
f"{image_name!r}. Each image must appear at most once."
)
with_masks = force_masks or any(
_with_seg_mask(annotation) for annotation in image_annotations
@ -578,12 +587,14 @@ def save_coco_annotations(
.. note::
This function ensures globally unique integer ``id`` values across
splits. It does **not** ensure unique ``file_name`` values — the
``file_name`` field is set to the bare image basename, so splits
that share filenames (e.g. ``000001.jpg`` in both train and valid)
will have duplicate ``file_name`` values when their COCO files are
merged. Use distinct output directories or rename images before
merging if downstream tools require unique ``file_name`` keys.
splits. It rejects duplicate image basenames before writing because
``file_name`` is set to the bare image basename, so two input paths
that differ only by directory would otherwise collapse to the same
COCO image record.
Raises:
ValueError: If two image paths share the same basename and would map to
the same COCO ``file_name``.
Example:
```python
@ -608,6 +619,11 @@ def save_coco_annotations(
"(COCO spec requires 1-indexed ids); "
f"got {starting_image_id=}, {starting_annotation_id=}"
)
check_no_basename_collisions(
image_paths=dataset.image_paths,
key=lambda image_path: Path(image_path).name,
output_kind="COCO image",
)
Path(annotation_path).parent.mkdir(parents=True, exist_ok=True)
licenses = [
{

29
src/supervision/dataset/formats/createml.py
View File

@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, cast
import numpy as np
from tqdm.auto import tqdm
from supervision.dataset.utils import check_no_basename_collisions
from supervision.detection.core import Detections
from supervision.utils.file import read_json_file, save_json_file
@ -20,7 +21,8 @@ def _resolve_image_path(images_directory_path: str, image_name: str) -> str:
Rejects annotations whose ``image`` field escapes ``images_directory_path``
(via ``..`` traversal, an absolute path, or a symlink pointing outside),
mirroring the protection used by the COCO loader.
mirroring the protection used by the COCO loader. Returns the canonical
resolved path so aliases collapse to a single dataset entry.
"""
images_directory_resolved = Path(images_directory_path).resolve()
image_path = Path(images_directory_path) / Path(image_name)
@ -49,7 +51,7 @@ def _resolve_image_path(images_directory_path: str, image_name: str) -> str:
f"resolves to directory {resolved_image_path}. Expected a path "
"to an image file."
)
return str(image_path)
return str(resolved_image_path)
def createml_annotations_to_detections(
@ -149,16 +151,16 @@ def load_createml_annotations(
- ``classes`` (``list[str]``): globally sorted class names inferred from
all labels present in the file.
- ``image_paths`` (``list[str]``): joined (but not fully resolved) path
for every entry in the JSON, in file order.
- ``annotations`` (``dict[str, Detections]``): mapping from joined image
path to its ``Detections``.
- ``image_paths`` (``list[str]``): canonical resolved path for every
entry in the JSON, in file order.
- ``annotations`` (``dict[str, Detections]``): mapping from canonical
resolved image path to its ``Detections``.
Raises:
ValueError: If the JSON root is not a list.
ValueError: If an entry is missing the required ``"image"`` key.
ValueError: If an annotation is missing required coordinate or label keys.
ValueError: If the same image filename appears more than once in the file.
ValueError: If two entries resolve to the same image path.
ValueError: If an annotation's ``image`` field resolves to the images
directory itself or to a path outside it (e.g. via ``..`` traversal
or an absolute path).
@ -287,14 +289,18 @@ def save_createml_annotations(
``"img.jpg"`` rather than ``"/data/train/img.jpg"``). This matches CreateML
convention and means the loader reconstructs paths relative to
``images_directory_path``. As a consequence, two images with the same
basename from different directories will produce duplicate ``"image"`` keys
in the output and cannot be round-tripped correctly.
basename from different directories would collapse to the same ``"image"``
key, so the exporter rejects that case before writing.
Args:
dataset: The ``DetectionDataset`` to write.
annotations_path: Output path for the CreateML JSON file. Parent
directories are created if they do not already exist.
Raises:
ValueError: If two image paths share the same basename and would map to
the same CreateML ``image`` entry.
Examples:
```python
import supervision as sv
@ -304,6 +310,11 @@ def save_createml_annotations(
save_createml_annotations(dataset, "/tmp/annotations.json")
```
"""
check_no_basename_collisions(
image_paths=dataset.image_paths,
key=lambda image_path: Path(image_path).name,
output_kind="CreateML image",
)
Path(annotations_path).parent.mkdir(parents=True, exist_ok=True)
createml_data: list[CreateMLDict] = [
{

170
src/supervision/detection/core.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
import warnings
from collections.abc import Iterator
from dataclasses import dataclass, field
from functools import reduce
@ -32,6 +33,7 @@ from supervision.detection.utils.boxes import (
from supervision.detection.utils.converters import (
mask_to_xyxy,
polygon_to_mask,
rle_to_mask,
xywh_to_xyxy,
)
from supervision.detection.utils.internal import (
@ -72,7 +74,11 @@ from supervision.detection.vlm import (
from_qwen_3_vl,
)
from supervision.geometry.core import Position
from supervision.utils.internal import get_instance_variables, warn_deprecated
from supervision.utils.internal import (
SupervisionWarnings,
get_instance_variables,
warn_deprecated,
)
from supervision.validators import (
_validate_data,
_validate_detections_fields,
@ -815,12 +821,33 @@ class Detections:
sorted_generated_masks = sorted(
sam_result, key=lambda x: x["area"], reverse=True
)
if len(sorted_generated_masks) == 0:
return cls.empty()
xywh = np.array([mask["bbox"] for mask in sorted_generated_masks])
mask = np.array([mask["segmentation"] for mask in sorted_generated_masks])
segmentations = [mask["segmentation"] for mask in sorted_generated_masks]
first_segmentation = segmentations[0]
if np.asarray(xywh).shape[0] == 0:
return cls.empty()
if all(isinstance(segmentation, np.ndarray) for segmentation in segmentations):
mask = np.stack(segmentations, axis=0)
elif all(isinstance(segmentation, dict) for segmentation in segmentations):
image_height, image_width = cast(
tuple[int, int], tuple(int(v) for v in first_segmentation["size"])
)
mask = np.stack(
[
rle_to_mask(
segmentation["counts"],
(image_width, image_height),
)
for segmentation in segmentations
],
axis=0,
)
else:
raise ValueError(
"SAM segmentations must all be dense arrays or COCO RLE dictionaries."
)
xyxy = xywh_to_xyxy(xywh=xywh)
return cls(xyxy=xyxy, mask=mask)
@ -1014,19 +1041,36 @@ class Detections:
x1 = x0 + bbox["w"]
y1 = y0 + bbox["h"]
for tag in tags:
confidence = tag["confidence"]
class_name: str = tag["name"]
class_id_val: int | None = inverted_map.get(class_name, None)
selected_tag: dict[str, Any] | None = None
selected_class_id: int | None = None
for tag in sorted(
tags, key=lambda candidate: candidate["confidence"], reverse=True
):
class_name = tag["name"]
class_id_val = inverted_map.get(class_name, None)
if is_dynamic_mapping and class_id_val is None:
class_id_val = len(inverted_map)
inverted_map[class_name] = class_id_val
if class_id_val is not None:
xyxy.append([x0, y0, x1, y1])
confidences.append(confidence)
class_ids.append(class_id_val)
selected_tag = tag
selected_class_id = class_id_val
break
if selected_tag is None:
if tags:
warnings.warn(
"Azure detection skipped because none of its tags matched "
"the provided class_map.",
category=SupervisionWarnings,
stacklevel=2,
)
continue
xyxy.append([x0, y0, x1, y1])
confidences.append(selected_tag["confidence"])
class_ids.append(cast(int, selected_class_id))
if len(xyxy) == 0:
return Detections.empty()
@ -1988,7 +2032,10 @@ class Detections:
vlm = _validate_vlm_parameters(vlm, result, kwargs)
if vlm == VLM.PALIGEMMA:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
xyxy, class_id, class_name = from_paligemma(result, **kwargs)
data: _DetectionDataType = {
CLASS_NAME_DATA_FIELD: class_name,
@ -1996,7 +2043,10 @@ class Detections:
return cls(xyxy=xyxy, class_id=class_id, data=data)
if vlm == VLM.QWEN_2_5_VL:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
xyxy, class_id, class_name = from_qwen_2_5_vl(result, **kwargs)
data = {CLASS_NAME_DATA_FIELD: class_name}
confidence_arr: npt.NDArray[np.floating[Any]] = np.ones(
@ -2007,7 +2057,10 @@ class Detections:
)
if vlm == VLM.QWEN_3_VL:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
xyxy, class_id, class_name = from_qwen_3_vl(result, **kwargs)
data = {CLASS_NAME_DATA_FIELD: class_name}
confidence_arr = np.ones(len(xyxy), dtype=float)
@ -2016,13 +2069,19 @@ class Detections:
)
if vlm == VLM.DEEPSEEK_VL_2:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
xyxy, class_id, class_name = from_deepseek_vl_2(result, **kwargs)
data = {CLASS_NAME_DATA_FIELD: class_name}
return cls(xyxy=xyxy, class_id=class_id, data=data)
if vlm == VLM.FLORENCE_2:
assert isinstance(result, dict)
if not isinstance(result, dict):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be dict."
)
xyxy, labels, mask, xyxyxyxy = from_florence_2(result, **kwargs)
if len(xyxy) == 0:
empty = cls.empty()
@ -2038,18 +2097,27 @@ class Detections:
return cls(xyxy=xyxy, mask=mask, data=data)
if vlm == VLM.GOOGLE_GEMINI_2_0:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
xyxy, class_id, class_name = from_google_gemini_2_0(result, **kwargs)
data = {CLASS_NAME_DATA_FIELD: class_name}
return cls(xyxy=xyxy, class_id=class_id, data=data)
if vlm == VLM.MOONDREAM:
assert isinstance(result, dict)
if not isinstance(result, dict):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be dict."
)
xyxy = from_moondream(result, **kwargs)
return cls(xyxy=xyxy)
if vlm == VLM.GOOGLE_GEMINI_2_5:
assert isinstance(result, str)
if not isinstance(result, str):
raise ValueError(
f"Invalid VLM result type: {type(result)}. Must be str."
)
gemini_result = from_google_gemini_2_5(result, **kwargs)
data = {CLASS_NAME_DATA_FIELD: gemini_result[2]}
return cls(
@ -2060,7 +2128,7 @@ class Detections:
data=data,
)
return cls.empty()
raise ValueError(f"Unsupported VLM value: {vlm}.")
@classmethod
def from_easyocr(cls, easyocr_results: list[Any]) -> Detections:
@ -2069,6 +2137,10 @@ class Detections:
[EasyOCR](https://github.com/JaidedAI/EasyOCR) result.
Results are placed in the `data` field with the key `"class_name"`.
When EasyOCR returns quadrilateral corners, the original corners are
preserved in ``ORIENTED_BOX_COORDINATES``. Call EasyOCR with
``detail=1`` so bounding boxes are available; ``detail=0`` returns text
strings only and cannot be converted into detections.
Args:
easyocr_results: The output Results instance from EasyOCR.
@ -2090,7 +2162,17 @@ class Detections:
if len(easyocr_results) == 0:
return cls.empty()
bbox = np.array([result[0] for result in easyocr_results])
if isinstance(easyocr_results[0], str):
raise ValueError(
"EasyOCR results produced with detail=0 do not include bounding "
"boxes. Call reader.readtext(..., detail=1) instead."
)
bbox = np.array([result[0] for result in easyocr_results], dtype=np.float32)
if bbox.ndim != 3 or bbox.shape[1:] != (4, 2):
raise ValueError(
"EasyOCR results must contain four corner points per detection."
)
xyxy = np.hstack((np.min(bbox, axis=1), np.max(bbox, axis=1)))
confidence = np.array(
[
@ -2100,12 +2182,14 @@ class Detections:
)
ocr_text = np.array([result[1] for result in easyocr_results])
data: _DetectionDataType = {
CLASS_NAME_DATA_FIELD: ocr_text,
ORIENTED_BOX_COORDINATES: bbox,
}
return cls(
xyxy=xyxy.astype(np.float32),
confidence=confidence.astype(np.float32),
data={
CLASS_NAME_DATA_FIELD: ocr_text,
},
data=data,
)
@classmethod
@ -2899,15 +2983,16 @@ class Detections:
after non-maximum suppression.
Raises:
AssertionError: If `confidence` is None and class_agnostic is False.
ValueError: If `confidence` is None and class_agnostic is False.
If `class_id` is None and class_agnostic is False.
"""
if len(self) == 0:
return self
assert self.confidence is not None, (
"Detections confidence must be given for NMS to be executed."
)
if self.confidence is None:
raise ValueError(
"Detections confidence must be given for NMS to be executed."
)
if class_agnostic:
predictions = cast(
@ -2915,10 +3000,12 @@ class Detections:
np.hstack((self.xyxy, self.confidence.reshape(-1, 1))),
)
else:
assert self.class_id is not None, (
"Detections class_id must be given for NMS to be executed. If you"
" intended to perform class agnostic NMS set class_agnostic=True."
)
if self.class_id is None:
raise ValueError(
"Detections class_id must be given for NMS to be executed. If "
"you intended to perform class agnostic NMS "
"set class_agnostic=True."
)
predictions = cast(
npt.NDArray[np.floating],
np.hstack(
@ -2992,7 +3079,7 @@ class Detections:
Groups of size 1 keep the original OBB unchanged.
Raises:
AssertionError: If `confidence` is None or `class_id` is None and
ValueError: If `confidence` is None or `class_id` is None and
class_agnostic is False.
![non-max-merging](https://media.roboflow.com/supervision-docs/non-max-merging.png){ align=center width="800" }
@ -3000,9 +3087,10 @@ class Detections:
if len(self) == 0:
return self
assert self.confidence is not None, (
"Detections confidence must be given for NMM to be executed."
)
if self.confidence is None:
raise ValueError(
"Detections confidence must be given for NMM to be executed."
)
if class_agnostic:
predictions = cast(
@ -3010,10 +3098,12 @@ class Detections:
np.hstack((self.xyxy, self.confidence.reshape(-1, 1))),
)
else:
assert self.class_id is not None, (
"Detections class_id must be given for NMM to be executed. If you"
" intended to perform class agnostic NMM set class_agnostic=True."
)
if self.class_id is None:
raise ValueError(
"Detections class_id must be given for NMM to be executed. If "
"you intended to perform class agnostic NMM "
"set class_agnostic=True."
)
predictions = cast(
npt.NDArray[np.floating],
np.hstack(

17
src/supervision/detection/line_zone.py
View File

@ -114,14 +114,14 @@ class LineZone:
self.vector = Vector(start=start, end=end)
self.limits = self._calculate_region_of_interest_limits(vector=self.vector)
self.crossing_history_length = max(2, minimum_crossing_threshold + 1)
self.crossing_state_history: dict[tuple[int, int | None], deque[bool]] = (
defaultdict(lambda: deque(maxlen=self.crossing_history_length))
self.crossing_state_history: dict[int, deque[bool]] = defaultdict(
lambda: deque(maxlen=self.crossing_history_length)
)
# Tracks consecutive frames a tracker key has been absent; eviction
# requires crossing_history_length absent frames so that ByteTrack
# coasting gaps (single-frame detection drops) don't reset mid-crossing
# state prematurely.
self._tracker_frames_absent: dict[tuple[int, int | None], int] = {}
self._tracker_frames_absent: dict[int, int] = {}
self._in_count_per_class: Counter[int | None] = Counter()
self._out_count_per_class: Counter[int | None] = Counter()
self.triggering_anchors = triggering_anchors
@ -181,10 +181,7 @@ class LineZone:
if detections.class_id is not None
else [None] * len(detections)
)
current_keys = {
(int(tracker_id), int(class_id) if class_id is not None else None)
for tracker_id, class_id in zip(detections.tracker_id, class_ids)
}
current_keys = {int(tracker_id) for tracker_id in detections.tracker_id}
self._evict_stale_crossing_history(current_keys)
self._update_class_id_to_name(detections)
@ -202,7 +199,7 @@ class LineZone:
continue
tracker_state: bool = has_any_left_trigger[i]
key = (int(tracker_id), int(class_id) if class_id is not None else None)
key = int(tracker_id)
crossing_history = self.crossing_state_history[key]
crossing_history.append(tracker_state)
@ -222,9 +219,7 @@ class LineZone:
return crossed_in, crossed_out
def _evict_stale_crossing_history(
self, current_keys: set[tuple[int, int | None]]
) -> None:
def _evict_stale_crossing_history(self, current_keys: set[int]) -> None:
for key in list(self.crossing_state_history):
if key in current_keys:
self._tracker_frames_absent.pop(key, None)

4
src/supervision/detection/tools/inference_slicer.py
View File

@ -62,7 +62,8 @@ def move_detections(
"""Translate detections by a pixel offset, repositioning boxes and masks.
Args:
detections: Detections object to be moved.
detections: Detections object to be moved. The input is left unchanged;
a fresh copy is returned.
offset: An array of shape `(2,)` containing offset values in the
format `[dx, dy]`.
resolution_wh: The width and height of the desired mask
@ -71,6 +72,7 @@ def move_detections(
Returns:
Repositioned Detections object.
"""
detections = detections.select(slice(None))
detections.xyxy = move_boxes(xyxy=detections.xyxy, offset=offset)
if ORIENTED_BOX_COORDINATES in detections.data:
detections.data[ORIENTED_BOX_COORDINATES] = move_oriented_boxes(

2
src/supervision/detection/tools/polygon_zone.py
View File

@ -91,7 +91,7 @@ class PolygonZone:
all_anchors = np.array(
[
np.ceil(detections.get_anchors_coordinates(anchors)).astype(int)
np.rint(detections.get_anchors_coordinates(anchors)).astype(int)
for anchors in self.triggering_anchors
]
)

14
src/supervision/detection/tools/smoother.py
View File

@ -125,7 +125,8 @@ class DetectionsSmoother:
if all([d is None for d in self.tracks[track_id]]):
del self.tracks[track_id]
return self.get_smoothed_detections()
current_track_ids = {int(track_id) for track_id in detections.tracker_id}
return self.get_smoothed_detections(track_ids=current_track_ids)
def get_track(self, track_id: int) -> Detections | None:
"""Return the smoothed `Detections` for a single track.
@ -160,9 +161,18 @@ class DetectionsSmoother:
return ret
def get_smoothed_detections(self) -> Detections:
def get_smoothed_detections(self, track_ids: set[int] | None = None) -> Detections:
"""Return the smoothed detections for the requested active tracks.
Args:
track_ids: Optional set of track IDs to include in the output. When
provided, tracks absent from the current frame are excluded from the
emitted detections but their history stays cached.
"""
tracked_detections = []
for track_id in self.tracks:
if track_ids is not None and track_id not in track_ids:
continue
track = self.get_track(track_id)
if track is not None:
tracked_detections.append(track)

22
src/supervision/detection/tools/transformers.py
View File

@ -1,3 +1,5 @@
from __future__ import annotations
import io
from typing import Any, cast
@ -209,19 +211,29 @@ def process_transformers_v5_panoptic_segmentation_result(
def png_string_to_segmentation_array(png_string: bytes) -> npt.NDArray[Any]:
"""
Convert a PNG byte string to a label mask array.
Convert a PNG byte string to a panoptic segmentation array.
Args:
png_string: A byte string representing the PNG image.
Returns:
A label mask array with shape (H, W), where H and W
are the height and width of the image. Each unique value in the array
represents a different object or category.
A segmentation ID array with shape (H, W), where each unique value
represents a different object or category. RGB-encoded panoptic
PNGs are decoded as little-endian 24-bit integers; alpha is ignored.
"""
image = Image.open(io.BytesIO(png_string))
mask = np.array(image, dtype=np.uint8)
return cast(npt.NDArray[Any], mask[:, :, 0])
if mask.ndim == 2:
return mask.astype(np.uint32)
if mask.shape[2] < 3:
raise ValueError("Panoptic PNG masks must have at least 3 channels.")
segmentation = (
mask[:, :, 0].astype(np.uint32)
+ (mask[:, :, 1].astype(np.uint32) << 8)
+ (mask[:, :, 2].astype(np.uint32) << 16)
)
return cast(npt.NDArray[Any], segmentation)
def append_class_names_to_data(

64
src/supervision/detection/utils/converters.py
View File

@ -1,10 +1,13 @@
from typing import Any, cast
from __future__ import annotations
from typing import Any, Literal, cast
import cv2
import numpy as np
import numpy.typing as npt
MIN_POLYGON_POINT_COUNT = 3
CoordinateConvention = Literal["inclusive", "exclusive"]
def xyxy_to_polygons(box: npt.NDArray[np.number]) -> npt.NDArray[np.number]:
@ -197,12 +200,19 @@ def xyxy_to_xcycarh(xyxy: npt.NDArray[np.number]) -> npt.NDArray[np.floating]:
return result.astype(float)
def mask_to_xyxy(masks: npt.NDArray[np.bool_]) -> npt.NDArray[np.int_]:
def mask_to_xyxy(
masks: npt.NDArray[np.bool_],
coordinate_convention: CoordinateConvention = "inclusive",
) -> npt.NDArray[np.int_]:
"""
Converts a 3D `np.array` of 2D bool masks into a 2D `np.array` of bounding boxes.
Args:
masks: A 3D `np.array` of shape `(N, H, W)` containing 2D bool masks.
coordinate_convention: How to interpret the returned max corner.
Use `"inclusive"` to preserve the legacy Supervision behavior,
where `x_max` and `y_max` are the last covered pixel. Use
`"exclusive"` for half-open boxes that match area and IoU arithmetic.
Returns:
A 2D `np.array` of shape `(N, 4)` containing the bounding boxes
@ -233,9 +243,19 @@ def mask_to_xyxy(masks: npt.NDArray[np.bool_]) -> npt.NDArray[np.int_]:
cols_any = cast(npt.NDArray[np.bool_], masks.any(axis=1)) # (N, W)
x_min = cols_any.argmax(axis=1)
x_max = width - 1 - cols_any[:, ::-1].argmax(axis=1)
y_min = rows_any.argmax(axis=1)
y_max = height - 1 - rows_any[:, ::-1].argmax(axis=1)
if coordinate_convention == "inclusive":
x_max = width - 1 - cols_any[:, ::-1].argmax(axis=1)
y_max = height - 1 - rows_any[:, ::-1].argmax(axis=1)
elif coordinate_convention == "exclusive":
x_max = width - cols_any[:, ::-1].argmax(axis=1)
y_max = height - rows_any[:, ::-1].argmax(axis=1)
else:
raise ValueError(
"coordinate_convention must be 'inclusive' or 'exclusive', "
f"got {coordinate_convention!r}."
)
xyxy = np.stack((x_min, y_min, x_max, y_max), axis=1).astype(int)
# Empty masks have no bounds; keep the original all-zeros box for them.
@ -244,7 +264,9 @@ def mask_to_xyxy(masks: npt.NDArray[np.bool_]) -> npt.NDArray[np.int_]:
def xyxy_to_mask(
boxes: npt.NDArray[np.number], resolution_wh: tuple[int, int]
boxes: npt.NDArray[np.number],
resolution_wh: tuple[int, int],
coordinate_convention: CoordinateConvention = "inclusive",
) -> npt.NDArray[np.bool_]:
"""
Converts a 2D `np.ndarray` of bounding boxes into a 3D `np.ndarray` of bool masks.
@ -254,6 +276,10 @@ def xyxy_to_mask(
`(x_min, y_min, x_max, y_max)`.
resolution_wh: A tuple `(width, height)` specifying the resolution of
the output masks.
coordinate_convention: How to interpret `x_max` and `y_max`.
Use `"inclusive"` for closed boxes with the legacy Supervision
convention. Use `"exclusive"` for half-open boxes that align with
box area and IoU calculations.
Returns:
A 3D `np.ndarray` of shape `(N, height, width)` containing 2D bool masks
@ -293,11 +319,21 @@ def xyxy_to_mask(
for i, (x_min, y_min, x_max, y_max) in enumerate(boxes):
x_min = max(0, int(x_min))
y_min = max(0, int(y_min))
x_max = min(width - 1, int(x_max))
y_max = min(height - 1, int(y_max))
if x_max >= x_min and y_max >= y_min:
masks[i, y_min : y_max + 1, x_min : x_max + 1] = True
if coordinate_convention == "inclusive":
x_max = min(width - 1, int(x_max))
y_max = min(height - 1, int(y_max))
if x_max >= x_min and y_max >= y_min:
masks[i, y_min : y_max + 1, x_min : x_max + 1] = True
elif coordinate_convention == "exclusive":
x_max = min(width, int(x_max))
y_max = min(height, int(y_max))
if x_max > x_min and y_max > y_min:
masks[i, y_min:y_max, x_min:x_max] = True
else:
raise ValueError(
"coordinate_convention must be 'inclusive' or 'exclusive', "
f"got {coordinate_convention!r}."
)
return masks
@ -676,7 +712,7 @@ def mask_to_rle(
When ``compressed`` is ``True``, a COCO compressed RLE string.
Raises:
AssertionError: If input mask is not 2D or is empty.
ValueError: If input mask is not 2D or is empty.
Examples:
```pycon
@ -715,8 +751,10 @@ def mask_to_rle(
![mask_to_rle](https://media.roboflow.com/supervision-docs/
mask-to-rle.png){ align=center width="800" }
"""
assert mask.ndim == 2, "Input mask must be 2D"
assert mask.size != 0, "Input mask cannot be empty"
if mask.ndim != 2:
raise ValueError("Input mask must be 2D")
if mask.size == 0:
raise ValueError("Input mask cannot be empty")
counts: list[int] = cast(list[int], _mask_to_rle_counts(mask).tolist())
if compressed:

4
src/supervision/detection/utils/internal.py
View File

@ -629,13 +629,13 @@ def merge_metadata(metadata_list: list[_MetadataType]) -> _MetadataType:
if not np.array_equal(merged_metadata[key], value):
raise ValueError(
f"Conflicting metadata for key: '{key}': "
"{type(value)}, {type(other_value)}."
f"{type(value)}, {type(other_value)}."
)
elif isinstance(value, np.ndarray) or isinstance(other_value, np.ndarray):
# Since [] == np.array([]).
raise ValueError(
f"Conflicting metadata for key: '{key}': "
"{type(value)}, {type(other_value)}."
f"{type(value)}, {type(other_value)}."
)
else:
if merged_metadata[key] != value:

59
src/supervision/detection/utils/iou_and_nms.py
View File

@ -88,6 +88,15 @@ class OverlapMetric(Enum):
)
def _validate_iou_threshold(iou_threshold: float) -> None:
"""Raise `ValueError` when an IoU threshold falls outside `[0, 1]`."""
if not 0 <= iou_threshold <= 1:
raise ValueError(
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
def box_iou(
box_true: list[float] | npt.NDArray[np.floating],
box_detection: list[float] | npt.NDArray[np.floating],
@ -878,13 +887,10 @@ def mask_non_max_suppression(
non-maximum suppression.
Raises:
AssertionError: If `iou_threshold` is not within the closed
range from `0` to `1`.
ValueError: If `iou_threshold` is not within the closed range
from `0` to `1`.
"""
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
_validate_iou_threshold(iou_threshold)
rows, columns = predictions.shape
if columns == 5:
@ -971,13 +977,10 @@ def box_non_max_suppression(
non-maximum suppression.
Raises:
AssertionError: If `iou_threshold` is not within the
closed range from `0` to `1`.
ValueError: If `iou_threshold` is not within the closed range
from `0` to `1`.
"""
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
_validate_iou_threshold(iou_threshold)
sort_index, predictions, categories = _prepare_predictions_for_nms(predictions)
ious = box_iou_batch(predictions[:, :4], predictions[:, :4], overlap_metric)
keep = _nms_loop_from_iou_matrix(ious, categories, iou_threshold)
@ -1076,8 +1079,8 @@ def mask_non_max_merge(
kept together as a single detection by non-maximum merging.
Raises:
AssertionError: If `iou_threshold` is not within the closed
range from `0` to `1`.
ValueError: If `iou_threshold` is not within the closed range
from `0` to `1`.
TypeError: If more than five positional arguments are passed.
Examples:
@ -1096,10 +1099,7 @@ def mask_non_max_merge(
```
"""
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
_validate_iou_threshold(iou_threshold)
if len(args) > 2:
raise TypeError(
"mask_non_max_merge accepts at most five positional arguments. "
@ -1309,7 +1309,12 @@ def box_non_max_merge(
Returns:
list[list[int]]: Groups of prediction indices be merged.
Each group may have 1 or more elements.
Raises:
ValueError: If `iou_threshold` is not within the closed range
from `0` to `1`.
"""
_validate_iou_threshold(iou_threshold)
def group_within(global_indices: npt.NDArray[np.int_]) -> list[list[int]]:
return _group_overlapping_boxes(
@ -1352,8 +1357,8 @@ def oriented_box_non_max_suppression(
to keep after non-maximum suppression.
Raises:
AssertionError: If ``iou_threshold`` is not within the closed
range from 0 to 1.
ValueError: If ``iou_threshold`` is not within the closed range
from 0 to 1.
ValueError: If ``predictions`` and ``oriented_boxes`` have
mismatched lengths or invalid shapes.
@ -1376,10 +1381,7 @@ def oriented_box_non_max_suppression(
>>> keep
array([ True, False])
"""
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
_validate_iou_threshold(iou_threshold)
for name, arr in (("predictions", predictions), ("oriented_boxes", oriented_boxes)):
if name == "predictions":
if arr.ndim != 2 or arr.shape[1] not in (5, 6):
@ -1478,8 +1480,8 @@ def oriented_box_non_max_merge(
or more elements.
Raises:
AssertionError: If ``iou_threshold`` is not within the closed
range from 0 to 1.
ValueError: If ``iou_threshold`` is not within the closed range
from 0 to 1.
ValueError: If ``predictions`` and ``oriented_boxes`` have
mismatched lengths or invalid shapes.
@ -1528,10 +1530,7 @@ def oriented_box_non_max_merge(
f"`predictions` and `oriented_boxes` must have the same length, "
f"got {len(predictions)} and {len(oriented_boxes)}."
)
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
_validate_iou_threshold(iou_threshold)
def group_within(global_indices: npt.NDArray[np.int_]) -> list[list[int]]:
return _group_overlapping_oriented_boxes(

18
src/supervision/detection/utils/masks.py
View File

@ -452,13 +452,14 @@ def filter_segments_by_distance(
return keep_labels[labels]
def _mask_to_roi(mask: npt.NDArray[np.bool_]) -> tuple[int, int, int, int] | None:
def mask_to_roi(mask: npt.NDArray[np.bool_]) -> tuple[int, int, int, int] | None:
"""Return exclusive ``(x1, y1, x2, y2)`` bounds for true mask pixels.
Unlike :func:`~supervision.detection.utils.converters.mask_to_xyxy`,
this function uses **exclusive** upper bounds (``+1``) and returns
``None`` for empty masks (instead of zeros). These semantics are
required for NumPy slice-based ROI extraction.
Use this helper when you need NumPy slice semantics. Unlike
:func:`~supervision.detection.utils.converters.mask_to_xyxy`, this
function uses exclusive upper bounds (``+1``) and returns ``None`` for
empty masks instead of zeros. The inclusive ``mask_to_xyxy`` convention
stays in place for compatibility with CompactMask and box-based adapters.
Args:
mask: 2D boolean array of shape ``(H, W)``.
@ -474,6 +475,9 @@ def _mask_to_roi(mask: npt.NDArray[np.bool_]) -> tuple[int, int, int, int] | Non
return int(cols[0]), int(rows[0]), int(cols[-1]) + 1, int(rows[-1]) + 1
_mask_to_roi = mask_to_roi
def _compact_masks_to_roi(
masks: CompactMask,
image_shape: tuple[int, int],
@ -554,9 +558,9 @@ def _masks_to_roi(
and not union[y1:y2, x2:].any()
):
return box_roi
return _mask_to_roi(union)
return mask_to_roi(union)
if mask_array.ndim == 2:
union = mask_array
else:
union = np.any(mask_array, axis=0)
return _mask_to_roi(union)
return mask_to_roi(union)

19
src/supervision/detection/vlm.py
View File

@ -524,8 +524,13 @@ def from_florence_2(
optional array of shape `(n, h, w)` with segmentation masks, and
`obb_boxes` is an optional array of shape `(n, 4, 2)` with oriented
bounding boxes.
Raises:
ValueError: If the top-level Florence 2 payload has multiple tasks or
if a task payload is malformed.
"""
assert len(result) == 1, f"Expected result with a single element. Got: {result}"
if len(result) != 1:
raise ValueError(f"Expected result with a single element. Got: {result}")
task = next(iter(result.keys()))
if task not in SUPPORTED_TASKS_FLORENCE_2:
raise ValueError(
@ -574,18 +579,18 @@ def from_florence_2(
return xyxy, labels, None, None
if task in ["<REGION_TO_CATEGORY>", "<REGION_TO_DESCRIPTION>"]:
assert isinstance(result, str), (
f"Expected string as <REGION_TO_CATEGORY> result, got {type(result)}"
)
if not isinstance(result, str):
raise ValueError(f"Expected string as {task} result, got {type(result)}")
if result == "No object detected.":
return np.empty((0, 4), dtype=np.float32), np.array([]), None, None
pattern = re.compile(r"<loc_(\d+)><loc_(\d+)><loc_(\d+)><loc_(\d+)>")
match = pattern.search(result)
assert match is not None, (
f"Expected string to end in location tags, but got {result}"
)
if match is None:
raise ValueError(
f"Expected string to end in location tags, but got {result}"
)
w, h = _validate_resolution(resolution_wh)
xyxy = np.array([match.groups()], dtype=np.float32)

10
src/supervision/metrics/detection.py
View File

@ -4,10 +4,8 @@ import warnings
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from typing import TYPE_CHECKING, Any, cast
import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import numpy.typing as npt
from deprecate import ( # type: ignore[import-untyped,unused-ignore]
@ -27,6 +25,9 @@ from supervision.detection.utils.iou_and_nms import (
from supervision.metrics.core import MetricTarget
from supervision.metrics.utils.matching import _greedy_match
if TYPE_CHECKING:
from matplotlib.figure import Figure
def _assert_supported_target(metric_target: MetricTarget) -> None:
if metric_target == MetricTarget.MASKS:
@ -1151,7 +1152,7 @@ class ConfusionMatrix:
classes: list[str] | None = None,
normalize: bool = False,
fig_size: tuple[int, int] = (12, 10),
) -> matplotlib.figure.Figure:
) -> Figure:
"""
Create confusion matrix plot and save it at selected location.
@ -1167,6 +1168,7 @@ class ConfusionMatrix:
Returns:
Confusion matrix plot.
"""
from matplotlib import pyplot as plt
# Cast to float so that the NaN masking below never hits an integer
# matrix (assigning NaN into an int array raises ValueError).

2
src/supervision/metrics/f1_score.py
View File

@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
import numpy as np
import numpy.typing as npt
from matplotlib import pyplot as plt
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.compact_mask import CompactMask
@ -758,6 +757,7 @@ class F1ScoreResult:
https://media.roboflow.com/supervision-docs/metrics/f1_plot_example.png
){ align=center width="800" }
"""
from matplotlib import pyplot as plt
labels = ["F1@50", "F1@75"]
values = [self.f1_50, self.f1_75]

2
src/supervision/metrics/mean_average_precision.py
View File

@ -11,7 +11,6 @@ from typing import TYPE_CHECKING, Any, TypeAlias, TypedDict
import numpy as np
import numpy.typing as npt
from matplotlib import pyplot as plt
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.core import Detections
@ -229,6 +228,7 @@ class MeanAveragePrecisionResult:
https://media.roboflow.com/supervision-docs/metrics/mAP_plot_example.png
){ align=center width="800" }
"""
from matplotlib import pyplot as plt
labels = ["mAP@50:95", "mAP@50", "mAP@75"]
values = [self.map50_95, self.map50, self.map75]

3
src/supervision/metrics/mean_average_recall.py
View File

@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
import numpy as np
import numpy.typing as npt
from matplotlib import pyplot as plt
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.compact_mask import CompactMask
@ -202,6 +201,8 @@ class MeanAverageRecallResult:
https://media.roboflow.com/supervision-docs/metrics/mAR_plot_example.png\
){ align=center width="800" }
"""
from matplotlib import pyplot as plt
labels = ["mAR @ 1", "mAR @ 10", "mAR @ 100"]
values = [self.mAR_at_1, self.mAR_at_10, self.mAR_at_100]
colors = [LEGACY_COLOR_PALETTE[0]] * 3

3
src/supervision/metrics/precision.py
View File

@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
import numpy as np
import numpy.typing as npt
from matplotlib import pyplot as plt
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.core import Detections
@ -758,6 +757,8 @@ class PrecisionResult:
){ align=center width="800" }
"""
from matplotlib import pyplot as plt
labels = ["Precision@50", "Precision@75"]
values = [self.precision_at_50, self.precision_at_75]
colors = [LEGACY_COLOR_PALETTE[0]] * 2

2
src/supervision/metrics/recall.py
View File

@ -6,7 +6,6 @@ from typing import TYPE_CHECKING, Any, cast
import numpy as np
import numpy.typing as npt
from matplotlib import pyplot as plt
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.compact_mask import CompactMask
@ -717,6 +716,7 @@ class RecallResult:
https://media.roboflow.com/supervision-docs/metrics/recall_plot_example.png
){ align=center width="800" }
"""
from matplotlib import pyplot as plt
labels = ["Recall@50", "Recall@75"]
values = [self.recall_at_50, self.recall_at_75]

5
src/supervision/tracker/__init__.py
View File

@ -0,0 +1,5 @@
"""Deprecated tracker compatibility exports."""
from supervision.tracker.byte_tracker.core import ByteTrack
__all__ = ["ByteTrack"]

10
src/supervision/utils/conversion.py
View File

@ -160,15 +160,21 @@ def images_to_cv2(
def pillow_to_cv2(image: Image.Image) -> npt.NDArray[np.uint8]:
"""
Converts Pillow image into OpenCV image, handling RGB -> BGR
conversion.
conversion. Palette images are first expanded to RGB so palette indices are
resolved to their actual colors.
Args:
image: Pillow image (in RGB format).
image: Pillow image in RGB, grayscale, or palette mode.
Returns:
Input image converted to OpenCV format.
"""
if image.mode == "P":
image = image.convert("RGB")
scene = np.array(image)
if scene.ndim == 2:
return cast(npt.NDArray[np.uint8], scene.astype(np.uint8, copy=False))
scene = cv2.cvtColor(scene, cv2.COLOR_RGB2BGR)
# cvtColor already returns uint8 here, so astype is a no-op other than the
# full-image copy it forces; copy=False keeps the dtype guard without it.

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

@ -51,6 +51,11 @@ def crop_image(
Cropped image matching input
type.
Note:
Coordinates are rounded to integers and clipped to the image bounds
before slicing. This keeps NumPy and Pillow inputs aligned and avoids
negative-index wrap-around on NumPy arrays.
Examples:
```pycon
>>> import numpy as np
@ -82,9 +87,19 @@ def crop_image(
x_min, y_min, x_max, y_max = xyxy_arr.flatten()
if isinstance(image, np.ndarray):
height, width = image.shape[:2]
x_min = int(np.clip(x_min, 0, width))
y_min = int(np.clip(y_min, 0, height))
x_max = int(np.clip(x_max, 0, width))
y_max = int(np.clip(y_max, 0, height))
return image[y_min:y_max, x_min:x_max]
if isinstance(image, Image.Image):
width, height = image.size
x_min = int(np.clip(x_min, 0, width))
y_min = int(np.clip(y_min, 0, height))
x_max = int(np.clip(x_max, 0, width))
y_max = int(np.clip(y_max, 0, height))
return image.crop((float(x_min), float(y_min), float(x_max), float(y_max)))
raise TypeError(

13
src/supervision/utils/video.py
View File

@ -444,9 +444,8 @@ def process_video(
try:
frame_write_queue.put(None, timeout=1)
except Full:
# Queue is full; this is a best-effort attempt to enqueue the sentinel.
# If we cannot enqueue it, the writer thread will still complete based
# on previously queued frames or other shutdown conditions.
# Best effort: if the writer is stuck and the queue never drains,
# do not block shutdown forever trying to enqueue the sentinel.
pass
if not read_finished:
while True:
@ -512,12 +511,14 @@ class FPSMonitor:
Computes and returns the average FPS based on the stored time stamps.
Returns:
The average FPS. Returns 0.0 if no time stamps are stored.
The average FPS across the recorded intervals. Returns 0.0 if fewer
than two time stamps are stored.
"""
if not self.all_timestamps:
if len(self.all_timestamps) < 2:
return 0.0
taken_time = self.all_timestamps[-1] - self.all_timestamps[0]
return (len(self.all_timestamps)) / taken_time if taken_time != 0 else 0.0
frame_intervals = len(self.all_timestamps) - 1
return frame_intervals / taken_time if taken_time != 0 else 0.0
def tick(self) -> None:
"""

105
tests/annotators/test_core.py
View File

@ -4,11 +4,14 @@ Tests for supervision/annotators/core.py
import warnings
from collections.abc import Iterator
from typing import Any, cast
import cv2
import numpy as np
import pytest
from PIL import Image
import supervision.annotators.core as annotators_core
from supervision.annotators.base import BaseAnnotator
from supervision.annotators.core import (
BackgroundOverlayAnnotator,
@ -972,6 +975,20 @@ class TestHeatMapAnnotator:
warnings.simplefilter("error", RuntimeWarning)
annotator.annotate(scene=test_image.copy(), detections=Detections.empty())
def test_annotate_resets_when_resolution_changes(self) -> None:
"""Changing frame resolution must reset heat state instead of crashing."""
annotator = HeatMapAnnotator()
detections = _create_detections(xyxy=[[20, 20, 60, 60]])
first_scene = np.zeros((100, 100, 3), dtype=np.uint8)
second_scene = np.zeros((120, 80, 3), dtype=np.uint8)
annotator.annotate(scene=first_scene.copy(), detections=detections)
result = annotator.annotate(scene=second_scene.copy(), detections=detections)
assert result.shape == second_scene.shape
assert annotator.heat_mask is not None
assert annotator.heat_mask.shape == second_scene.shape[:2]
def test_annotate_hottest_region_survives_uint8_wrap(
self, test_image: np.ndarray
) -> None:
@ -1138,6 +1155,35 @@ class TestLabelAnnotator:
)
assert_image_mostly_same(test_image, result, similarity_threshold=0.93)
def test_smart_position_spreads_boxes_once(
self, monkeypatch: pytest.MonkeyPatch, test_image: np.ndarray
) -> None:
"""smart_position should spread labels once per annotate call."""
calls = 0
original_spread_out_boxes = annotators_core.spread_out_boxes
def counting_spread_out_boxes(
boxes: np.ndarray, *args: object, **kwargs: object
) -> np.ndarray:
nonlocal calls
calls += 1
return original_spread_out_boxes(boxes, *args, **kwargs)
monkeypatch.setattr(
annotators_core, "spread_out_boxes", counting_spread_out_boxes
)
detections = _create_detections(
xyxy=[[10, 10, 90, 90], [15, 15, 85, 85]], class_id=[0, 1]
)
annotator = LabelAnnotator(color_lookup=ColorLookup.INDEX, smart_position=True)
annotator.annotate(
scene=test_image.copy(), detections=detections, labels=["one", "two"]
)
assert calls == 1
class TestRichLabelAnnotator:
"""Tests for RichLabelAnnotator class"""
@ -1158,6 +1204,39 @@ class TestRichLabelAnnotator:
)
assert_image_mostly_same(test_image, result, similarity_threshold=0.95)
def test_smart_position_spreads_boxes_once(
self, monkeypatch: pytest.MonkeyPatch, test_image: np.ndarray
) -> None:
"""smart_position should spread rich labels once per annotate call."""
calls = 0
original_spread_out_boxes = annotators_core.spread_out_boxes
def counting_spread_out_boxes(
boxes: np.ndarray, *args: object, **kwargs: object
) -> np.ndarray:
nonlocal calls
calls += 1
return original_spread_out_boxes(boxes, *args, **kwargs)
monkeypatch.setattr(
annotators_core, "spread_out_boxes", counting_spread_out_boxes
)
detections = _create_detections(
xyxy=[[10, 10, 90, 90], [15, 15, 85, 85]], class_id=[0, 1]
)
annotator = RichLabelAnnotator(
color_lookup=ColorLookup.INDEX, smart_position=True
)
annotator.annotate(
scene=Image.fromarray(test_image.copy()),
detections=detections,
labels=["one", "two"],
)
assert calls == 1
class TestBlurAnnotator:
"""Tests for BlurAnnotator class"""
@ -1319,6 +1398,32 @@ class TestPercentageBarAnnotator:
assert_image_mostly_same(test_image, result, similarity_threshold=0.93)
class TestPositionHelpers:
"""Tests for helper methods that map `Position` to coordinates."""
@pytest.mark.parametrize(
("helper", "args"),
[
pytest.param(
PercentageBarAnnotator.calculate_border_coordinates,
((10, 10), (4, 4), cast(Position, "invalid")),
id="percentage-bar",
),
pytest.param(
CropAnnotator.calculate_crop_coordinates,
((10, 10), (4, 4), cast(Position, "invalid")),
id="crop",
),
],
)
def test_unknown_position_raises(
self, helper: Any, args: tuple[Any, Any, Any]
) -> None:
"""Unsupported positions must raise instead of returning None."""
with pytest.raises(ValueError, match="Unsupported position"):
helper(*args)
class TestCropAnnotator:
"""Tests for CropAnnotator class"""

33
tests/annotators/test_utils.py
View File

@ -1,19 +1,23 @@
from contextlib import ExitStack as DoesNotRaise
from typing import cast
import numpy as np
import pytest
from supervision.annotators.utils import (
ColorLookup,
Trace,
hex_to_rgba,
is_valid_hex,
resolve_color,
resolve_color_idx,
resolve_text_background_xyxy,
rgba_to_hex,
wrap_text,
)
from supervision.detection.core import Detections
from supervision.draw.color import Color, ColorPalette
from supervision.geometry.core import Position
from tests.helpers import _create_detections
@ -212,6 +216,35 @@ def test_wrap_text(
assert result == expected_result
def test_resolve_text_background_xyxy_rejects_unknown_position() -> None:
"""Unsupported positions must raise instead of returning an implicit None."""
with pytest.raises(ValueError, match="Unsupported position"):
resolve_text_background_xyxy(
center_coordinates=(10, 10),
text_wh=(20, 10),
position=cast(Position, "invalid"),
)
def test_trace_put_requires_tracker_id_before_mutation() -> None:
"""Trace.put must not mutate internal history before validating tracker ids."""
trace = Trace()
detections = _create_detections(xyxy=[[0, 0, 1, 1]], class_id=[0])
before_frame_id = trace.current_frame_id
before_history = trace.frame_id.copy()
before_xy = trace.xy.copy()
before_tracker_id = trace.tracker_id.copy()
with pytest.raises(ValueError, match="tracker_id"):
trace.put(detections)
assert trace.current_frame_id == before_frame_id
assert np.array_equal(trace.frame_id, before_history)
assert np.array_equal(trace.xy, before_xy)
assert np.array_equal(trace.tracker_id, before_tracker_id)
@pytest.mark.parametrize(
("hex_color", "expected_rgba"),
[

107
tests/assets/test_downloader.py
View File

@ -2,7 +2,11 @@ from unittest.mock import MagicMock, mock_open, patch
import pytest
from supervision.assets.downloader import download_assets, is_md5_hash_matching
from supervision.assets.downloader import (
_download_asset,
download_assets,
is_md5_hash_matching,
)
from supervision.assets.list import ImageAssets, VideoAssets
@ -36,16 +40,20 @@ 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) -> None:
def test_already_exists_and_valid(
self, mock_exists, mock_md5, mock_logger, mock_replace
) -> 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(
@ -69,6 +77,7 @@ class TestDownloadAssets:
mock_md5,
mock_remove,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets when file exists but is corrupted (re-downloads)."""
filename = "vehicles.mp4"
@ -82,7 +91,7 @@ class TestDownloadAssets:
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
@ -92,6 +101,7 @@ class TestDownloadAssets:
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.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@ -110,6 +120,7 @@ class TestDownloadAssets:
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets verifies a freshly downloaded file."""
filename = "vehicles.mp4"
@ -123,7 +134,7 @@ class TestDownloadAssets:
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
assert result == filename
@ -133,6 +144,7 @@ class TestDownloadAssets:
mock_copyfileobj.assert_called_once()
mock_md5.assert_called_once_with(filename, "8155ff4e4de08cfa25f39de96483f918")
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("os.remove")
@patch(
@ -156,6 +168,7 @@ class TestDownloadAssets:
mock_md5,
mock_remove,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets retries once when a fresh payload fails MD5."""
filename = "vehicles.mp4"
@ -169,7 +182,7 @@ class TestDownloadAssets:
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
result = download_assets(filename)
@ -179,6 +192,7 @@ class TestDownloadAssets:
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(
@ -202,6 +216,7 @@ class TestDownloadAssets:
mock_md5,
mock_remove,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets fails after the verified retry is also corrupted."""
filename = "vehicles.mp4"
@ -215,7 +230,7 @@ class TestDownloadAssets:
mock_tqdm.wrapattr.return_value.__enter__ = MagicMock(
return_value=mock_response.raw
)
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock()
mock_tqdm.wrapattr.return_value.__exit__ = MagicMock(return_value=False)
with pytest.raises(ValueError, match="failed MD5 verification"):
download_assets(filename)
@ -225,6 +240,82 @@ class TestDownloadAssets:
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,
) -> None:
"""Test download_assets writes into an explicit output directory."""
filename = "vehicles.mp4"
target_directory = tmp_path / "nested" / "assets"
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)
assert result == str(target_directory / filename)
assert (target_directory / filename).exists()
assert (target_directory / filename).read_bytes() == b"asset-bytes"
mock_md5.assert_called_once_with(
str(target_directory / filename), "8155ff4e4de08cfa25f39de96483f918"
)
@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,
) -> 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")
mock_replace.side_effect = OSError("boom")
with 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()
@patch("pathlib.Path.exists", return_value=False)
def test_invalid_asset(self, mock_exists) -> None:
"""Test download_assets with invalid asset name."""
@ -247,6 +338,7 @@ 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.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@ -265,6 +357,7 @@ class TestDownloadAssets:
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets with VideoAssets enum."""
asset = VideoAssets.VEHICLES
@ -285,6 +378,7 @@ class TestDownloadAssets:
asset.filename, "8155ff4e4de08cfa25f39de96483f918"
)
@patch("os.replace")
@patch("supervision.assets.downloader.logger")
@patch("supervision.assets.downloader.is_md5_hash_matching", return_value=True)
@patch("pathlib.Path.open", new_callable=mock_open)
@ -303,6 +397,7 @@ class TestDownloadAssets:
mock_open_file,
mock_md5,
mock_logger,
mock_replace,
) -> None:
"""Test download_assets with ImageAssets enum."""
asset = ImageAssets.SOCCER

16
tests/conftest.py
View File

@ -1,11 +1,21 @@
import matplotlib
import sys
import warnings
from pathlib import Path
import numpy as np
import pytest
with warnings.catch_warnings():
warnings.simplefilter("ignore")
import matplotlib
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "src"))
matplotlib.use("Agg")
import supervision as sv
from tests.helpers import _create_key_points
import supervision as sv # noqa: E402
from tests.helpers import _create_key_points # noqa: E402
@pytest.fixture

83
tests/dataset/formats/test_coco.py
View File

@ -1577,6 +1577,37 @@ def test_load_coco_annotations_rejects_file_name_resolving_to_directory(
)
def test_load_coco_annotations_rejects_unresolvable_file_name(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Reject file_name values whose resolved path cannot be computed."""
images_directory = tmp_path / "images"
images_directory.mkdir()
annotations_path = tmp_path / "annotations.json"
coco_data = {
"categories": [{"id": 1, "name": "object", "supercategory": "none"}],
"images": [{"id": 1, "file_name": "bad.jpg", "width": 5, "height": 5}],
"annotations": [],
}
annotations_path.write_text(json.dumps(coco_data), encoding="utf-8")
original_resolve = Path.resolve
def fake_resolve(self: Path, *args: object, **kwargs: object) -> Path:
if self == images_directory / "bad.jpg":
raise OSError("unresolvable path")
return original_resolve(self, *args, **kwargs)
monkeypatch.setattr(Path, "resolve", fake_resolve)
with pytest.raises(ValueError, match="invalid path"):
load_coco_annotations(
images_directory_path=str(images_directory),
annotations_path=str(annotations_path),
)
def test_load_coco_annotations_accepts_valid_nested_file_name(tmp_path) -> None:
"""Accept a legitimate nested file_name inside images/ without raising."""
images_directory = tmp_path / "images"
@ -1599,6 +1630,32 @@ def test_load_coco_annotations_accepts_valid_nested_file_name(tmp_path) -> None:
assert expected_path in annotations
def test_load_coco_annotations_rejects_duplicate_resolved_file_names(
tmp_path: Path,
) -> None:
"""Aliases for the same file resolve to one canonical COCO entry."""
images_directory = tmp_path / "images"
images_directory.mkdir()
(images_directory / "nested").mkdir()
annotations_path = tmp_path / "annotations.json"
coco_data = {
"categories": [{"id": 1, "name": "object", "supercategory": "none"}],
"images": [
{"id": 1, "file_name": "image.jpg", "width": 5, "height": 5},
{"id": 2, "file_name": "nested/../image.jpg", "width": 5, "height": 5},
],
"annotations": [],
}
annotations_path.write_text(json.dumps(coco_data), encoding="utf-8")
with pytest.raises(ValueError, match="duplicate entries for image"):
load_coco_annotations(
images_directory_path=str(images_directory),
annotations_path=str(annotations_path),
)
def test_load_coco_annotations_force_masks_handles_missing_segmentation(
tmp_path,
) -> None:
@ -1872,6 +1929,32 @@ def _read_ids(annotation_path) -> tuple[list[int], list[int]]:
return image_ids, annotation_ids
class TestSaveCocoAnnotationsCollisionGuard:
"""COCO export must reject same-basename images before writing."""
def test_raises_on_duplicate_image_basenames(self, tmp_path: Path) -> None:
"""Duplicate image basenames are rejected instead of being collapsed."""
image_paths = []
annotations: dict[str, Detections] = {}
for parent in ("dir_a", "dir_b"):
image_path = tmp_path / parent / "img.jpg"
image_path.parent.mkdir(parents=True, exist_ok=True)
assert cv2.imwrite(str(image_path), np.zeros((10, 10, 3), dtype=np.uint8))
image_path_str = str(image_path)
image_paths.append(image_path_str)
annotations[image_path_str] = Detections.empty()
dataset = DetectionDataset(
classes=["object"], images=image_paths, annotations=annotations
)
with pytest.raises(ValueError, match="COCO image file"):
save_coco_annotations(
dataset=dataset,
annotation_path=str(tmp_path / "annotations.json"),
)
def test_save_coco_annotations_defaults_start_at_one(tmp_path):
dataset = _tiny_detection_dataset(tmp_path, "img", num_images=2, dets_per_image=3)
annotation_path = tmp_path / "annotations.json"

69
tests/dataset/formats/test_createml.py
View File

@ -191,6 +191,58 @@ class TestLoadCreatemlAnnotations:
np.testing.assert_array_equal(detections.class_id, np.array([0], dtype=int))
assert len(annotations[str(tmp_path / "b.jpg")]) == 0
def test_rejects_duplicate_resolved_image_paths(self, tmp_path: Path) -> None:
"""Aliases for the same file resolve to one canonical CreateML entry."""
annotations_path = tmp_path / "annotations.json"
(tmp_path / "nested").mkdir()
payload = [
{"image": "a.jpg", "annotations": []},
{"image": "nested/../a.jpg", "annotations": []},
]
annotations_path.write_text(json.dumps(payload))
with pytest.raises(ValueError, match="duplicate entries for image"):
load_createml_annotations(
images_directory_path=str(tmp_path),
annotations_path=str(annotations_path),
)
def test_rejects_unresolvable_image_path(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Unresolvable image paths are rejected before they enter the dataset."""
annotations_path = tmp_path / "annotations.json"
payload = [{"image": "bad.jpg", "annotations": []}]
annotations_path.write_text(json.dumps(payload))
original_resolve = Path.resolve
def fake_resolve(self: Path, *args: object, **kwargs: object) -> Path:
if self == tmp_path / "bad.jpg":
raise OSError("unresolvable path")
return original_resolve(self, *args, **kwargs)
monkeypatch.setattr(Path, "resolve", fake_resolve)
with pytest.raises(ValueError, match="invalid path"):
load_createml_annotations(
images_directory_path=str(tmp_path),
annotations_path=str(annotations_path),
)
def test_rejects_image_path_resolving_to_directory(self, tmp_path: Path) -> None:
"""CreateML loader rejects entries that resolve to a directory."""
annotations_path = tmp_path / "annotations.json"
(tmp_path / "nested").mkdir()
payload = [{"image": "nested", "annotations": []}]
annotations_path.write_text(json.dumps(payload))
with pytest.raises(ValueError, match="directory"):
load_createml_annotations(
images_directory_path=str(tmp_path),
annotations_path=str(annotations_path),
)
def test_assigns_global_sorted_class_ids(self, tmp_path: Path) -> None:
"""Class ids are globally sorted regardless of per-image label order."""
annotations_path = tmp_path / "annotations.json"
@ -308,6 +360,23 @@ class TestLoadCreatemlAnnotations:
class TestSaveCreatemlAnnotations:
"""CreateML export must reject same-basename images before writing."""
def test_raises_on_duplicate_image_basenames(self, tmp_path: Path) -> None:
"""Duplicate image basenames are rejected instead of becoming duplicates."""
image_paths = ["dir_a/img.jpg", "dir_b/img.jpg"]
dataset = DetectionDataset(
classes=["object"],
images=image_paths,
annotations={path: Detections.empty() for path in image_paths},
)
with pytest.raises(ValueError, match="CreateML image file"):
save_createml_annotations(
dataset=dataset,
annotations_path=str(tmp_path / "annotations.json"),
)
def test_empty_dataset_writes_empty_list(self, tmp_path: Path) -> None:
"""Empty dataset serialises to an empty JSON array."""
annotations_path = tmp_path / "nested" / "annotations.json"

70
tests/dataset/test_core.py
View File

@ -1,3 +1,4 @@
import warnings
from contextlib import ExitStack as DoesNotRaise
from pathlib import Path
@ -381,13 +382,17 @@ class TestDetectionDatasetInMemoryImages:
assert len(dataset) == 2
def test_merge_preserves_in_memory_pixel_access(self) -> None:
"""Merging two in-memory datasets keeps pixel access via public __getitem__."""
"""Merging two in-memory datasets keeps pixel access without re-warning."""
image_1 = _create_image(fill_value=10)
image_2 = _create_image(fill_value=20)
ds_1 = self._build_dataset({"img1.jpg": image_1})
ds_2 = self._build_dataset({"img2.jpg": image_2})
with pytest.warns(SupervisionWarnings, match="deprecated"):
ds_1 = self._build_dataset({"img1.jpg": image_1})
with pytest.warns(SupervisionWarnings, match="deprecated"):
ds_2 = self._build_dataset({"img2.jpg": image_2})
merged = DetectionDataset.merge([ds_1, ds_2])
with warnings.catch_warnings():
warnings.simplefilter("error", SupervisionWarnings)
merged = DetectionDataset.merge([ds_1, ds_2])
assert len(merged) == 2
_, loaded_1, _ = merged[0]
@ -395,6 +400,24 @@ class TestDetectionDatasetInMemoryImages:
np.testing.assert_array_equal(loaded_1, image_1)
np.testing.assert_array_equal(loaded_2, image_2)
def test_split_preserves_in_memory_pixel_access_without_warning(self) -> None:
"""Splitting an in-memory dataset keeps pixel access without re-warning."""
image_1 = _create_image(fill_value=11)
image_2 = _create_image(fill_value=22)
with pytest.warns(SupervisionWarnings, match="deprecated"):
dataset = self._build_dataset({"img1.jpg": image_1, "img2.jpg": image_2})
with warnings.catch_warnings():
warnings.simplefilter("error", SupervisionWarnings)
train, test = dataset.split(split_ratio=0.5, shuffle=False)
assert train.image_paths == ["img1.jpg"]
assert test.image_paths == ["img2.jpg"]
_, loaded_train, _ = train[0]
_, loaded_test, _ = test[0]
np.testing.assert_array_equal(loaded_train, image_1)
np.testing.assert_array_equal(loaded_test, image_2)
def test_iteration_yields_in_memory_images(self) -> None:
"""Iteration yields (path, image, annotation) with correct pixels."""
images = {
@ -568,6 +591,45 @@ class TestDetectionDatasetExportCollisions:
annotations_directory_path=str(tmp_path / "annotations"),
)
def test_as_pascal_voc_rejects_annotation_collisions_before_writing(
self, tmp_path: Path
) -> None:
"""Pascal VOC export preflights annotation collisions before copying images."""
source_root = tmp_path / "source"
source_a = source_root / "dir_a"
source_b = source_root / "dir_b"
source_a.mkdir(parents=True)
source_b.mkdir(parents=True)
image_a_path = source_a / "img.jpg"
image_b_path = source_b / "img.png"
image_a_path.write_bytes(b"image-a")
image_b_path.write_bytes(b"image-b")
dataset = DetectionDataset(
classes=["cat"],
images=[str(image_a_path), str(image_b_path)],
annotations={
str(image_a_path): _create_detections(
xyxy=[[0, 0, 10, 10]], class_id=[0]
),
str(image_b_path): _create_detections(
xyxy=[[0, 0, 10, 10]], class_id=[0]
),
},
)
images_directory = tmp_path / "images"
annotations_directory = tmp_path / "annotations"
with pytest.raises(ValueError, match="both map to Pascal VOC annotation file"):
dataset.as_pascal_voc(
images_directory_path=str(images_directory),
annotations_directory_path=str(annotations_directory),
)
assert not images_directory.exists()
assert not annotations_directory.exists()
# ---------------------------------------------------------------------------
# TST-03 - DetectionDataset.split()

38
tests/detection/test_core.py
View File

@ -1631,6 +1631,44 @@ class TestDetectionsObbDispatch:
assert len(result) == 1
class TestDetectionsOverlapValidation:
"""`with_nms` and `with_nmm` require confidence and class IDs by default."""
@pytest.mark.parametrize(
"method",
[
pytest.param("with_nms", id="with_nms"),
pytest.param("with_nmm", id="with_nmm"),
],
)
def test_requires_confidence(self, method: str) -> None:
"""Missing confidence raises a descriptive `ValueError`."""
detections = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
class_id=np.array([0], dtype=int),
)
with pytest.raises(ValueError, match="Detections confidence must be given"):
getattr(detections, method)(threshold=0.5)
@pytest.mark.parametrize(
"method",
[
pytest.param("with_nms", id="with_nms"),
pytest.param("with_nmm", id="with_nmm"),
],
)
def test_requires_class_id_when_not_class_agnostic(self, method: str) -> None:
"""Missing class IDs raise a descriptive `ValueError`."""
detections = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
confidence=np.array([0.9], dtype=np.float32),
)
with pytest.raises(ValueError, match="Detections class_id must be given"):
getattr(detections, method)(threshold=0.5)
class TestGetAnchorsObbDispatch:
"""`get_anchors_coordinates` reads oriented corners when OBB data is present."""

46
tests/detection/test_from_adapters.py
View File

@ -2,7 +2,7 @@ import numpy as np
import pytest
import supervision.detection.core as detection_core
from supervision.config import CLASS_NAME_DATA_FIELD
from supervision.config import CLASS_NAME_DATA_FIELD, ORIENTED_BOX_COORDINATES
from supervision.detection.core import LMM, Detections
from supervision.detection.vlm import VLM
from supervision.utils.internal import SupervisionWarnings
@ -583,6 +583,21 @@ class TestFromEasyOCR:
assert len(det) == 1
assert float(det.confidence[0]) == pytest.approx(0.0)
def test_preserves_oriented_corners_in_data(self) -> None:
"""Quadrilateral EasyOCR boxes must be preserved in the data payload."""
bbox = [[0, 0], [8, 1], [7, 5], [1, 4]]
results = [(bbox, "text", 0.9)]
det = Detections.from_easyocr(results)
assert ORIENTED_BOX_COORDINATES in det.data
np.testing.assert_allclose(det.data[ORIENTED_BOX_COORDINATES], np.array([bbox]))
def test_detail_zero_results_raise_clear_error(self) -> None:
"""detail=0 EasyOCR results must fail with a descriptive ValueError."""
with pytest.raises(ValueError, match="detail=1"):
Detections.from_easyocr(["text"])
# ---------------------------------------------------------------------------
# from_azure_analyze_image
@ -629,7 +644,7 @@ class TestFromAzureAnalyzeImage:
np.testing.assert_allclose(det.xyxy[0], [0, 0, 10, 10])
def test_explicit_class_map_filters_unknown_classes(self) -> None:
"""With class_map, tags whose name is absent from the map are dropped."""
"""With class_map, the highest-confidence mapped tag is selected."""
class_map = {5: "cat"}
result = _make_azure_result(
[
@ -639,8 +654,8 @@ class TestFromAzureAnalyzeImage:
10,
10,
[
{"name": "unknown", "confidence": 0.95},
{"name": "cat", "confidence": 0.9},
{"name": "unknown", "confidence": 0.5},
],
),
]
@ -648,9 +663,32 @@ class TestFromAzureAnalyzeImage:
det = Detections.from_azure_analyze_image(result, class_map=class_map)
# Only 'cat' (id=5) survives; 'unknown' is filtered
assert len(det) == 1
assert int(det.class_id[0]) == 5
np.testing.assert_allclose(det.confidence, [0.9])
def test_unmapped_tags_warn_and_skip_detection(self) -> None:
"""With class_map, completely unmapped tags should warn before skipping."""
class_map = {5: "cat"}
result = _make_azure_result(
[
_make_azure_detection(
0,
0,
10,
10,
[
{"name": "unknown", "confidence": 0.95},
{"name": "other", "confidence": 0.9},
],
),
]
)
with pytest.warns(SupervisionWarnings, match="none of its tags matched"):
det = Detections.from_azure_analyze_image(result, class_map=class_map)
assert len(det) == 0
def test_empty_values_list_returns_empty_detections(self) -> None:
"""Zero detections in values list produce an empty Detections."""

39
tests/detection/test_from_sam.py
View File

@ -2,6 +2,7 @@ import numpy as np
import pytest
from supervision.detection.core import Detections
from supervision.detection.utils.converters import mask_to_rle
SERVERLESS_SAM3_DICT = {
"prompt_results": [
@ -109,6 +110,44 @@ def test_from_sam(
assert detections.mask is None
def test_from_sam_decodes_coco_rle_masks() -> None:
"""COCO RLE SAM outputs are decoded to dense boolean masks."""
small_mask = np.zeros((4, 4), dtype=bool)
small_mask[3, 3] = True
large_mask = np.zeros((4, 4), dtype=bool)
large_mask[:2, :2] = True
sam_result = [
{
"segmentation": {
"size": [4, 4],
"counts": mask_to_rle(small_mask, compressed=True),
},
"bbox": [3, 3, 1, 1],
"area": 1,
},
{
"segmentation": {
"size": [4, 4],
"counts": mask_to_rle(large_mask, compressed=True),
},
"bbox": [0, 0, 2, 2],
"area": 4,
},
]
detections = Detections.from_sam(sam_result=sam_result)
assert len(detections) == 2
assert isinstance(detections.mask, np.ndarray)
assert detections.mask.dtype == bool
assert detections.mask.shape == (2, 4, 4)
np.testing.assert_array_equal(detections.mask, np.stack([large_mask, small_mask]))
np.testing.assert_array_equal(
detections.xyxy,
np.array([[0, 0, 2, 2], [3, 3, 4, 4]], dtype=np.float32),
)
@pytest.mark.parametrize(
(
"sam3_result",

30
tests/detection/test_line_counter.py
View File

@ -915,7 +915,7 @@ def test_line_zone_trigger_evicts_stale_crossing_history() -> None:
line_zone.trigger(second_detections)
line_zone.trigger(second_detections)
assert set(line_zone.crossing_state_history) == {(1, 2)}
assert set(line_zone.crossing_state_history) == {1}
def test_line_zone_trigger_evicts_stale_crossing_history_on_empty_frames() -> None:
@ -931,7 +931,7 @@ def test_line_zone_trigger_evicts_stale_crossing_history_on_empty_frames() -> No
def test_line_zone_trigger_evicts_stale_crossing_history_on_class_change() -> None:
"""Class changes age out stale per-class crossing history."""
"""Class changes must not split a tracker crossing history."""
line_zone = LineZone(start=Point(0, 0), end=Point(10, 0))
first_detections = _create_detections(
xyxy=[[4, 4, 6, 6]], tracker_id=[0], class_id=[1]
@ -944,7 +944,31 @@ def test_line_zone_trigger_evicts_stale_crossing_history_on_class_change() -> No
for _ in range(line_zone.crossing_history_length):
line_zone.trigger(second_detections)
assert set(line_zone.crossing_state_history) == {(0, 2)}
assert set(line_zone.crossing_state_history) == {0}
def test_line_zone_class_flicker_keeps_crossing_counts_continuous() -> None:
"""A tracker's class change must not suppress a real crossing."""
line_zone = LineZone(start=Point(0, 0), end=Point(10, 0))
detections_sequence = [
_create_detections(xyxy=[[4, 4, 6, 6]], tracker_id=[0], class_id=[0]),
_create_detections(xyxy=[[4, -6, 6, -4]], tracker_id=[0], class_id=[1]),
_create_detections(xyxy=[[4, -6, 6, -4]], tracker_id=[0], class_id=[1]),
_create_detections(xyxy=[[4, 4, 6, 6]], tracker_id=[0], class_id=[1]),
]
crossed_in = []
crossed_out = []
for detections in detections_sequence:
crossed_in_frame, crossed_out_frame = line_zone.trigger(detections)
crossed_in.append(bool(crossed_in_frame[0]))
crossed_out.append(bool(crossed_out_frame[0]))
assert crossed_in == [False, True, False, False]
assert crossed_out == [False, False, False, True]
assert line_zone.in_count_per_class == {1: 1}
assert line_zone.out_count_per_class == {1: 1}
assert set(line_zone.crossing_state_history) == {0}
def test_line_zone_annotator_multiclass_supports_none_class_id() -> None:

25
tests/detection/test_polygonzone.py
View File

@ -116,7 +116,7 @@ class TestPolygonZoneTrigger:
from the original unclipped box, giving a single consistent position.
Setup: zone_left covers x 0-99, zone_right covers x 100-200.
Detection [60, 80, 140, 120] has BOTTOM_CENTER = ceil((60+140)/2), ceil(120)
Detection [60, 80, 140, 120] has BOTTOM_CENTER = round((60+140)/2), round(120)
= (100, 120), which falls in zone_right only.
"""
zone_left = sv.PolygonZone(
@ -145,7 +145,7 @@ class TestPolygonZoneTrigger:
np.array([[0, 0], [100, 0], [100, 100], [0, 100]]),
triggering_anchors=[sv.Position.CENTER],
)
# CENTER = (ceil((-50+0)/2), ceil((25+75)/2)) = (-25, 50) — x < 0.
# CENTER = (round((-50+0)/2), round((25+75)/2)) = (-25, 50) — x < 0.
detections = _create_detections(
xyxy=[[-50.0, 25.0, 0.0, 75.0]],
class_id=[0],
@ -157,10 +157,29 @@ class TestPolygonZoneTrigger:
"""An anchor landing exactly on a polygon corner is counted as inside."""
polygon = np.array([[0, 0], [100, 0], [100, 100], [0, 100]])
zone = sv.PolygonZone(polygon, triggering_anchors=[sv.Position.BOTTOM_RIGHT])
# BOTTOM_RIGHT = (ceil(x2), ceil(y2)) = (100, 100) — the polygon corner.
# BOTTOM_RIGHT = (round(x2), round(y2)) = (100, 100) — the polygon corner.
detections = _create_detections(
xyxy=[[50.0, 50.0, 100.0, 100.0]],
class_id=[0],
)
result = zone.trigger(detections)
assert result[0]
def test_half_pixel_anchor_uses_nearest_pixel(self) -> None:
"""Half-pixel anchors should not be biased toward the larger x and y."""
zone_left = sv.PolygonZone(
np.array([[0, 0], [100, 0], [100, 200], [0, 200]], dtype=np.int32)
)
zone_right = sv.PolygonZone(
np.array([[101, 0], [200, 0], [200, 200], [101, 200]], dtype=np.int32)
)
detections = _create_detections(
xyxy=[[60.0, 80.0, 141.0, 120.0]],
class_id=[0],
)
left_result = zone_left.trigger(detections)[0]
right_result = zone_right.trigger(detections)[0]
assert left_result
assert not right_result

61
tests/detection/test_vlm.py
View File

@ -4,6 +4,7 @@ from contextlib import nullcontext as does_not_raise
import numpy as np
import pytest
import supervision.detection.core as detection_core
from supervision.config import CLASS_NAME_DATA_FIELD
from supervision.detection.core import Detections
from supervision.detection.vlm import (
@ -991,6 +992,39 @@ def test_florence_2(
np.testing.assert_array_equal(result[3], expected_results[3])
@pytest.mark.parametrize(
("florence_result", "match"),
[
pytest.param(
{
"<REGION_TO_CATEGORY>": (
"some object<loc_300><loc_400><loc_500><loc_600>"
),
"<REGION_TO_DESCRIPTION>": "other",
},
"single element",
id="multiple-top-level-tasks",
),
pytest.param(
{"<REGION_TO_CATEGORY>": 123},
"Expected string as <REGION_TO_CATEGORY> result",
id="non-string-region-result",
),
pytest.param(
{"<REGION_TO_CATEGORY>": "some object"},
"Expected string to end in location tags",
id="missing-location-tags",
),
],
)
def test_florence_2_invalid_payloads_raise_value_error(
florence_result: dict[str, object], match: str
) -> None:
"""Malformed Florence 2 region payloads raise `ValueError`."""
with pytest.raises(ValueError, match=match):
from_florence_2(florence_result, (10, 10))
@pytest.mark.parametrize(
("exception", "result", "resolution_wh", "classes", "expected_results"),
[
@ -1390,6 +1424,33 @@ def test_from_google_gemini_2_5_malformed_mask_keeps_confidence_aligned():
assert masks.shape == (2, 480, 640)
def test_from_vlm_unsupported_future_enum_raises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Unknown VLM members should raise instead of returning empty detections."""
class FakeVLM:
PALIGEMMA = object()
FLORENCE_2 = object()
QWEN_2_5_VL = object()
QWEN_3_VL = object()
DEEPSEEK_VL_2 = object()
GOOGLE_GEMINI_2_0 = object()
GOOGLE_GEMINI_2_5 = object()
MOONDREAM = object()
FUTURE = object()
monkeypatch.setattr(detection_core, "VLM", FakeVLM)
monkeypatch.setattr(
detection_core,
"_validate_vlm_parameters",
lambda vlm, result, kwargs: vlm,
)
with pytest.raises(ValueError, match="Unsupported VLM value"):
Detections.from_vlm(vlm=FakeVLM.FUTURE, result="ignored")
@pytest.mark.parametrize(
("parser", "result"),
[

22
tests/detection/tools/test_inference_slicer.py
View File

@ -6,7 +6,10 @@ import pytest
from supervision.config import ORIENTED_BOX_COORDINATES
from supervision.detection.core import Detections
from supervision.detection.tools.inference_slicer import InferenceSlicer
from supervision.detection.tools.inference_slicer import (
InferenceSlicer,
move_detections,
)
from supervision.detection.utils.iou_and_nms import OverlapFilter
from supervision.utils.internal import SupervisionWarnings
@ -696,3 +699,20 @@ class TestInferenceSlicerBatch:
)
with pytest.warns(SupervisionWarnings, match="outside the slice bounds"):
slicer(image)
def test_move_detections_returns_a_copy(self) -> None:
"""move_detections must not mutate the caller's Detections object."""
detections = Detections(
xyxy=np.array([[1.0, 2.0, 3.0, 4.0]], dtype=np.float32),
class_id=np.array([0]),
)
original_xyxy = detections.xyxy.copy()
moved = move_detections(
detections=detections,
offset=np.array([10, 20]),
resolution_wh=(100, 100),
)
np.testing.assert_array_equal(detections.xyxy, original_xyxy)
np.testing.assert_array_equal(moved.xyxy, np.array([[11.0, 22.0, 13.0, 24.0]]))

64
tests/detection/tools/test_smoother.py
View File

@ -63,25 +63,33 @@ class TestDetectionsSmoother:
assert smoothed.confidence is not None
assert_allclose(smoothed.confidence, expected_confidence, atol=1e-5)
def test_smoother_multi_track_mixed_confidence_does_not_crash(self) -> None:
"""Two tracks with different confidence presence must not raise on merge."""
def test_smoother_reappearing_track_keeps_history(self) -> None:
"""Missing tracks stay silent but still contribute when they return."""
smoother = DetectionsSmoother(length=3)
smoother.update_with_detections(
Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
confidence=np.array([0.5]),
tracker_id=np.array([1]),
)
first = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
confidence=np.array([0.5]),
tracker_id=np.array([1]),
)
smoothed = smoother.update_with_detections(
Detections(
xyxy=np.array([[20, 20, 30, 30]], dtype=np.float32),
tracker_id=np.array([2]),
)
missing = Detections(
xyxy=np.empty((0, 4), dtype=np.float32),
tracker_id=np.array([], dtype=int),
)
returned = Detections(
xyxy=np.array([[2, 2, 12, 12]], dtype=np.float32),
confidence=np.array([0.7]),
tracker_id=np.array([1]),
)
assert len(smoothed) == 2
assert smoothed.confidence is None
smoother.update_with_detections(first)
smoothed_missing = smoother.update_with_detections(missing)
smoothed_returned = smoother.update_with_detections(returned)
assert len(smoothed_missing) == 0
assert len(smoothed_returned) == 1
assert smoothed_returned.confidence is not None
assert_allclose(smoothed_returned.xyxy, np.array([[1, 1, 11, 11]]), atol=1e-5)
assert_allclose(smoothed_returned.confidence, np.array([0.6]), atol=1e-5)
def test_smoother_tracker_id_none_warns_and_returns_unchanged(self) -> None:
"""update_with_detections warns and returns input when tracker_id is None."""
@ -124,3 +132,29 @@ class TestDetectionsSmoother:
assert_allclose(smoothed.xyxy, np.array([[3, 3, 13, 13]]), atol=1e-5)
assert smoothed.confidence is not None
assert_allclose(smoothed.confidence, np.array([0.6]), atol=1e-5)
def test_smoother_does_not_emit_missing_tracks(self) -> None:
"""A missing track should keep history but stop emitting ghost boxes."""
smoother = DetectionsSmoother(length=3)
first = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
confidence=np.array([0.3]),
tracker_id=np.array([1]),
)
missing = Detections(
xyxy=np.empty((0, 4), dtype=np.float32),
tracker_id=np.array([], dtype=int),
)
second = Detections(
xyxy=np.array([[2, 2, 12, 12]], dtype=np.float32),
confidence=np.array([0.9]),
tracker_id=np.array([1]),
)
smoother.update_with_detections(first)
smoothed_missing = smoother.update_with_detections(missing)
smoothed_returned = smoother.update_with_detections(second)
assert len(smoothed_missing) == 0
assert smoothed_returned.confidence is not None
assert_allclose(smoothed_returned.xyxy, np.array([[1, 1, 11, 11]]), atol=1e-5)

15
tests/detection/tools/test_transformers.py
View File

@ -29,10 +29,10 @@ from tests.helpers import _FakeDetachTensor, make_panoptic_png
class TestPngStringToSegmentationArray:
"""png_string_to_segmentation_array extracts the red channel as a label map."""
"""png_string_to_segmentation_array decodes RGB-encoded panoptic IDs."""
def test_extracts_red_channel_as_segment_ids(self) -> None:
"""RGBA PNG: red channel values become the returned label array."""
def test_extracts_rgb_channels_as_segment_ids(self) -> None:
"""RGBA PNG: RGB channels become the returned label array."""
seg_map = np.array([[1, 2], [3, 0]], dtype=np.uint8)
png_bytes = make_panoptic_png(seg_map)
@ -40,6 +40,15 @@ class TestPngStringToSegmentationArray:
np.testing.assert_array_equal(result, seg_map)
def test_decodes_segment_ids_above_255(self) -> None:
"""RGB panoptic encoding preserves segment IDs beyond one byte."""
seg_map = np.array([[1, 257], [513, 0]], dtype=np.uint32)
png_bytes = make_panoptic_png(seg_map)
result = png_string_to_segmentation_array(png_bytes)
np.testing.assert_array_equal(result, seg_map)
def test_returns_array_of_shape_h_w(self) -> None:
"""Output shape matches the image height and width."""
seg_map = np.zeros((6, 8), dtype=np.uint8)

68
tests/detection/utils/test_converters.py
View File

@ -318,17 +318,25 @@ def test_xyxy_to_mask(boxes: np.ndarray, resolution_wh, expected: np.ndarray) ->
np.testing.assert_array_equal(result, expected)
def _mask_to_xyxy_reference(masks: np.ndarray) -> np.ndarray:
def _mask_to_xyxy_reference(
masks: np.ndarray, coordinate_convention: str = "inclusive"
) -> np.ndarray:
"""Per-mask `np.where` loop used as a ground-truth oracle."""
xyxy = np.zeros((masks.shape[0], 4), dtype=int)
for i, mask in enumerate(masks):
rows, cols = np.where(mask)
if len(rows) > 0 and len(cols) > 0:
if coordinate_convention == "exclusive":
x_max = int(cols.max()) + 1
y_max = int(rows.max()) + 1
else:
x_max = int(cols.max())
y_max = int(rows.max())
xyxy[i, :] = [
int(cols.min()),
int(rows.min()),
int(cols.max()),
int(rows.max()),
x_max,
y_max,
]
return xyxy
@ -397,6 +405,52 @@ class TestMaskToXyxy:
assert result.dtype == reference.dtype
np.testing.assert_array_equal(result, reference)
def test_mask_to_xyxy_exclusive_matches_reference(self) -> None:
"""Exclusive bounds should return one-past-the-end coordinates."""
masks = np.array(
[
[[False, False], [False, True]],
[[True, True], [True, True]],
],
dtype=bool,
)
result = mask_to_xyxy(masks, coordinate_convention="exclusive")
reference = _mask_to_xyxy_reference(masks, coordinate_convention="exclusive")
np.testing.assert_array_equal(result, reference)
def test_xyxy_to_mask_exclusive_round_trip(self) -> None:
"""Exclusive boxes should round-trip through `xyxy_to_mask`."""
boxes = np.array([[1, 1, 3, 3], [0, 0, 2, 1]], dtype=float)
result = xyxy_to_mask(
boxes=boxes,
resolution_wh=(4, 4),
coordinate_convention="exclusive",
)
expected = np.array(
[
[
[False, False, False, False],
[False, True, True, False],
[False, True, True, False],
[False, False, False, False],
],
[
[True, True, False, False],
[False, False, False, False],
[False, False, False, False],
[False, False, False, False],
],
],
dtype=bool,
)
assert result.dtype == np.bool_
np.testing.assert_array_equal(result, expected)
@pytest.mark.parametrize(
("mask", "compressed", "expected_rle", "exception"),
@ -458,14 +512,14 @@ class TestMaskToXyxy:
np.array([[[]]]).astype(bool),
False,
None,
pytest.raises(AssertionError, match="Input mask must be 2D"),
), # raises AssertionError because mask dimensionality is not 2D
pytest.raises(ValueError, match="Input mask must be 2D"),
), # raises ValueError because mask dimensionality is not 2D
(
np.array([[]]).astype(bool),
False,
None,
pytest.raises(AssertionError, match="Input mask cannot be empty"),
), # raises AssertionError because mask is empty
pytest.raises(ValueError, match="Input mask cannot be empty"),
), # raises ValueError because mask is empty
],
)
def test_mask_to_rle(

27
tests/detection/utils/test_internal.py
View File

@ -1081,13 +1081,27 @@ def test_get_data_item(
(
[{"key1": [1, 2, 3]}, {"key1": np.array([1, 2, 3])}],
None,
pytest.raises(ValueError, match="type\\(value\\)"),
pytest.raises(
ValueError,
match=(
r"Conflicting metadata for key: 'key1': "
r"(?:<class 'list'>, <class 'numpy\.ndarray'>|"
r"<class 'numpy\.ndarray'>, <class 'list'>)\."
),
),
),
# Empty lists and numpy arrays for the same key
(
[{"key1": []}, {"key1": np.array([])}],
None,
pytest.raises(ValueError, match="type\\(other_value\\)"),
pytest.raises(
ValueError,
match=(
r"Conflicting metadata for key: 'key1': "
r"(?:<class 'list'>, <class 'numpy\.ndarray'>|"
r"<class 'numpy\.ndarray'>, <class 'list'>)\."
),
),
),
# Identical multi-dimensional lists across metadata dictionaries
(
@ -1123,7 +1137,14 @@ def test_get_data_item(
(
[{"key1": [[1, 2], [3, 4]]}, {"key1": np.arange(4).reshape(2, 2)}],
None,
pytest.raises(ValueError, match="type\\(value\\)"),
pytest.raises(
ValueError,
match=(
r"Conflicting metadata for key: 'key1': "
r"(?:<class 'list'>, <class 'numpy\.ndarray'>|"
r"<class 'numpy\.ndarray'>, <class 'list'>)\."
),
),
),
# Identical higher-dimensional (3D) numpy arrays across
# metadata dictionaries

63
tests/detection/utils/test_iou_and_nms.py
View File

@ -1,5 +1,6 @@
from __future__ import annotations
from collections.abc import Callable
from contextlib import ExitStack as DoesNotRaise
import numpy as np
@ -1674,6 +1675,68 @@ class TestOrientedBoxNonMaxMerge:
assert sorted_groups == [[0, 1], [2]]
class TestIouThresholdValidation:
"""Invalid IoU thresholds must raise `ValueError` on public NMS/NMM APIs."""
@pytest.mark.parametrize(
("function", "kwargs"),
[
pytest.param(
box_non_max_suppression,
{"predictions": np.empty((0, 5), dtype=np.float32)},
id="box-nms",
),
pytest.param(
box_non_max_merge,
{"predictions": np.empty((0, 5), dtype=np.float32)},
id="box-nmm",
),
pytest.param(
mask_non_max_suppression,
{
"predictions": np.empty((0, 5), dtype=np.float32),
"masks": np.empty((0, 1, 1), dtype=bool),
},
id="mask-nms",
),
pytest.param(
mask_non_max_merge,
{
"predictions": np.empty((0, 5), dtype=np.float32),
"masks": np.empty((0, 1, 1), dtype=bool),
},
id="mask-nmm",
),
pytest.param(
oriented_box_non_max_suppression,
{
"predictions": np.empty((0, 5), dtype=np.float32),
"oriented_boxes": np.empty((0, 4, 2), dtype=np.float32),
},
id="obb-nms",
),
pytest.param(
oriented_box_non_max_merge,
{
"predictions": np.empty((0, 5), dtype=np.float32),
"oriented_boxes": np.empty((0, 4, 2), dtype=np.float32),
},
id="obb-nmm",
),
],
)
@pytest.mark.parametrize("iou_threshold", [-0.1, 1.1])
def test_rejects_thresholds_outside_closed_unit_interval(
self,
function: Callable[..., object],
kwargs: dict[str, object],
iou_threshold: float,
) -> None:
"""Each public overlap filter rejects thresholds outside [0, 1]."""
with pytest.raises(ValueError, match="closed range from 0 to 1"):
function(iou_threshold=iou_threshold, **kwargs)
def _naive_mask_iou(
masks_true: np.ndarray,
masks_detection: np.ndarray,

7
tests/detection/utils/test_masks.py
View File

@ -13,6 +13,7 @@ from supervision.detection.utils.masks import (
contains_holes,
contains_multiple_segments,
filter_segments_by_distance,
mask_to_roi,
move_masks,
resize_masks,
)
@ -21,6 +22,12 @@ from supervision.detection.utils.masks import (
class TestMaskROIHelpers:
"""Tests for _mask_to_roi, _compact_masks_to_roi, _masks_to_roi helpers."""
def test_mask_to_roi_public_helper_exposes_exclusive_bounds(self) -> None:
"""Public mask_to_roi should return slice-friendly exclusive bounds."""
mask = np.zeros((10, 15), dtype=bool)
mask[3, 5] = True
assert mask_to_roi(mask) == (5, 3, 6, 4)
def test_mask_to_roi_all_false_returns_none(self):
"""All-false mask should return None."""
mask = np.zeros((10, 15), dtype=bool)

14
tests/helpers.py
View File

@ -19,13 +19,17 @@ from supervision.key_points.core import KeyPoints
def make_panoptic_png(segment_map: np.ndarray) -> bytes:
"""Encode a (H, W) uint8 segment-ID array as a 4-channel RGBA PNG byte string.
"""Encode a segment-ID array as a 24-bit RGBA PNG byte string.
The segment IDs are stored in the red channel (channel 0). Used by
panoptic segmentation tests that construct PNG-encoded segment maps.
Segment IDs are stored in RGB little-endian order so tests can cover
panoptic labels above 255 without collisions.
"""
arr = np.zeros((*segment_map.shape, 4), dtype=np.uint8)
arr[:, :, 0] = segment_map.astype(np.uint8)
segment_map_u32 = np.asarray(segment_map, dtype=np.uint32)
arr = np.zeros((*segment_map_u32.shape, 4), dtype=np.uint8)
arr[:, :, 0] = (segment_map_u32 & 0xFF).astype(np.uint8)
arr[:, :, 1] = ((segment_map_u32 >> 8) & 0xFF).astype(np.uint8)
arr[:, :, 2] = ((segment_map_u32 >> 16) & 0xFF).astype(np.uint8)
arr[:, :, 3] = 255
buf = io.BytesIO()
Image.fromarray(arr).save(buf, format="PNG")
return buf.getvalue()

17
tests/key_points/test_docs.py Normal file
View File

@ -0,0 +1,17 @@
"""Regression tests for docs/keypoint/annotators.md."""
from pathlib import Path
_REPO_ROOT = Path(__file__).resolve().parent
while not (_REPO_ROOT / "pyproject.toml").exists():
_REPO_ROOT = _REPO_ROOT.parent
REPO_ROOT = _REPO_ROOT
def test_keypoint_annotators_doc_mentions_vertex_ellipse_alias() -> None:
"""The keypoint docs must mention the VertexEllipseAnnotator alias."""
docs_path = REPO_ROOT / "docs" / "keypoint" / "annotators.md"
content = docs_path.read_text(encoding="utf-8")
assert "VertexEllipseAnnotator" in content

89
tests/metrics/test_lazy_imports.py Normal file
View File

@ -0,0 +1,89 @@
"""Regression tests for lazy metric plotting imports."""
from __future__ import annotations
import importlib
import sys
from types import ModuleType
from unittest.mock import MagicMock
import numpy as np
from supervision.detection.core import Detections
def _clear_metric_modules() -> None:
"""Remove cached metric modules so import-time side effects stay visible."""
for module_name in [
"supervision.metrics",
"supervision.metrics.detection",
"supervision.metrics.f1_score",
"supervision.metrics.mean_average_precision",
"supervision.metrics.mean_average_recall",
"supervision.metrics.precision",
"supervision.metrics.recall",
]:
sys.modules.pop(module_name, None)
def _make_pyplot_stub() -> ModuleType:
"""Build a minimal pyplot stub for plot smoke tests."""
pyplot = ModuleType("matplotlib.pyplot")
pyplot.rcParams = {}
figure = MagicMock(name="figure")
axis = MagicMock(name="axis")
bar = MagicMock(name="bar")
bar.get_height.return_value = 1.0
bar.get_x.return_value = 0.0
bar.get_width.return_value = 1.0
axis.bar.return_value = [bar]
pyplot.subplots = MagicMock(return_value=(figure, axis))
pyplot.tight_layout = MagicMock()
pyplot.show = MagicMock()
pyplot.setp = MagicMock()
return pyplot
def test_metrics_package_import_keeps_pyplot_lazy() -> None:
"""Importing supervision.metrics must not pull in matplotlib.pyplot."""
sys.modules.pop("matplotlib.pyplot", None)
_clear_metric_modules()
importlib.import_module("supervision.metrics")
assert "matplotlib.pyplot" not in sys.modules
def test_precision_plot_imports_pyplot_on_demand(monkeypatch) -> None:
"""Precision.plot should import pyplot only when plotting is requested."""
sys.modules.pop("matplotlib.pyplot", None)
_clear_metric_modules()
precision_module = importlib.import_module("supervision.metrics.precision")
assert "matplotlib.pyplot" not in sys.modules
pyplot = _make_pyplot_stub()
monkeypatch.setitem(sys.modules, "matplotlib.pyplot", pyplot)
import matplotlib
monkeypatch.setattr(matplotlib, "pyplot", pyplot, raising=False)
predictions = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
confidence=np.array([0.9], dtype=np.float32),
class_id=np.array([0]),
)
targets = Detections(
xyxy=np.array([[0, 0, 10, 10]], dtype=np.float32),
class_id=np.array([0]),
)
result = precision_module.Precision().update(predictions, targets).compute()
result.plot()
pyplot.subplots.assert_called_once()
pyplot.show.assert_called_once()

2
tests/test_public_api.py
View File

@ -11,6 +11,8 @@ import supervision as sv
)
def test_all_symbols_are_importable(symbol_name: str) -> None:
"""Every name in supervision.__all__ must be a non-None accessible attribute."""
if symbol_name == "ByteTrack":
sv.__dict__.pop("ByteTrack", None)
val = getattr(sv, symbol_name, None)
assert val is not None, (
f"supervision.{symbol_name} is listed in __all__ but not accessible "

35
tests/test_validate_deprecations.py
View File

@ -1,5 +1,9 @@
import os
import subprocess
import sys
import warnings
from collections.abc import Callable
from pathlib import Path
import numpy as np
import pytest
@ -115,3 +119,34 @@ def test_private_validation_paths_do_not_warn() -> None:
warning for warning in recorded_warnings if warning.category is FutureWarning
]
assert future_warnings == []
def test_import_supervision_stays_silent_about_bytetrack() -> None:
"""Plain supervision import should not surface the ByteTrack warning."""
repo_root = Path(__file__).resolve().parents[1]
env = os.environ.copy()
env["PYTHONPATH"] = str(repo_root / "src")
script = """
import warnings
with warnings.catch_warnings(record=True) as recorded:
warnings.simplefilter("always")
import supervision
byte_track_warnings = [
warning
for warning in recorded
if warning.category is FutureWarning and "ByteTrack" in str(warning.message)
]
raise SystemExit(1 if byte_track_warnings else 0)
"""
completed = subprocess.run( # noqa: S603 - trusted fixed command in a test helper.
[sys.executable, "-c", script],
check=False,
capture_output=True,
text=True,
env=env,
)
assert completed.returncode == 0, completed.stderr

7
tests/tracker/test_byte_tracker.py
View File

@ -18,6 +18,13 @@ def _detections_from_boxes(
)
def test_top_level_bytetrack_access_returns_class() -> None:
"""Top-level ByteTrack access should still resolve to the class object."""
sv.__dict__.pop("ByteTrack", None)
tracker_cls = sv.ByteTrack
assert tracker_cls is not None
@pytest.mark.parametrize(
("detections", "expected_results"),
[

11
tests/utils/test_conversion.py
View File

@ -108,6 +108,17 @@ def test_pillow_to_cv2(
)
def test_pillow_to_cv2_handles_palette_images() -> None:
"""Palette images must resolve their palette colors before BGR conversion."""
image = Image.new("P", (1, 1))
image.putpalette([0, 0, 0, 255, 0, 0] + [0, 0, 0] * 254)
image.putdata([1])
result = pillow_to_cv2(image=image)
np.testing.assert_array_equal(result, np.array([[[0, 0, 255]]], dtype=np.uint8))
def test_images_to_cv2_when_empty_input_provided() -> None:
# when
result = images_to_cv2(images=[])

14
tests/utils/test_image.py
View File

@ -255,6 +255,20 @@ def test_crop_image(image, xyxy, expected_size) -> None:
assert cropped.size == expected_size
def test_crop_image_clips_out_of_bounds_coordinates() -> None:
"""Out-of-bounds crops must clip consistently for NumPy and Pillow inputs."""
image_np = np.arange(16, dtype=np.uint8).reshape(4, 4)
image_pil = Image.fromarray(image_np)
xyxy = (-2, -1, 3, 3)
expected = image_np[0:3, 0:3]
expected_pil = np.repeat(expected[:, :, None], 3, axis=2)
np.testing.assert_array_equal(crop_image(image=image_np, xyxy=xyxy), expected)
np.testing.assert_array_equal(
np.asarray(crop_image(image=image_pil, xyxy=xyxy)), expected_pil
)
@pytest.mark.parametrize(
("image", "expected"),
[

335
tests/utils/test_video.py
View File

@ -1,5 +1,9 @@
import os
import shutil
from pathlib import Path
from queue import Empty, Full
from queue import Queue as StdQueue
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import cv2
@ -7,6 +11,7 @@ import numpy as np
import pytest
from supervision.utils.video import (
FPSMonitor,
VideoInfo,
_mux_audio,
get_video_frames_generator,
@ -94,6 +99,321 @@ def test_process_video_exception_with_small_buffer(dummy_video_path, tmp_path) -
)
def test_process_video_enqueues_writer_sentinel_with_timeout(
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""process_video enqueues the writer sentinel and bounded worker joins."""
read_queue = StdQueue()
read_queue.put((0, np.zeros((2, 2, 3), dtype=np.uint8)))
read_queue.put((1, np.zeros((2, 2, 3), dtype=np.uint8)))
read_queue.put(None)
class RecordingWriteQueue:
"""Record writer queue puts so the shutdown path can be asserted."""
def __init__(self) -> None:
"""Initialize the queue call log."""
self.put_calls: list[tuple[object, object]] = []
def put(self, item: object, timeout: object | None = None) -> None:
"""Record each put call and its timeout."""
self.put_calls.append((item, timeout))
def get(self, timeout: object | None = None) -> object:
"""The writer thread is disabled, so reads are not expected."""
raise AssertionError("writer queue should not be read in this test")
join_calls: list[object | None] = []
class FakeThread:
"""Thread stand-in that keeps the test single-threaded."""
def __init__(
self,
target: object,
args: tuple[object, ...] = (),
daemon: bool = False,
) -> None:
"""Store the thread target without starting it."""
self.target = target
self.args = args
self.daemon = daemon
def start(self) -> None:
"""Do nothing; the test preloads the queues instead."""
def join(self, timeout=None) -> None:
"""Do nothing; the worker targets are intentionally never started."""
join_calls.append(timeout)
class FakeVideoSink:
"""Minimal sink context manager used to verify shutdown ordering."""
def __init__(self, target_path: str, video_info: object) -> None:
"""Store constructor arguments for completeness."""
self.target_path = target_path
self.video_info = video_info
def __enter__(self) -> "FakeVideoSink":
"""Return the sink context manager."""
return self
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
"""Propagate any exception without side effects."""
return None
def write_frame(self, frame: object) -> None:
"""The writer thread is disabled in this test."""
write_queue = RecordingWriteQueue()
queue_factory_calls = iter([read_queue, write_queue])
monkeypatch.setattr(
"supervision.utils.video.Queue",
lambda *args, **kwargs: next(queue_factory_calls),
)
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
monkeypatch.setattr(
"supervision.utils.video.VideoInfo.from_video_path",
lambda video_path: SimpleNamespace(total_frames=2),
)
target_path = str(tmp_path / "target_sentinel.mp4")
def callback(frame, index):
if index == 1:
raise ValueError("Test exception at frame 1")
return frame
with pytest.raises(ValueError, match="Test exception at frame 1"):
process_video(
source_path=dummy_video_path,
target_path=target_path,
callback=callback,
show_progress=False,
)
assert write_queue.put_calls[-1] == (None, 1)
assert all(timeout is None for _item, timeout in write_queue.put_calls[:-1])
assert join_calls == [10, 10]
def test_process_video_best_effort_sentinel_handles_full_queue(
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""process_video should not hang if the writer queue is already full."""
read_queue = StdQueue()
read_queue.put((0, np.zeros((2, 2, 3), dtype=np.uint8)))
read_queue.put(None)
class FullWriteQueue:
"""Record writer queue puts and fail the shutdown sentinel."""
def __init__(self) -> None:
"""Initialize the queue call log."""
self.put_calls: list[tuple[object, object | None]] = []
def put(self, item: object, timeout: object | None = None) -> None:
"""Record the put and raise Full for the shutdown sentinel."""
self.put_calls.append((item, timeout))
if item is None:
raise Full
def get(self, timeout: object | None = None) -> object:
"""The writer thread is disabled, so reads are not expected."""
raise AssertionError("writer queue should not be read in this test")
join_calls: list[object | None] = []
class FakeThread:
"""Thread stand-in that keeps the test single-threaded."""
def __init__(
self,
target: object,
args: tuple[object, ...] = (),
daemon: bool = False,
) -> None:
"""Store the thread target without starting it."""
self.target = target
self.args = args
self.daemon = daemon
def start(self) -> None:
"""Do nothing; the test preloads the queues instead."""
def join(self, timeout=None) -> None:
"""Record join timeouts for shutdown verification."""
join_calls.append(timeout)
class FakeVideoSink:
"""Minimal sink context manager used to verify shutdown ordering."""
def __init__(self, target_path: str, video_info: object) -> None:
"""Store constructor arguments for completeness."""
self.target_path = target_path
self.video_info = video_info
def __enter__(self) -> "FakeVideoSink":
"""Return the sink context manager."""
return self
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
"""Propagate any exception without side effects."""
return None
def write_frame(self, frame: object) -> None:
"""The writer thread is disabled in this test."""
write_queue = FullWriteQueue()
queue_factory_calls = iter([read_queue, write_queue])
monkeypatch.setattr(
"supervision.utils.video.Queue",
lambda *args, **kwargs: next(queue_factory_calls),
)
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
monkeypatch.setattr(
"supervision.utils.video.VideoInfo.from_video_path",
lambda video_path: SimpleNamespace(total_frames=1),
)
target_path = str(tmp_path / "target_full_queue.mp4")
def callback(frame, index):
raise ValueError("Test exception at frame 0")
with pytest.raises(ValueError, match="Test exception at frame 0"):
process_video(
source_path=dummy_video_path,
target_path=target_path,
callback=callback,
show_progress=False,
)
assert write_queue.put_calls[-1] == (None, 1)
assert join_calls == [10, 10]
def test_process_video_waits_for_reader_timeout_when_queue_is_empty(
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""process_video should keep waiting briefly when the reader queue times out."""
class TimeoutReadQueue:
"""Record the first frame read and then time out in shutdown."""
def __init__(self) -> None:
"""Initialize the queue call log."""
self.get_calls: list[object | None] = []
def put(self, item: object, timeout: object | None = None) -> None:
"""The reader thread is disabled, so writes are not expected."""
raise AssertionError("reader queue should not be written in this test")
def get(self, timeout: object | None = None) -> object:
"""Yield one frame during processing, then time out during shutdown."""
self.get_calls.append(timeout)
if timeout is None:
return (0, np.zeros((2, 2, 3), dtype=np.uint8))
raise Empty
class RecordingWriteQueue:
"""Record writer queue puts so the shutdown path can be asserted."""
def __init__(self) -> None:
"""Initialize the queue call log."""
self.put_calls: list[tuple[object, object | None]] = []
def put(self, item: object, timeout: object | None = None) -> None:
"""Record each put call and its timeout."""
self.put_calls.append((item, timeout))
def get(self, timeout: object | None = None) -> object:
"""The writer thread is disabled, so reads are not expected."""
raise AssertionError("writer queue should not be read in this test")
join_calls: list[object | None] = []
reader_alive_states = iter([True, False])
class FakeThread:
"""Thread stand-in that keeps the test single-threaded."""
def __init__(
self,
target: object,
args: tuple[object, ...] = (),
daemon: bool = False,
) -> None:
"""Store the thread target without starting it."""
self.target = target
self.args = args
self.daemon = daemon
def start(self) -> None:
"""Do nothing; the test preloads the queues instead."""
def join(self, timeout=None) -> None:
"""Record join timeouts for shutdown verification."""
join_calls.append(timeout)
def is_alive(self) -> bool:
"""Return a short-lived alive state so the timeout branch is hit."""
return next(reader_alive_states, False)
class FakeVideoSink:
"""Minimal sink context manager used to verify shutdown ordering."""
def __init__(self, target_path: str, video_info: object) -> None:
"""Store constructor arguments for completeness."""
self.target_path = target_path
self.video_info = video_info
def __enter__(self) -> "FakeVideoSink":
"""Return the sink context manager."""
return self
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
"""Propagate any exception without side effects."""
return None
def write_frame(self, frame: object) -> None:
"""The writer thread is disabled in this test."""
read_queue = TimeoutReadQueue()
write_queue = RecordingWriteQueue()
queue_factory_calls = iter([read_queue, write_queue])
monkeypatch.setattr(
"supervision.utils.video.Queue",
lambda *args, **kwargs: next(queue_factory_calls),
)
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
monkeypatch.setattr(
"supervision.utils.video.VideoInfo.from_video_path",
lambda video_path: SimpleNamespace(total_frames=1),
)
target_path = str(tmp_path / "target_timeout.mp4")
def callback(frame, index):
raise ValueError("Test exception at frame 0")
with pytest.raises(ValueError, match="Test exception at frame 0"):
process_video(
source_path=dummy_video_path,
target_path=target_path,
callback=callback,
show_progress=False,
)
assert read_queue.get_calls == [None, 1, 1]
assert join_calls == [10, 10]
def test_process_video_max_frames(dummy_video_path, tmp_path) -> None:
"""
Verify that process_video respects the max_frames parameter.
@ -249,6 +569,21 @@ def test_get_video_frames_generator_with_stride(dummy_video_path) -> None:
assert len(frames) == 5
def test_fps_monitor_uses_frame_intervals(monkeypatch) -> None:
"""FPSMonitor must divide elapsed time by intervals, not sample count."""
timestamps = iter([0.0, 0.5, 1.0])
monkeypatch.setattr(
"supervision.utils.video.time.monotonic", lambda: next(timestamps)
)
fps_monitor = FPSMonitor()
fps_monitor.tick()
fps_monitor.tick()
fps_monitor.tick()
assert fps_monitor.fps == pytest.approx(2.0)
def test_process_video_preserve_audio_calls_mux(dummy_video_path, tmp_path) -> None:
"""
Verify that process_video calls _mux_audio when preserve_audio=True.

1
tests/validators/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for `supervision.validators`."""

16
tests/validators/test_shims.py Normal file
View File

@ -0,0 +1,16 @@
"""Smoke tests for the validators module surface."""
import warnings
import numpy as np
import supervision.validators as validators
def test_private_validate_xyxy_does_not_warn() -> None:
"""The private validator path stays quiet for a valid input."""
with warnings.catch_warnings(record=True) as captured:
warnings.simplefilter("always")
validators._validate_xyxy(np.array([[0, 0, 1, 1]]))
assert captured == []