Fix detection medium review findings (#2400)

- Fixed detection medium findings across adapters, mask non-max merge, sinks, segmentation parsing, LineZone history, and mask ROI handling
- Fixed mask non-max merge deprecation warnings to honor the standard warning opt-out and include version context
- Fixed mask non-max merge validation for invalid IoU thresholds
- Fixed CompactMask non-max merge grouping to update merged mask candidates correctly
- Fixed selected and compacted detections to copy arrays and metadata, preventing mutations from leaking back to source detections
- Fixed LineZone crossing history eviction to tolerate short tracking gaps and evict stale state per tracker/class key
- Fixed semantic segmentation handling to preserve class ID 0
- Improved mask ROI conversion performance by avoiding unnecessary full-frame copies and repeated scans
- Updated JSONSink changelog/docs to document native bool/int/float output while leaving CSVSink unchanged
- Updated detection docstrings for mask parsing, selection copy semantics, validation errors, and argument readability guidance

---------

Co-authored-by: Codex <codex@openai.com>
Co-authored-by: claude[bot] <209825114+claude[bot]@users.noreply.github.com>
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
Jirka Borovec 2026-07-06 19:02:38 +02:00 committed by GitHub
parent 8fd47a9ec6
commit e13090f84b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 812 additions and 185 deletions

View File

@ -246,6 +246,10 @@ Every docstring should include a usage example. When the example only uses `supe
Type hints are required on all new code. mypy is enforced by the pre-commit hook configured in `.pre-commit-config.yaml` — your PR will fail CI if mypy reports errors.
### Readability
Avoid multi-branch conditional expressions inside function or constructor arguments. If an argument needs more than a simple `a if condition else b`, assign it to a named local variable before the call.
### Performance
- Avoid unnecessary copies of NumPy arrays.

View File

@ -54,6 +54,8 @@ These supplement [CONTRIBUTING.md](.github/CONTRIBUTING.md) — covering gaps or
**Type hints**: required on all new code. mypy is enforced by pre-commit (`.pre-commit-config.yaml`).
**Readable argument lists**: do not put multi-branch conditional expressions inside function or constructor arguments. If an argument needs more than a simple `a if condition else b`, assign it to a named local variable before the call.
**Doctest determinism** — output must be reproducible across platforms:
- Use `# doctest: +ELLIPSIS` for floats that vary by platform.

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-06-25
date_modified: 2026-07-05
---
# Changelog
@ -13,6 +13,10 @@ date_modified: 2026-06-25
Users on Python 3.9 should upgrade their environment before updating supervision.
### Breaking Changes
- `sv.JSONSink` now emits native JSON types for numeric and boolean data fields instead of stringified values. Fields previously serialized as `"True"`/`"False"`, `"1"`/`"0.85"`, or `"400.0"` are now `true`/`false`, `1`/`0.85`, `400.0`. Downstream consumers that compare field values as strings (e.g. `row["score"] == "1"`) or use strict string-typed schema validators must be updated. `sv.CSVSink` remains textual, but its custom-data slicing now matches `sv.JSONSink`: NumPy arrays, lists, and tuples are sliced per row only when their length matches the detection count; mismatched-length values are broadcast unchanged ([#2400](https://github.com/roboflow/supervision/pull/2400)).
- `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 [#2393](https://github.com/roboflow/supervision/pull/2393): `sv.CropAnnotator.annotate` no longer raises `cv2.error` when detections extend outside the scene; out-of-bounds boxes are clipped to scene bounds and zero-area results are skipped silently.
- Fixed [#2393](https://github.com/roboflow/supervision/pull/2393): `sv.HeatMapAnnotator.annotate` no longer blanks the hottest region when the per-pixel hit count exceeds 255; the heat mask is now derived from the float32 accumulator directly, avoiding uint8 wrap-around.

View File

@ -1234,12 +1234,12 @@ class CompactMask:
result[y1 : y1 + crop_h, x1 : x1 + crop_w] = crop
return result
# Slice: use direct Python list slice and numpy view — O(k), no arange.
if isinstance(index, slice):
selected_rles = [rle.copy() for rle in self._rles[index]]
return CompactMask(
self._rles[index],
self._crop_shapes[index],
self._offsets[index],
selected_rles,
self._crop_shapes[index].copy(),
self._offsets[index].copy(),
self._image_shape,
)
@ -1253,9 +1253,9 @@ class CompactMask:
else:
idx_arr = np.asarray(list(index), dtype=np.intp)
new_rles = [self._rles[int(mask_idx)] for mask_idx in idx_arr]
new_crop_shapes: npt.NDArray[np.int32] = self._crop_shapes[idx_arr]
new_offsets: npt.NDArray[np.int32] = self._offsets[idx_arr]
new_rles = [self._rles[int(mask_idx)].copy() for mask_idx in idx_arr]
new_crop_shapes: npt.NDArray[np.int32] = self._crop_shapes[idx_arr].copy()
new_offsets: npt.NDArray[np.int32] = self._offsets[idx_arr].copy()
return CompactMask(new_rles, new_crop_shapes, new_offsets, self._image_shape)
def __array__(

View File

@ -73,7 +73,11 @@ from supervision.detection.vlm import (
)
from supervision.geometry.core import Position
from supervision.utils.internal import get_instance_variables, warn_deprecated
from supervision.validators import _validate_detections_fields, _validate_resolution
from supervision.validators import (
_validate_data,
_validate_detections_fields,
_validate_resolution,
)
@dataclass
@ -335,9 +339,11 @@ class Detections:
)
if hasattr(ultralytics_results, "boxes") and ultralytics_results.boxes is None:
masks = cast(
npt.NDArray[np.bool_], extract_ultralytics_masks(ultralytics_results)
)
masks = extract_ultralytics_masks(ultralytics_results)
if masks is None:
empty = cls.empty()
empty.data = {CLASS_NAME_DATA_FIELD: np.empty(0, dtype=str)}
return empty
return cls(
xyxy=mask_to_xyxy(masks),
mask=masks,
@ -2536,8 +2542,9 @@ class Detections:
index: Row index, indices, slice, or boolean mask selecting detections.
Returns:
A `Detections` instance containing the selected rows. Empty detections
are returned unchanged.
A new `Detections` instance containing the selected rows. Always returns
a fresh copy arrays and metadata are never shared with the original,
even when the selection is empty or the input has zero detections.
Example:
>>> import numpy as np
@ -2546,25 +2553,64 @@ class Detections:
>>> detections.select([1]).xyxy.tolist()
[[1, 1, 2, 2]]
"""
mask: npt.NDArray[np.bool_] | CompactMask | None
if len(self) == 0:
return self
if isinstance(self.mask, CompactMask):
mask = self.mask[:0]
elif self.mask is not None:
mask = self.mask[:0].copy()
else:
mask = None
data = {
key: value.copy() if isinstance(value, np.ndarray) else list(value)
for key, value in self.data.items()
}
return Detections(
xyxy=self.xyxy.copy(),
mask=mask,
confidence=(
self.confidence.copy() if self.confidence is not None else None
),
class_id=self.class_id.copy() if self.class_id is not None else None,
tracker_id=(
self.tracker_id.copy() if self.tracker_id is not None else None
),
data=data,
metadata=dict(self.metadata),
)
if isinstance(index, (int, np.integer)):
index = [int(index)]
array_index = cast(
slice | list[int] | npt.NDArray[np.integer | np.bool_], index
)
data = {
key: value.copy() if isinstance(value, np.ndarray) else list(value)
for key, value in get_data_item(self.data, array_index).items()
}
if isinstance(self.mask, CompactMask):
mask = self.mask[cast(Any, array_index)]
elif self.mask is not None:
mask = self.mask[cast(Any, array_index)].copy()
else:
mask = None
return Detections(
xyxy=self.xyxy[array_index],
mask=self.mask[cast(Any, array_index)] if self.mask is not None else None,
xyxy=self.xyxy[array_index].copy(),
mask=mask,
confidence=(
self.confidence[array_index] if self.confidence is not None else None
self.confidence[array_index].copy()
if self.confidence is not None
else None
),
class_id=(
self.class_id[array_index].copy() if self.class_id is not None else None
),
class_id=self.class_id[array_index] if self.class_id is not None else None,
tracker_id=(
self.tracker_id[array_index] if self.tracker_id is not None else None
self.tracker_id[array_index].copy()
if self.tracker_id is not None
else None
),
data=get_data_item(self.data, array_index),
metadata=self.metadata,
data=data,
metadata=dict(self.metadata),
)
def __getitem__(
@ -2636,6 +2682,11 @@ class Detections:
in detections.class_id
]
```
Raises:
TypeError: If `value` is not a `np.ndarray` or `list`.
ValueError: If `value` has a length or shape incompatible with
the detection count.
"""
if not isinstance(value, (np.ndarray, list)):
raise TypeError("Value must be a np.ndarray or a list")
@ -2643,6 +2694,7 @@ class Detections:
if isinstance(value, list):
value = np.array(value)
_validate_data({key: value}, len(self))
self.data[key] = value
@property
@ -2816,7 +2868,7 @@ class Detections:
class_id=self.class_id,
tracker_id=self.tracker_id,
data=self.data,
metadata=self.metadata,
metadata=dict(self.metadata),
)
return new
@ -3116,10 +3168,31 @@ def _merge_detection_group(detections: list[Detections]) -> Detections:
)
data = winner.data
# Mask union via logical OR
# Mask union via logical OR. Preserve CompactMask outputs without re-cropping
# to the merged box, because source masks may legitimately extend outside it.
masks = [d.mask for d in detections if d.mask is not None]
if masks:
mask = np.logical_or.reduce(np.concatenate(masks, axis=0))[np.newaxis]
if all(isinstance(m, CompactMask) for m in masks):
compact_masks = cast(list[CompactMask], masks)
image_shape = compact_masks[0].image_shape
if any(m.image_shape != image_shape for m in compact_masks):
raise ValueError(
"Cannot merge CompactMask objects with different image shapes."
)
union_mask = np.zeros(image_shape, dtype=bool)
for compact_mask in compact_masks:
union_mask |= compact_mask.to_dense()[0]
union_xyxy = mask_to_xyxy(union_mask[np.newaxis]).astype(np.float32)
mask = CompactMask.from_dense(
masks=union_mask[np.newaxis],
xyxy=union_xyxy,
image_shape=image_shape,
)
else:
dense_masks = [
m.to_dense() if isinstance(m, CompactMask) else m for m in masks
]
mask = np.logical_or.reduce(np.concatenate(dense_masks, axis=0))[np.newaxis]
else:
mask = None

View File

@ -117,6 +117,11 @@ class LineZone:
self.crossing_state_history: dict[tuple[int, int | None], 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._in_count_per_class: Counter[int | None] = Counter()
self._out_count_per_class: Counter[int | None] = Counter()
self.triggering_anchors = triggering_anchors
@ -159,6 +164,7 @@ class LineZone:
crossed_out = np.full(len(detections), False)
if len(detections) == 0:
self._evict_stale_crossing_history(set())
return crossed_in, crossed_out
if detections.tracker_id is None:
@ -170,17 +176,21 @@ class LineZone:
)
return crossed_in, crossed_out
self._update_class_id_to_name(detections)
in_limits, has_any_left_trigger, has_any_right_trigger = (
self._compute_anchor_sides(detections)
)
class_ids: list[int | None] = (
list(detections.class_id)
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)
}
self._evict_stale_crossing_history(current_keys)
self._update_class_id_to_name(detections)
in_limits, has_any_left_trigger, has_any_right_trigger = (
self._compute_anchor_sides(detections)
)
for i, (class_id, tracker_id) in enumerate(
zip(class_ids, detections.tracker_id)
@ -192,7 +202,8 @@ class LineZone:
continue
tracker_state: bool = has_any_left_trigger[i]
crossing_history = self.crossing_state_history[(tracker_id, class_id)]
key = (int(tracker_id), int(class_id) if class_id is not None else None)
crossing_history = self.crossing_state_history[key]
crossing_history.append(tracker_state)
if len(crossing_history) < self.crossing_history_length:
@ -211,6 +222,20 @@ class LineZone:
return crossed_in, crossed_out
def _evict_stale_crossing_history(
self, current_keys: set[tuple[int, int | None]]
) -> None:
for key in list(self.crossing_state_history):
if key in current_keys:
self._tracker_frames_absent.pop(key, None)
else:
absent = self._tracker_frames_absent.get(key, 0) + 1
if absent >= self.crossing_history_length:
del self.crossing_state_history[key]
self._tracker_frames_absent.pop(key, None)
else:
self._tracker_frames_absent[key] = absent
@staticmethod
def _calculate_region_of_interest_limits(vector: Vector) -> tuple[Vector, Vector]:
magnitude = vector.magnitude

View File

@ -39,9 +39,10 @@ class CSVSink:
CSVSink allows passing custom data alongside detection fields, providing
flexibility for logging various types of information.
When a list or tuple value in custom_data (or detections.data) has the
same length as the detection count, each element is written to the
corresponding detection row; any other value is broadcast to all rows.
When a NumPy array, list, or tuple value in custom_data (or
detections.data) has the same length as the detection count, each
element is written to the corresponding detection row; any other value
is broadcast to all rows.
Args:
file_name: The name of the CSV file where the detections will be stored.
@ -122,7 +123,7 @@ class CSVSink:
Dispatch rules:
- np.ndarray with ndim == 0: return as-is for broadcasting
- np.ndarray with ndim >= 1: return value[i]
- np.ndarray with len equal to n: return value[i]
- list or tuple with len equal to n: return value[i]
- any other type: return as-is for broadcasting
@ -136,7 +137,7 @@ class CSVSink:
otherwise value unchanged.
"""
if isinstance(value, np.ndarray):
return value if value.ndim == 0 else value[i]
return value[i] if value.ndim > 0 and len(value) == n else value
if isinstance(value, (list, tuple)) and len(value) == n:
return value[i]
return value
@ -150,9 +151,9 @@ class CSVSink:
Builds one dictionary per detection containing bounding box coordinates,
detection attributes, and any values from ``detections.data`` or
``custom_data``. List and tuple values in ``custom_data`` with length
equal to ``len(detections.xyxy)`` are sliced one element per row; all
other values are broadcast to every row.
``custom_data``. NumPy array, list, and tuple values in
``custom_data`` with length equal to ``len(detections.xyxy)`` are
sliced one element per row; all other values are broadcast to every row.
Args:
detections: Detection data to serialize into row dictionaries.

View File

@ -21,9 +21,10 @@ class JSONSink:
JSONSink allows passing custom data alongside detection fields, providing
flexibility for logging various types of information.
When a list or tuple value in custom_data (or detections.data) has the
same length as the detection count, each element is written to the
corresponding detection row; any other value is broadcast to all rows.
When a NumPy array, list, or tuple value in custom_data (or
detections.data) has the same length as the detection count, each
element is written to the corresponding detection row; any other value
is broadcast to all rows.
NumPy scalars (e.g. ``np.int64``, ``np.float32``) are serialized as
JSON numbers; NumPy arrays are serialized as JSON arrays.
@ -125,7 +126,7 @@ class JSONSink:
Dispatch rules:
- np.ndarray with ndim == 0: return as-is for broadcasting
- np.ndarray with ndim >= 1: return value[i]
- np.ndarray with len equal to n: return value[i]
- list or tuple with len equal to n: return value[i]
- any other type: return as-is for broadcasting
@ -139,7 +140,7 @@ class JSONSink:
otherwise value unchanged.
"""
if isinstance(value, np.ndarray):
return value if value.ndim == 0 else value[i]
return value[i] if value.ndim > 0 and len(value) == n else value
if isinstance(value, (list, tuple)) and len(value) == n:
return value[i]
return value
@ -153,9 +154,9 @@ class JSONSink:
Builds one dictionary per detection containing bounding box coordinates,
detection attributes, and any values from ``detections.data`` or
``custom_data``. List and tuple values in ``custom_data`` with length
equal to ``len(detections.xyxy)`` are sliced one element per row; all
other values are broadcast to every row.
``custom_data``. NumPy array, list, and tuple values in
``custom_data`` with length equal to ``len(detections.xyxy)`` are
sliced one element per row; all other values are broadcast to every row.
Args:
detections: Detection data to serialize into row dictionaries.
@ -187,12 +188,11 @@ class JSONSink:
if hasattr(detections, "data"):
for key, value in detections.data.items():
row[key] = str(JSONSink._slice_value(value, i, n))
row[key] = JSONSink._slice_value(value, i, n)
if custom_data:
for key, value in custom_data.items():
v = JSONSink._slice_value(value, i, n)
row[key] = str(v) if isinstance(value, np.ndarray) else v
row[key] = JSONSink._slice_value(value, i, n)
parsed_rows.append(row)
return parsed_rows

View File

@ -88,7 +88,7 @@ def process_transformers_v5_segmentation_result(
Args:
segmentation_result: Either a dictionary containing segmentation results
(`segments_info` and `segmentation`) or a tensor object
representing a panoptic segmentation map.
representing a semantic segmentation map.
id2label: A dictionary mapping class IDs to labels,
typically part of the `transformers` model configuration. If provided, the
resulting dictionary will include class names.
@ -126,12 +126,15 @@ def process_transformers_v5_semantic_or_instance_segmentation_result(
scores, class IDs, and data.
"""
segments_info = segmentation_result["segments_info"]
scores = np.array([segment["score"] for segment in segments_info])
class_ids = np.array([segment["label_id"] for segment in segments_info])
segmentation_array = segmentation_result["segmentation"].cpu().detach().numpy()
masks = np.array(
[segmentation_array == segment["id"] for segment in segments_info]
).astype(bool)
scores = np.array([segment["score"] for segment in segments_info], dtype=float)
class_ids = np.array([segment["label_id"] for segment in segments_info], dtype=int)
if len(segments_info) == 0:
masks = np.empty((0, *segmentation_array.shape), dtype=bool)
else:
masks = np.array(
[segmentation_array == segment["id"] for segment in segments_info]
).astype(bool)
data = append_class_names_to_data(class_ids, id2label, {})
return dict(
@ -181,11 +184,10 @@ def process_transformers_v5_panoptic_segmentation_result(
segmentation_array: npt.NDArray[Any], id2label: dict[int, str] | None
) -> dict[str, Any]:
"""
Process the result of the Transformers function
`post_process_panoptic_segmentation` (v5).
Process a v5 Transformers semantic segmentation tensor.
Args:
segmentation_array: Segmentation array.
segmentation_array: Segmentation array where unique values are class IDs.
id2label: A dictionary mapping class IDs to labels,
typically part of the `transformers` model configuration. If provided, the
resulting dictionary will include class names.
@ -194,10 +196,13 @@ def process_transformers_v5_panoptic_segmentation_result(
Processed segmentation result including bounding boxes, masks,
class IDs, and data.
"""
class_ids = np.unique(segmentation_array)
masks = np.stack(
[segmentation_array == class_id for class_id in class_ids], axis=0
).astype(bool)
class_ids = np.unique(segmentation_array).astype(int)
if len(class_ids) == 0:
masks = np.empty((0, *segmentation_array.shape), dtype=bool)
else:
masks = np.stack(
[segmentation_array == class_id for class_id in class_ids], axis=0
).astype(bool)
data = append_class_names_to_data(class_ids, id2label, {})
return dict(xyxy=mask_to_xyxy(masks), mask=masks, class_id=class_ids, data=data)

View File

@ -313,7 +313,11 @@ def process_roboflow_result(
where each array is aligned with the others. ``masks`` is ``None``
when no predictions include mask data, or when only a subset do
(mixed-modality batch) in that case all masks are dropped to preserve
alignment with ``xyxy``. When ``compact_masks=True`` and masks are
alignment with ``xyxy``. Note: a single malformed polygon prediction
(fewer than 3 points) in an otherwise fully-segmented batch causes all
masks to be dropped from the result; the detection itself is kept as a
box-only entry with a ``logger.warning``. When ``compact_masks=True``
and masks are
present, ``masks`` is a :class:`CompactMask`; otherwise it is a dense
boolean array. ``tracker_ids`` is ``None`` when no predictions carry a
tracker ID, or when only a subset do (mixed batch) in that case all
@ -434,6 +438,17 @@ def process_roboflow_result(
else:
masks.append(mask)
tracker_ids.append(prediction.get("tracker_id"))
else:
logger.warning(
"Invalid polygon prediction with fewer than 3 points; falling back "
"to box-only detection."
)
xyxy.append([x_min, y_min, x_max, y_max])
class_id.append(prediction["class_id"])
class_name.append(prediction["class"])
confidence.append(prediction["confidence"])
masks.append(None)
tracker_ids.append(prediction.get("tracker_id"))
xyxy_arr: npt.NDArray[np.floating] = (
np.array(xyxy, dtype=np.float64) if len(xyxy) > 0 else np.empty((0, 4))

View File

@ -10,7 +10,8 @@ import numpy as np
import numpy.typing as npt
from supervision.detection.compact_mask import CompactMask
from supervision.detection.utils.masks import resize_masks
from supervision.detection.utils.converters import mask_to_xyxy
from supervision.utils.internal import warn_deprecated
class OverlapFilter(Enum):
@ -1045,8 +1046,9 @@ def mask_non_max_merge(
predictions: npt.NDArray[np.floating],
masks: npt.NDArray[Any] | CompactMask,
iou_threshold: float = 0.5,
mask_dimension: int = 640,
*args: Any,
overlap_metric: OverlapMetric = OverlapMetric.IOU,
mask_dimension: int = 640,
) -> list[list[int]]:
"""
Perform Non-Maximum Merging (NMM) on segmentation predictions.
@ -1060,11 +1062,12 @@ def mask_non_max_merge(
Shape: `(N, H, W)`, where N is the number of predictions, and H, W are the
dimensions of each mask.
iou_threshold: The intersection-over-union threshold
to use for non-maximum suppression.
mask_dimension: The dimension to which the masks should be
resized before computing IOU values. Defaults to 640.
to use for non-maximum merging.
overlap_metric: Metric used to compute the degree of overlap
between pairs of masks (e.g., IoU, IoS).
mask_dimension: Deprecated in `0.30.0`, removed in `0.33.0`. No longer
used; the parameter is silently ignored. Passing `mask_dimension`
positionally emits a deprecation warning.
Returns:
A list of groups of prediction indices. Each inner list contains
@ -1075,61 +1078,104 @@ def mask_non_max_merge(
Raises:
AssertionError: If `iou_threshold` is not within the closed
range from `0` to `1`.
TypeError: If more than five positional arguments are passed.
Examples:
```pycon
>>> import numpy as np
>>> import supervision as sv
>>> predictions = np.array([
... [0, 0, 4, 4, 0.9, 0],
... [0, 0, 4, 4, 0.8, 0],
... ])
>>> masks = np.zeros((2, 4, 4), dtype=bool)
>>> masks[:, :2, :2] = True
>>> sv.mask_non_max_merge(predictions, masks, iou_threshold=0.5)
[[0, 1]]
```
"""
if isinstance(masks, CompactMask):
# _group_overlapping_masks needs dense arrays for logical_or union merging.
# Note: np.asarray(masks) first materialises a full-resolution (N, H, W)
# dense array before downscaling with resize_masks. This reduces the size
# of the array used for overlap computation but does not avoid the initial
# full-frame materialisation, which may still be memory-intensive for very
# large images or object counts.
masks = resize_masks(np.asarray(masks), mask_dimension)
else:
masks = resize_masks(masks, mask_dimension)
masks_resized = masks
if predictions.shape[1] == 5:
return _group_overlapping_masks(
predictions, masks_resized, iou_threshold, overlap_metric
assert 0 <= iou_threshold <= 1, (
"Value of `iou_threshold` must be in the closed range from 0 to 1, "
f"{iou_threshold} given."
)
if len(args) > 2:
raise TypeError(
"mask_non_max_merge accepts at most five positional arguments. "
"Pass overlap_metric and mask_dimension by keyword."
)
if args:
warn_deprecated(
"Passing `overlap_metric` or `mask_dimension` positionally to "
"`mask_non_max_merge` is deprecated in `0.30.0` and will be removed "
"in `0.33.0`. Pass them by keyword instead."
)
first = args[0]
if isinstance(first, OverlapMetric):
overlap_metric = first
else:
mask_dimension = cast(int, first)
if len(args) == 2:
second = args[1]
if isinstance(first, OverlapMetric):
mask_dimension = cast(int, second)
else:
overlap_metric = cast(OverlapMetric, second)
category_ids = predictions[:, 5]
merge_groups = []
for category_id in np.unique(category_ids):
curr_indices = np.where(category_ids == category_id)[0]
merge_class_groups = _group_overlapping_masks(
predictions[curr_indices],
masks_resized[curr_indices],
del mask_dimension
def group_within(global_indices: npt.NDArray[np.int_]) -> list[list[int]]:
if isinstance(masks, CompactMask):
return _group_overlapping_masks_pairwise(
predictions[global_indices],
masks[global_indices],
iou_threshold,
overlap_metric,
)
return _group_overlapping_masks(
predictions[global_indices],
masks[global_indices],
iou_threshold,
overlap_metric,
)
for merge_class_group in merge_class_groups:
merge_groups.append(curr_indices[merge_class_group].tolist())
for merge_group in merge_groups:
if len(merge_group) == 0:
raise ValueError(
f"Empty group detected when non-max-merging detections: {merge_groups}"
)
return merge_groups
return _non_max_merge_per_category(predictions, group_within)
def _greedy_nmm_via_iou_callback(
def _update_mask_candidate(
masks: npt.NDArray[Any] | CompactMask,
candidate: npt.NDArray[Any] | CompactMask,
above_idx: npt.NDArray[np.int_],
) -> npt.NDArray[Any] | CompactMask:
if isinstance(masks, CompactMask):
compact_candidate = cast(CompactMask, candidate)
union_mask = np.logical_or.reduce(
np.concatenate([masks[above_idx].to_dense(), compact_candidate.to_dense()]),
axis=0,
keepdims=True,
)
return CompactMask.from_dense(
masks=union_mask,
xyxy=mask_to_xyxy(union_mask),
image_shape=masks.image_shape,
)
dense_candidate = cast(npt.NDArray[Any], candidate)
dense_union: npt.NDArray[Any] = np.logical_or.reduce(
np.concatenate([masks[above_idx], dense_candidate]),
axis=0,
keepdims=True,
)
return dense_union
def _greedy_nmm_via_mask_candidate(
predictions: npt.NDArray[np.floating],
iou_against_candidate: Callable[
[npt.NDArray[np.int_], int], npt.NDArray[np.floating]
],
iou_threshold: float,
masks: npt.NDArray[Any] | CompactMask,
iou_threshold: float = 0.5,
overlap_metric: OverlapMetric = OverlapMetric.IOU,
) -> list[list[int]]:
"""Greedy non-maximum merging loop, independent of how overlap is computed.
``iou_against_candidate(order_indices, candidate_idx)`` must return the IoU
vector between every prediction in ``order_indices`` and the candidate at
``candidate_idx``. Predictions whose IoU meets ``iou_threshold`` are
grouped with the candidate.
"""
"""Group masks by exact overlap while updating the merged candidate union."""
merge_groups: list[list[int]] = []
scores = predictions[:, 4]
order = scores.argsort()
@ -1139,14 +1185,32 @@ def _greedy_nmm_via_iou_callback(
if len(order) == 0:
merge_groups.append([idx])
break
ious = iou_against_candidate(order, idx)
above_threshold = ious >= iou_threshold
merge_group = [idx, *np.flip(order[above_threshold]).tolist()]
candidate = masks[[idx]]
merge_group = [idx]
while len(order) > 0:
ious = mask_iou_batch(masks[order], candidate, overlap_metric).flatten()
above_threshold = ious >= iou_threshold
if not above_threshold.any():
break
above_idx = order[above_threshold]
candidate = _update_mask_candidate(masks, candidate, above_idx)
merge_group.extend(np.flip(above_idx).tolist())
order = order[~above_threshold]
merge_groups.append(merge_group)
order = order[~above_threshold]
return merge_groups
def _group_overlapping_masks_pairwise(
predictions: npt.NDArray[np.floating],
masks: npt.NDArray[Any] | CompactMask,
iou_threshold: float = 0.5,
overlap_metric: OverlapMetric = OverlapMetric.IOU,
) -> list[list[int]]:
return _greedy_nmm_via_mask_candidate(
predictions, masks, iou_threshold, overlap_metric
)
def _non_max_merge_per_category(
predictions: npt.NDArray[np.floating],
group_within: Callable[[npt.NDArray[np.int_]], list[list[int]]],
@ -1202,19 +1266,25 @@ def _group_overlapping_boxes(
Groups of prediction indices to be merged.
Each group may have 1 or more elements.
"""
def iou_against_candidate(
order: npt.NDArray[np.int_], idx: int
) -> npt.NDArray[np.floating]:
return box_iou_batch(
merge_groups: list[list[int]] = []
scores = predictions[:, 4]
order = scores.argsort()
while len(order) > 0:
idx = int(order[-1])
order = order[:-1]
if len(order) == 0:
merge_groups.append([idx])
break
ious = box_iou_batch(
predictions[order][:, :4],
predictions[idx : idx + 1, :4],
overlap_metric,
).flatten()
return _greedy_nmm_via_iou_callback(
predictions, iou_against_candidate, iou_threshold
)
above_threshold = ious >= iou_threshold
merge_group = [idx, *np.flip(order[above_threshold]).tolist()]
merge_groups.append(merge_group)
order = order[~above_threshold]
return merge_groups
def box_non_max_merge(
@ -1355,19 +1425,25 @@ def _group_overlapping_oriented_boxes(
Greedy non-maximum merging on oriented boxes. Mirrors
:func:`_group_overlapping_boxes` but uses :func:`oriented_box_iou_batch`.
"""
def iou_against_candidate(
order: npt.NDArray[np.int_], idx: int
) -> npt.NDArray[np.floating]:
return oriented_box_iou_batch(
merge_groups: list[list[int]] = []
scores = predictions[:, 4]
order = scores.argsort()
while len(order) > 0:
idx = int(order[-1])
order = order[:-1]
if len(order) == 0:
merge_groups.append([idx])
break
ious = oriented_box_iou_batch(
oriented_boxes[order],
oriented_boxes[idx][None, ...],
overlap_metric,
).flatten()
return _greedy_nmm_via_iou_callback(
predictions, iou_against_candidate, iou_threshold
)
above_threshold = ious >= iou_threshold
merge_group = [idx, *np.flip(order[above_threshold]).tolist()]
merge_groups.append(merge_group)
order = order[~above_threshold]
return merge_groups
def oriented_box_non_max_merge(

View File

@ -300,7 +300,9 @@ def resize_masks(
"""
max_height: int = masks.shape[1]
max_width: int = masks.shape[2]
scale = min(max_dimension / max_height, max_dimension / max_width)
scale = min(1.0, max_dimension / max_height, max_dimension / max_width)
if scale == 1.0:
return masks
new_height = int(scale * max_height)
new_width = int(scale * max_width)
@ -516,7 +518,8 @@ def _masks_to_roi(
image_shape: Image dimensions as ``(height, width)``.
xyxy: Optional detection boxes of shape ``(N, 4)`` in
``[x1, y1, x2, y2]`` format. When provided, the dense path
uses box union (O(N)) instead of a full pixel scan (O(N·H·W)).
first checks whether all true pixels fall within the box union;
if so the box bounds are returned without a full pixel scan.
Returns:
Exclusive ``(x1, y1, x2, y2)`` bounds, or ``None`` when no true
@ -527,16 +530,31 @@ def _masks_to_roi(
mask_array = np.asarray(masks, dtype=bool)
if mask_array.size == 0 or not mask_array.any():
return None
# Fast path: union of detection boxes (O(N)) avoids full N·H·W pixel scan.
# supervision xyxy uses inclusive max coords; floor(x2)+1 converts to exclusive.
# Fast path: union of detection boxes (O(N)) avoids a full N·H·W pixel scan
# when boxes are mask-derived. Guard it with an `any()` check over the ROI so
# loose boxes cannot clip true pixels that lie outside the box union.
if xyxy is not None and len(xyxy) > 0:
image_h, image_w = image_shape
return (
box_roi = (
max(0, int(np.floor(xyxy[:, 0].min()))),
max(0, int(np.floor(xyxy[:, 1].min()))),
min(image_w, int(np.floor(xyxy[:, 2].max())) + 1),
min(image_h, int(np.floor(xyxy[:, 3].max())) + 1),
)
x1, y1, x2, y2 = box_roi
if x1 < x2 and y1 < y2:
union = mask_array if mask_array.ndim == 2 else np.any(mask_array, axis=0)
# Return box bounds only when all true pixels fall within the box.
# Slice-based checks avoid allocating a full-frame copy.
if (
union[y1:y2, x1:x2].any()
and not union[:y1].any()
and not union[y2:].any()
and not union[y1:y2, :x1].any()
and not union[y1:y2, x2:].any()
):
return box_roi
return _mask_to_roi(union)
if mask_array.ndim == 2:
union = mask_array
else:

View File

@ -335,6 +335,82 @@ def test_select_returns_detection_subset() -> None:
)
def test_select_empty_returns_fresh_metadata_dict() -> None:
"""Selecting empty detections returns a fresh metadata dictionary."""
detections = Detections.empty()
detections.metadata["source"] = "camera"
result = detections.select([])
result.metadata["source"] = "other"
assert detections.metadata["source"] == "camera"
def test_select_non_empty_slice_returns_fresh_arrays() -> None:
"""Selecting non-empty detections does not share array storage."""
detections = Detections(
xyxy=np.array([[0, 0, 1, 1], [2, 2, 3, 3]], dtype=np.float32),
mask=np.array(
[
[[True, False], [False, False]],
[[False, True], [False, False]],
]
),
confidence=np.array([0.1, 0.2], dtype=np.float32),
class_id=np.array([1, 2]),
tracker_id=np.array([10, 20]),
data={"features": np.array([[1, 2], [3, 4]])},
)
result = detections.select(slice(0, 1))
assert isinstance(result.mask, np.ndarray)
assert result.confidence is not None
assert result.class_id is not None
assert result.tracker_id is not None
assert isinstance(result.data["features"], np.ndarray)
result.xyxy[0, 0] = 99
result.mask[0, 0, 0] = False
result.confidence[0] = 0.9
result.class_id[0] = 9
result.tracker_id[0] = 90
result.data["features"][0, 0] = 99
assert detections.xyxy[0, 0] == 0
assert detections.mask[0, 0, 0]
assert detections.confidence[0] == pytest.approx(0.1)
assert detections.class_id[0] == 1
assert detections.tracker_id[0] == 10
assert detections.data["features"][0, 0] == 1
def test_select_compact_mask_slice_returns_fresh_arrays() -> None:
"""Selecting CompactMask detections by slice does not share public arrays."""
masks = np.zeros((2, 4, 4), dtype=bool)
masks[:, :2, :2] = True
xyxy = np.array([[0, 0, 1, 1], [1, 1, 2, 2]], dtype=np.float32)
compact_mask = CompactMask.from_dense(masks, xyxy, image_shape=(4, 4))
detections = Detections(xyxy=xyxy.copy(), mask=compact_mask)
result = detections.select(slice(0, 1))
assert isinstance(result.mask, CompactMask)
result.mask.offsets[0, 0] = 3
assert isinstance(detections.mask, CompactMask)
assert detections.mask.offsets[0, 0] == 0
def test_setitem_rejects_data_length_mismatch() -> None:
"""Data assignment rejects values not aligned with detections length."""
detections = Detections(
xyxy=np.array([[0, 0, 1, 1], [2, 2, 3, 3]], dtype=np.float32)
)
with pytest.raises(ValueError, match=r"must be \(2,\)"):
detections["name"] = np.array(["cat"])
def test_get_data_returns_detection_data_value() -> None:
"""Get data returns the stored data value or None."""
result = TEST_DET_1.get_data("some_key")
@ -2318,6 +2394,35 @@ class TestDetectionsWithNMM:
assert len(result) == 0
def test_compact_mask_nmm_preserves_full_frame_union(self) -> None:
"""CompactMask NMM keeps full-frame mask pixels after merging."""
masks = np.zeros((2, 10, 10), dtype=bool)
masks[0, 1, 1] = True
masks[0, 8, 8] = True
masks[1, 1, 1] = True
masks[1, 7, 7] = True
compact_mask = CompactMask.from_dense(
masks=masks,
xyxy=np.array([[0, 0, 9, 9], [0, 0, 9, 9]], dtype=np.float32),
image_shape=(10, 10),
)
detections = Detections(
xyxy=np.array([[0, 0, 1, 1], [0, 0, 1, 1]], dtype=np.float32),
mask=compact_mask,
confidence=np.array([0.9, 0.8], dtype=np.float32),
class_id=np.array([0, 0]),
)
result = detections.with_nmm(threshold=0.1)
assert len(result) == 1
assert isinstance(result.mask, CompactMask)
assert result.mask.bbox_xyxy.tolist() == [[1, 1, 8, 8]]
result_mask = result.mask.to_dense()[0]
assert result_mask[1, 1]
assert result_mask[7, 7]
assert result_mask[8, 8]
class TestDetectionsArea:
"""Selection order for the `area` property: mask → OBB → AABB."""

View File

@ -537,3 +537,28 @@ def assert_csv_equal(file_name, expected_rows) -> None:
)
def test_csv_sink_slice_value(value: Any, i: int, n: int, expected: Any) -> None:
assert CSVSink._slice_value(value, i, n) == expected
@pytest.mark.parametrize(
("custom_data", "expected_value"),
[
pytest.param(
{"embedding": np.array([1, 2, 3])},
np.array([1, 2, 3]),
id="custom_data_ndarray_length_mismatch",
)
],
)
def test_csv_sink_broadcasts_ndarray_when_length_mismatches_detection_count(
custom_data: dict[str, Any] | None, expected_value: np.ndarray
) -> None:
"""Mismatched ndarray data is broadcast instead of indexed per row."""
detections = sv.Detections(
xyxy=np.array([[0, 0, 10, 10], [20, 20, 30, 30]]),
)
rows = CSVSink.parse_detection_data(detections, custom_data=custom_data)
assert len(rows) == 2
np.testing.assert_array_equal(rows[0]["embedding"], expected_value)
np.testing.assert_array_equal(rows[1]["embedding"], expected_value)

View File

@ -82,6 +82,20 @@ def test_from_ultralytics_segmentation_only_branch_uses_masks_and_arange(
np.testing.assert_array_equal(det.class_id, np.arange(len(results)))
def test_from_ultralytics_segmentation_only_without_masks_returns_empty() -> None:
"""Segmentation-only Ultralytics results without masks return empty detections."""
results = _FakeUltralyticsResults(boxes=None, names={}, length=0)
det = Detections.from_ultralytics(results)
assert len(det) == 0
assert det.xyxy.shape == (0, 4)
assert det.mask is None
np.testing.assert_array_equal(
det.data[CLASS_NAME_DATA_FIELD], np.array([], dtype=str)
)
@pytest.mark.parametrize(
("bboxes", "conf", "labels", "expected_len"),
[

View File

@ -168,8 +168,8 @@ from tests.helpers import _create_detections
"confidence": 0.949999988079071,
"tracker_id": "",
"class_name": "unknown",
"is_detected": "True",
"score": "1",
"is_detected": True,
"score": 1,
"frame_number": 46,
},
{
@ -181,8 +181,8 @@ from tests.helpers import _create_detections
"confidence": "",
"tracker_id": "",
"class_name": "artifact",
"is_detected": "False",
"score": "0.85",
"is_detected": False,
"score": 0.85,
"frame_number": 47,
},
],
@ -251,7 +251,7 @@ from tests.helpers import _create_detections
"class_id": 0,
"confidence": 0.8999999761581421,
"tracker_id": "",
"area": "400.0",
"area": 400.0,
},
{
"x_min": 50,
@ -261,7 +261,7 @@ from tests.helpers import _create_detections
"class_id": 1,
"confidence": 0.800000011920929,
"tracker_id": "",
"area": "400.0",
"area": 400.0,
},
{
"x_min": 15,
@ -271,7 +271,7 @@ from tests.helpers import _create_detections
"class_id": 2,
"confidence": 0.699999988079071,
"tracker_id": "",
"area": "400.0",
"area": 400.0,
},
],
), # numpy array in custom_data sliced per detection row
@ -434,6 +434,55 @@ def test_json_sink_serializes_nested_numpy_array_custom_data(tmp_path: Any) -> N
assert data[0]["meta"]["arr"] == [1, 2, 3]
@pytest.mark.parametrize(
("custom_data", "expected_value"),
[
pytest.param(
{"embedding": np.array([1, 2, 3])},
[1, 2, 3],
id="custom_data_ndarray_length_mismatch",
)
],
)
def test_json_sink_broadcasts_ndarray_when_length_mismatches_detection_count(
tmp_path: Any, custom_data: dict[str, Any] | None, expected_value: list[float]
) -> None:
"""Mismatched ndarray data is broadcast and serialized as a JSON array."""
file_name = str(tmp_path / "test_mismatched_array.json")
detections = sv.Detections(
xyxy=np.array([[0, 0, 10, 10], [20, 20, 30, 30]]),
)
with sv.JSONSink(file_name) as sink:
sink.append(detections, custom_data=custom_data)
with open(file_name) as f:
data = json.load(f)
assert data[0]["embedding"] == expected_value
assert data[1]["embedding"] == expected_value
def test_json_sink_serializes_matching_ndarray_rows_as_json_arrays(
tmp_path: Any,
) -> None:
"""Matching 2D ndarray row data serializes as JSON arrays, not strings."""
file_name = str(tmp_path / "test_matching_array_rows.json")
detections = sv.Detections(
xyxy=np.array([[0, 0, 10, 10], [20, 20, 30, 30]]),
data={"embedding": np.array([[1, 2], [3, 4]])},
)
with sv.JSONSink(file_name) as sink:
sink.append(detections, custom_data={"score": np.array([0.5, 0.75])})
with open(file_name) as f:
data = json.load(f)
assert data[0]["embedding"] == [1, 2]
assert data[1]["embedding"] == [3, 4]
assert data[0]["score"] == 0.5
assert data[1]["score"] == 0.75
def test_json_default_raises_for_unserializable_type() -> None:
"""_json_default raises TypeError for non-numpy objects."""
with pytest.raises(TypeError, match="is not JSON serializable"):

View File

@ -3,7 +3,7 @@ from contextlib import ExitStack as DoesNotRaise
import numpy as np
import pytest
from supervision import LineZone, LineZoneAnnotatorMulticlass
from supervision import Detections, LineZone, LineZoneAnnotatorMulticlass
from supervision.geometry.core import Point, Position, Vector
from tests.helpers import _create_detections
@ -899,6 +899,54 @@ def test_line_zone_trigger_does_not_call_np_cross(
assert line_zone.out_count == 1
def test_line_zone_trigger_evicts_stale_crossing_history() -> None:
"""History for tracker IDs absent from the current frame is evicted."""
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]
)
second_detections = _create_detections(
xyxy=[[4, 4, 6, 6]], tracker_id=[1], class_id=[2]
)
line_zone.trigger(first_detections)
# Trigger twice with second_detections so tracker_id=0 accumulates
# crossing_history_length absent frames (default=2) and is evicted.
line_zone.trigger(second_detections)
line_zone.trigger(second_detections)
assert set(line_zone.crossing_state_history) == {(1, 2)}
def test_line_zone_trigger_evicts_stale_crossing_history_on_empty_frames() -> None:
"""Empty frames age out tracker crossing history."""
line_zone = LineZone(start=Point(0, 0), end=Point(10, 0))
detections = _create_detections(xyxy=[[4, 4, 6, 6]], tracker_id=[0], class_id=[1])
line_zone.trigger(detections)
for _ in range(line_zone.crossing_history_length):
line_zone.trigger(Detections.empty())
assert not line_zone.crossing_state_history
def test_line_zone_trigger_evicts_stale_crossing_history_on_class_change() -> None:
"""Class changes age out stale per-class 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]
)
second_detections = _create_detections(
xyxy=[[4, 4, 6, 6]], tracker_id=[0], class_id=[2]
)
line_zone.trigger(first_detections)
for _ in range(line_zone.crossing_history_length):
line_zone.trigger(second_detections)
assert set(line_zone.crossing_state_history) == {(0, 2)}
def test_line_zone_annotator_multiclass_supports_none_class_id() -> None:
line_zone = LineZone(start=Point(0, 0), end=Point(0, 10))
for xyxy in [[4, 4, 6, 6], [-6, 4, -4, 6]]:

View File

@ -243,16 +243,34 @@ class TestProcessTransformersV4PanopticSegmentationResult:
class TestProcessTransformersV5PanopticSegmentationResult:
"""process_transformers_v5_panoptic_segmentation_result uses unique pixel values."""
"""process_transformers_v5_panoptic_segmentation_result handles semantic tensors."""
def test_two_unique_ids_produce_two_masks(self) -> None:
"""Array with two unique values produces two boolean masks."""
seg_array = np.array([[0, 0, 1, 1], [0, 0, 1, 1]], dtype=np.int64)
@pytest.mark.parametrize(
("seg_array", "expected_class_ids"),
[
pytest.param(
np.array([[0, 0, 1, 1], [0, 2, 2, 0]], dtype=np.int64),
np.array([0, 1, 2]),
id="preserves-class-zero",
),
pytest.param(
np.zeros((2, 2), dtype=np.int64),
np.array([0]),
id="single-zero-class",
),
],
)
def test_semantic_tensor_preserves_class_zero(
self, seg_array: np.ndarray, expected_class_ids: np.ndarray
) -> None:
"""Bare tensor semantic maps preserve class id zero."""
expected_count = len(expected_class_ids)
out = process_transformers_v5_panoptic_segmentation_result(seg_array, None)
assert out["mask"].shape[0] == 2
np.testing.assert_array_equal(out["class_id"], [0, 1])
assert out["mask"].shape == (expected_count, *seg_array.shape)
assert out["xyxy"].shape == (expected_count, 4)
np.testing.assert_array_equal(out["class_id"], expected_class_ids)
def test_with_id2label_sets_class_names(self) -> None:
"""id2label maps unique IDs to class name strings in output data."""
@ -266,6 +284,19 @@ class TestProcessTransformersV5PanopticSegmentationResult:
out["data"][CLASS_NAME_DATA_FIELD], ["tree", "sky"]
)
def test_with_id2label_preserves_zero_class_name(self) -> None:
"""id2label maps class id zero when it appears in a tensor map."""
seg_array = np.array([[0, 0], [1, 1]], dtype=np.int64)
out = process_transformers_v5_panoptic_segmentation_result(
seg_array, {0: "class-zero", 1: "class-one"}
)
np.testing.assert_array_equal(out["class_id"], [0, 1])
np.testing.assert_array_equal(
out["data"][CLASS_NAME_DATA_FIELD], ["class-zero", "class-one"]
)
# ---------------------------------------------------------------------------
# process_transformers_v5_semantic_or_instance_segmentation_result
@ -296,16 +327,8 @@ class TestProcessTransformersV5SemanticOrInstanceSegmentationResult:
np.testing.assert_array_equal(out["class_id"], [0, 1])
np.testing.assert_allclose(out["confidence"], [0.9, 0.7])
@pytest.mark.xfail(
raises=ValueError,
reason=(
"empty segments_info produces masks shape (0,) instead of (0,H,W),"
" causing mask_to_xyxy to crash — source bug, not a test setup issue"
),
strict=True,
)
def test_empty_segments_info_returns_zero_detections(self) -> None:
"""Empty segments_info list should yield zero-length arrays (xfail: bug)."""
"""Empty segments_info list yields zero-length detection arrays."""
seg_result = {
"segmentation": _FakeDetachTensor(np.zeros((2, 2), dtype=np.int64)),
"segments_info": [],
@ -316,6 +339,9 @@ class TestProcessTransformersV5SemanticOrInstanceSegmentationResult:
)
assert len(out["class_id"]) == 0
assert out["xyxy"].shape == (0, 4)
assert out["mask"].shape == (0, 2, 2)
assert out["confidence"].shape == (0,)
# ---------------------------------------------------------------------------
@ -343,11 +369,11 @@ class TestProcessTransformersV5SegmentationResult:
assert len(out["class_id"]) == 2
def test_tensor_like_object_routes_to_panoptic_path(self) -> None:
"""Object whose class is named 'Tensor' routes to panoptic sub-processor."""
def test_tensor_like_object_routes_to_semantic_tensor_path(self) -> None:
"""Object whose class is named 'Tensor' routes to semantic tensor path."""
class Tensor:
"""Minimal fake torch.Tensor for the panoptic path."""
"""Minimal fake torch.Tensor for the semantic tensor path."""
def __init__(self, arr: np.ndarray) -> None:
self._arr = arr
@ -369,5 +395,4 @@ class TestProcessTransformersV5SegmentationResult:
out = process_transformers_v5_segmentation_result(tensor_result, None)
# Panoptic path: unique IDs [0, 1] → two masks
assert len(out["class_id"]) == 2
np.testing.assert_array_equal(out["class_id"], [0, 1])

View File

@ -173,15 +173,15 @@ TEST_RLE_NONCONTIGUOUS_MASK[0, 3, 2:4] = True
),
),
(
np.empty((0, 4)),
np.empty(0),
np.empty(0),
np.array([[175.0, 275.0, 225.0, 325.0]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.empty(0, dtype=str)},
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single incorrect instance segmentation result with no points
), # single invalid polygon result with no points falls back to box-only
(
_result_1k(
_pred(
@ -191,15 +191,15 @@ TEST_RLE_NONCONTIGUOUS_MASK[0, 3, 2:4] = True
),
),
(
np.empty((0, 4)),
np.empty(0),
np.empty(0),
np.array([[175.0, 275.0, 225.0, 325.0]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.empty(0, dtype=str)},
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single incorrect instance segmentation result with no enough points
), # single invalid polygon result with too few points falls back to box-only
(
_result_1k(
_pred(
@ -245,15 +245,15 @@ TEST_RLE_NONCONTIGUOUS_MASK[0, 3, 2:4] = True
),
),
(
np.array([[175.0, 275.0, 225.0, 325.0]]),
np.array([0.9]),
np.array([0]),
TEST_MASK,
np.array([[175.0, 275.0, 225.0, 325.0], [450.0, 450.0, 550.0, 550.0]]),
np.array([0.9, 0.8]),
np.array([0, 7]),
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
None,
{CLASS_NAME_DATA_FIELD: np.array(["person", "truck"])},
),
DoesNotRaise(),
), # two instance segmentation results - one correct, one incorrect
), # mixed valid polygon and invalid polygon keeps boxes and drops masks
(
_result(_pred(rle={"size": [4, 4], "counts": "52203"})),
(
@ -522,6 +522,32 @@ def test_process_roboflow_result_uses_rle_mask_when_rle_invalid() -> None:
np.testing.assert_array_equal(compact_result[3].to_dense(), dense_result[3])
def test_process_roboflow_result_invalid_polygon_is_box_only(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Predictions with fewer than three polygon points are kept as box-only."""
roboflow_result = _result(_pred(points=[{"x": 1, "y": 1}, {"x": 2, "y": 2}]))
with caplog.at_level("WARNING"):
xyxy, confidence, class_id, masks, tracker_ids, data = process_roboflow_result(
roboflow_result=roboflow_result
)
np.testing.assert_array_equal(xyxy, np.array([[0.5, 0.5, 2.5, 2.5]]))
np.testing.assert_array_equal(confidence, np.array([0.9]))
np.testing.assert_array_equal(class_id, np.array([0]))
assert masks is None
assert tracker_ids is None
np.testing.assert_array_equal(data[CLASS_NAME_DATA_FIELD], np.array(["person"]))
assert "fewer than 3 points" in caplog.text
compact_result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
np.testing.assert_array_equal(compact_result[0], xyxy)
assert compact_result[3] is None
def test_polygon_prediction_compact_masks_true() -> None:
"""polygon prediction with compact_masks=True returns a CompactMask."""
roboflow_result = _result(

View File

@ -20,6 +20,7 @@ from supervision.detection.utils.iou_and_nms import (
oriented_box_non_max_merge,
oriented_box_non_max_suppression,
)
from supervision.utils.internal import SupervisionWarnings
from tests.helpers import _generate_random_boxes
@ -600,7 +601,7 @@ def test_mask_non_max_suppression(
]
),
0.6,
[[0, 1]],
[[0], [1]],
DoesNotRaise(),
), # two masks partially overlapping with no category, no merge
(
@ -669,6 +670,94 @@ def test_mask_non_max_merge(
assert sorted_result == sorted_expected_result
def test_mask_non_max_merge_ignores_mask_dimension_for_exact_iou() -> None:
"""Mask NMM ignores downscale dimension and uses exact mask overlap."""
predictions = np.array(
[[0, 0, 4, 4, 0.9, 0], [0, 0, 4, 4, 0.8, 0]], dtype=np.float32
)
masks = np.zeros((2, 4, 4), dtype=bool)
masks[0, 0, 0] = True
masks[0, 0, 1] = True
masks[1, 0, 0] = True
masks[1, 1, 0] = True
result = mask_non_max_merge(
predictions=predictions,
masks=masks,
iou_threshold=0.9,
mask_dimension=1,
)
assert sorted([sorted(group) for group in result]) == [[0], [1]]
def test_mask_non_max_merge_warns_for_legacy_positional_trailing_args() -> None:
"""Mask NMM supports legacy positional trailing args with warning."""
predictions = np.array(
[[0, 0, 4, 4, 0.9, 0], [0, 0, 4, 4, 0.8, 0]], dtype=np.float32
)
masks = np.zeros((2, 4, 4), dtype=bool)
masks[0, 0:2, 0:2] = True
masks[1, 0:2, 0:2] = True
with pytest.warns(SupervisionWarnings, match="positionally.*deprecated"):
legacy_result = mask_non_max_merge(
predictions,
masks,
0.5,
640,
OverlapMetric.IOU,
)
with pytest.warns(SupervisionWarnings, match="positionally.*deprecated"):
reordered_result = mask_non_max_merge(
predictions,
masks,
0.5,
OverlapMetric.IOU,
640,
)
assert legacy_result == [[0, 1]]
assert reordered_result == [[0, 1]]
def test_mask_non_max_merge_compact_mask_matches_dense_chained_union() -> None:
"""CompactMask NMM matches dense masks when merge candidates expand."""
from supervision.detection.compact_mask import CompactMask
predictions = np.array(
[
[0, 0, 6, 2, 0.9, 0],
[0, 0, 6, 2, 0.8, 0],
[0, 0, 6, 2, 0.7, 0],
],
dtype=np.float32,
)
masks = np.zeros((3, 2, 6), dtype=bool)
masks[0, :, 0:2] = True
masks[1, :, 1:4] = True
masks[2, :, 3:5] = True
compact_mask = CompactMask.from_dense(
masks=masks,
xyxy=np.array([[0, 0, 5, 1], [0, 0, 5, 1], [0, 0, 5, 1]], dtype=np.float32),
image_shape=(2, 6),
)
dense_result = mask_non_max_merge(
predictions=predictions,
masks=masks,
iou_threshold=0.2,
)
compact_result = mask_non_max_merge(
predictions=predictions,
masks=compact_mask,
iou_threshold=0.2,
)
assert sorted([sorted(group) for group in dense_result]) == [[0, 1, 2]]
assert sorted([sorted(group) for group in compact_result]) == [[0, 1, 2]]
@pytest.mark.parametrize(
("box_true", "box_detection", "overlap_metric", "expected_overlap", "exception"),
[

View File

@ -14,6 +14,7 @@ from supervision.detection.utils.masks import (
contains_multiple_segments,
filter_segments_by_distance,
move_masks,
resize_masks,
)
@ -127,6 +128,17 @@ class TestMaskROIHelpers:
assert x2 >= 21 # floor(20.0) + 1
assert y2 >= 21
def test_masks_to_roi_dense_with_xyxy_falls_back_for_pixels_outside_box(self):
"""Dense xyxy path scans pixels when true pixels fall outside the box union."""
masks = np.zeros((1, 30, 40), dtype=bool)
masks[0, 10:12, 10:12] = True
masks[0, 25:27, 30:32] = True
xyxy = np.array([[10.0, 10.0, 11.0, 11.0]])
result = _masks_to_roi(masks, (30, 40), xyxy)
assert result == (10, 10, 32, 27)
def test_masks_to_roi_dense_with_xyxy_all_false_returns_none(self):
"""All-false masks with xyxy provided should still return None."""
masks = np.zeros((2, 30, 40), dtype=bool)
@ -135,6 +147,17 @@ class TestMaskROIHelpers:
assert result is None
def test_resize_masks_does_not_upscale_small_masks() -> None:
"""resize_masks returns small masks unchanged instead of upscaling."""
masks = np.zeros((2, 4, 5), dtype=bool)
masks[:, 1:3, 2:4] = True
result = resize_masks(masks, max_dimension=640)
assert result is masks
assert result.shape == (2, 4, 5)
@pytest.mark.parametrize(
("masks", "offset", "resolution_wh", "expected_result", "exception"),
[