Files
wehub-resource-sync 9194ef5abd
Docs/Test Workflow / Test docs build (push) Failing after 0s
Check links & references / links-check (push) Failing after 1s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.10) (push) Failing after 0s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.11) (push) Failing after 0s
PR Conflict Labeler / main (push) Failing after 2s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.12) (push) Failing after 2s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.13) (push) Failing after 0s
Pytest/Test Workflow / Build this Package (push) Failing after 5s
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.10) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.11) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.12) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.13) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.10) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.11) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.12) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.13) (push) Has been cancelled
Pytest/Test Workflow / testing-guardian (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:06:10 +08:00

1280 lines
43 KiB
Python

from contextlib import ExitStack as DoesNotRaise
from typing import Any
import cv2
import numpy as np
import pytest
from supervision.config import CLASS_NAME_DATA_FIELD
from supervision.detection.compact_mask import CompactMask
from supervision.detection.utils.internal import (
extract_ultralytics_masks,
get_data_item,
merge_data,
merge_metadata,
process_roboflow_result,
)
class _FakeMasksData:
"""Ultralytics-like mask tensor exposing shape and cpu().numpy()."""
def __init__(self, arr: np.ndarray) -> None:
self._arr = np.asarray(arr, dtype=np.float32)
self.shape = self._arr.shape
def cpu(self) -> "_FakeMasksData":
return self
def numpy(self) -> np.ndarray:
return self._arr
class _FakeMasks:
"""Ultralytics-like masks container holding a data tensor."""
def __init__(self, data: _FakeMasksData) -> None:
self.data = data
def __bool__(self) -> bool:
return True
class _FakeYOLOMaskResults:
"""Minimal Ultralytics results exposing masks and orig_shape."""
def __init__(self, masks_arr: np.ndarray, orig_shape: tuple[int, int]) -> None:
self.masks = _FakeMasks(_FakeMasksData(masks_arr))
self.orig_shape = orig_shape
def test_extract_ultralytics_masks_thresholds_resized_proto_at_half() -> None:
"""Resized proto masks threshold at 0.5, so interpolated edges don't dilate."""
orig_shape = (4, 4)
proto = np.array([[[1.0, 1.0], [0.0, 0.0]]], dtype=np.float32)
results = _FakeYOLOMaskResults(masks_arr=proto, orig_shape=orig_shape)
masks = extract_ultralytics_masks(results)
resized = cv2.resize(proto[0], (orig_shape[1], orig_shape[0]))
assert masks is not None
assert masks.dtype == bool
np.testing.assert_array_equal(masks[0], resized > 0.5)
# A naive `> 0` cast would flag the interpolated boundary row as True.
assert masks[0].sum() < int((resized > 0).sum())
def _pred(
yx: tuple[float, float] = (1.5, 1.5),
size: tuple[float, float] = (2.0, 2.0),
confidence: float = 0.9,
class_id: int = 0,
class_name: str = "person",
**extra: Any,
) -> dict[str, Any]:
"""Build a minimal Roboflow prediction dict; `extra` adds/overrides fields."""
pred: dict[str, Any] = {
"x": yx[1],
"y": yx[0],
"width": size[0],
"height": size[1],
"confidence": confidence,
"class_id": class_id,
"class": class_name,
}
pred.update(extra)
return pred
def _result(
*predictions: dict[str, Any],
img_w: int = 4,
img_h: int = 4,
) -> dict[str, Any]:
"""Wrap predictions in a Roboflow result envelope."""
return {
"predictions": list(predictions),
"image": {"width": img_w, "height": img_h},
}
def _result_1k(*predictions: dict[str, Any]) -> dict[str, Any]:
"""Wrap predictions in a 1000x1000 Roboflow result envelope."""
return _result(*predictions, img_w=1000, img_h=1000)
TEST_MASK = np.zeros((1, 1000, 1000), dtype=bool)
TEST_MASK[:, 300:351, 200:251] = True
TEST_RLE_MASK = np.zeros((1, 4, 4), dtype=bool)
TEST_RLE_MASK[0, 1:3, 1:3] = True
TEST_RLE_NONCONTIGUOUS_MASK = np.zeros((1, 4, 4), dtype=bool)
TEST_RLE_NONCONTIGUOUS_MASK[0, 0:2, 0:2] = True
TEST_RLE_NONCONTIGUOUS_MASK[0, 3, 2:4] = True
@pytest.mark.parametrize(
("roboflow_result", "expected_result", "exception"),
[
(
{"predictions": [], "image": {"width": 1000, "height": 1000}},
(
np.empty((0, 4)),
np.empty(0),
np.empty(0),
None,
None,
{CLASS_NAME_DATA_FIELD: np.empty(0, dtype=str)},
),
DoesNotRaise(),
), # empty result
(
_result_1k(_pred(yx=(300.0, 200.0), size=(50.0, 50.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.array(["person"])},
),
DoesNotRaise(),
), # single correct object detection result
(
_result_1k(
_pred(yx=(300.0, 200.0), size=(50.0, 50.0), tracker_id=1),
_pred(
yx=(500.0, 500.0),
size=(100.0, 100.0),
confidence=0.8,
class_id=7,
class_name="truck",
tracker_id=2,
),
),
(
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,
np.array([1, 2]),
{CLASS_NAME_DATA_FIELD: np.array(["person", "truck"])},
),
DoesNotRaise(),
), # two correct object detection result
(
_result_1k(
_pred(
yx=(300.0, 200.0),
size=(50.0, 50.0),
points=[],
tracker_id=None,
),
),
(
np.array([[175.0, 275.0, 225.0, 325.0]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single invalid polygon result with no points falls back to box-only
(
_result_1k(
_pred(
yx=(300.0, 200.0),
size=(50.0, 50.0),
points=[{"x": 200.0, "y": 300.0}, {"x": 250.0, "y": 300.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.array(["person"])},
),
DoesNotRaise(),
), # single invalid polygon result with too few points falls back to box-only
(
_result_1k(
_pred(
yx=(300.0, 200.0),
size=(50.0, 50.0),
points=[
{"x": 200.0, "y": 300.0},
{"x": 250.0, "y": 300.0},
{"x": 250.0, "y": 350.0},
{"x": 200.0, "y": 350.0},
],
),
),
(
np.array([[175.0, 275.0, 225.0, 325.0]]),
np.array([0.9]),
np.array([0]),
TEST_MASK,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single incorrect instance segmentation result with no enough points
(
_result_1k(
_pred(
yx=(300.0, 200.0),
size=(50.0, 50.0),
points=[
{"x": 200.0, "y": 300.0},
{"x": 250.0, "y": 300.0},
{"x": 250.0, "y": 350.0},
{"x": 200.0, "y": 350.0},
],
),
_pred(
yx=(500.0, 500.0),
size=(100.0, 100.0),
confidence=0.8,
class_id=7,
class_name="truck",
points=[],
),
),
(
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,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person", "truck"])},
),
DoesNotRaise(),
), # mixed valid polygon and invalid polygon keeps boxes and drops masks
(
_result(_pred(rle={"size": [4, 4], "counts": "52203"})),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
TEST_RLE_MASK,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single RLE prediction with compressed string counts
(
_result(
_pred(
yx=(2.0, 2.0),
size=(4.0, 4.0),
confidence=0.85,
class_id=1,
class_name="cat",
rle={"size": [4, 4], "counts": "02203ON0"},
)
),
(
np.array([[0.0, 0.0, 4.0, 4.0]]),
np.array([0.85]),
np.array([1]),
TEST_RLE_NONCONTIGUOUS_MASK,
None,
{CLASS_NAME_DATA_FIELD: np.array(["cat"])},
),
DoesNotRaise(),
), # single RLE prediction with non-contiguous mask
(
_result(_pred(rle={"size": [4, 4], "counts": "52203"}, tracker_id=5)),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
TEST_RLE_MASK,
np.array([5]),
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # RLE prediction with tracker_id
(
_result(_pred(rle_mask={"size": [4, 4], "counts": "52203"})),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
TEST_RLE_MASK,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # single RLE prediction with compressed string counts under rle_mask key
(
_result(_pred(rle="bad_string")),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # malformed RLE payload should fall through to box-only detection
(
_result(_pred(rle={"size": [4, 4]})),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # RLE dict missing counts falls through to box-only detection
(
_result(_pred(rle={"size": [4, 4], "counts": "!"})),
(
np.array([[0.5, 0.5, 2.5, 2.5]]),
np.array([0.9]),
np.array([0]),
None,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person"])},
),
DoesNotRaise(),
), # malformed compressed counts falls through to box-only detection
(
_result(
_pred(rle={"size": [4, 4], "counts": "52203"}),
_pred(yx=(3.0, 3.0), confidence=0.8, class_id=1, class_name="car"),
),
(
np.array([[0.5, 0.5, 2.5, 2.5], [2.0, 2.0, 4.0, 4.0]]),
np.array([0.9, 0.8]),
np.array([0, 1]),
# Mixed-modality batch: only a subset carries masks.
# All masks dropped to preserve xyxy alignment (mirrors
# the tracker_id mixed-batch handling).
None,
None,
{CLASS_NAME_DATA_FIELD: np.array(["person", "car"])},
),
DoesNotRaise(),
), # mixed RLE + box-only batch — masks dropped to preserve xyxy alignment
pytest.param(
_result(
_pred(tracker_id=7),
_pred(yx=(3.0, 3.0), confidence=0.8, class_id=1, class_name="car"),
),
(
np.array([[0.5, 0.5, 2.5, 2.5], [2.0, 2.0, 4.0, 4.0]]),
np.array([0.9, 0.8]),
np.array([0, 1]),
None,
# tracker_id is None when only some detections carry one, rather
# than an array misaligned with xyxy that would raise ValueError.
None,
{CLASS_NAME_DATA_FIELD: np.array(["person", "car"])},
),
DoesNotRaise(),
id="mixed_tracker_id_batch_drops_to_none",
),
pytest.param(
_result(
_pred(),
_pred(yx=(3.0, 3.0), confidence=0.8, class_id=1, class_name="car"),
),
(
np.array([[0.5, 0.5, 2.5, 2.5], [2.0, 2.0, 4.0, 4.0]]),
np.array([0.9, 0.8]),
np.array([0, 1]),
None,
# None not in tracker_ids prevents np.array([None, None], dtype=int64)
None,
{CLASS_NAME_DATA_FIELD: np.array(["person", "car"])},
),
DoesNotRaise(),
id="all_absent_tracker_id_no_raise",
),
],
)
def test_process_roboflow_result(
roboflow_result: dict,
expected_result: tuple[
np.ndarray, np.ndarray, np.ndarray, np.ndarray | None, np.ndarray
],
exception: Exception,
) -> None:
with exception:
result = process_roboflow_result(roboflow_result=roboflow_result)
assert np.array_equal(result[0], expected_result[0])
assert np.array_equal(result[1], expected_result[1])
assert np.array_equal(result[2], expected_result[2])
assert (result[3] is None and expected_result[3] is None) or (
np.array_equal(result[3], expected_result[3])
)
assert (result[4] is None and expected_result[4] is None) or (
np.array_equal(result[4], expected_result[4])
)
for key in result[5]:
if isinstance(result[5][key], np.ndarray):
assert np.array_equal(result[5][key], expected_result[5][key]), (
f"Mismatch in arrays for key {key}"
)
assert result[5][key].dtype == expected_result[5][key].dtype, (
f"dtype mismatch for key {key}: "
f"got {result[5][key].dtype}, "
f"expected {expected_result[5][key].dtype}"
)
else:
assert result[5][key] == expected_result[5][key], (
f"Mismatch in non-array data for key {key}"
)
def test_process_roboflow_result_compact_masks_returns_compact_mask() -> None:
"""compact_masks=True should return CompactMask for valid RLE predictions."""
roboflow_result = _result(_pred(rle={"size": [4, 4], "counts": "52203"}))
result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert isinstance(result[3], CompactMask)
np.testing.assert_array_equal(result[3].to_dense(), TEST_RLE_MASK)
def test_process_roboflow_result_compact_masks_matches_resized_dense_rle() -> None:
"""compact_masks=True should preserve current RLE resize behavior."""
roboflow_result = _result(
_pred(yx=(2.0, 2.0), size=(4.0, 4.0), rle={"size": [2, 2], "counts": [0, 4]})
)
dense_result = process_roboflow_result(roboflow_result=roboflow_result)
compact_result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert isinstance(compact_result[3], CompactMask)
np.testing.assert_array_equal(compact_result[3].to_dense(), dense_result[3])
def test_process_roboflow_result_compact_masks_invalid_rle_is_box_only() -> None:
"""compact_masks=True should keep malformed RLE fallback behavior."""
roboflow_result = _result(_pred(rle={"size": [4, 4], "counts": "!"}))
result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert result[3] is None
np.testing.assert_array_equal(result[0], np.array([[0.5, 0.5, 2.5, 2.5]]))
def test_process_roboflow_result_compact_masks_overflow_rle_is_box_only() -> None:
"""compact_masks=True should not leak OverflowError from invalid counts."""
roboflow_result = _result(_pred(rle={"size": [4, 4], "counts": [2**31]}))
result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert result[3] is None
np.testing.assert_array_equal(result[0], np.array([[0.5, 0.5, 2.5, 2.5]]))
def test_process_roboflow_result_compact_masks_partial_failure_drops_all_masks() -> (
None
):
"""One invalid RLE in a two-prediction batch drops all masks but keeps both xyxy."""
roboflow_result = _result(
_pred(rle={"size": [4, 4], "counts": "52203"}), # valid: sum == 16
_pred(rle={"size": [4, 4], "counts": [1, 2, 3]}), # invalid: sum == 6, not 16
)
result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
# Mixed-modality: one mask decoded, one failed → all masks dropped for alignment.
assert result[3] is None
# Both detections preserved in xyxy.
assert result[0].shape == (2, 4)
def test_process_roboflow_result_uses_rle_mask_when_rle_invalid() -> None:
"""Valid rle_mask should be used when rle is present but invalid."""
roboflow_result = _result(
_pred(rle={"foo": "bar"}, rle_mask={"size": [4, 4], "counts": "52203"})
)
dense_result = process_roboflow_result(roboflow_result=roboflow_result)
compact_result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert isinstance(dense_result[3], np.ndarray)
assert isinstance(compact_result[3], CompactMask)
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(
_pred(
yx=(2.5, 2.5),
size=(4.0, 4.0),
confidence=0.75,
class_name="dog",
points=[
{"x": 1, "y": 1},
{"x": 4, "y": 1},
{"x": 4, "y": 4},
{"x": 1, "y": 4},
],
),
img_w=6,
img_h=6,
)
_, _, _, masks, _, _ = process_roboflow_result(roboflow_result, compact_masks=True)
assert isinstance(masks, CompactMask)
assert len(masks) == 1
def test_box_only_compact_masks_true_returns_none_mask() -> None:
"""box-only predictions with compact_masks=True yield None mask."""
roboflow_result = _result(
_pred(yx=(2.0, 2.0), size=(3.0, 3.0), class_name="cat"),
img_w=5,
img_h=5,
)
_, _, _, masks, _, _ = process_roboflow_result(roboflow_result, compact_masks=True)
assert masks is None
def test_rle_size_mismatch_resizes_dense_mask() -> None:
"""Dense path resizes mask when RLE size differs from image dimensions."""
# counts=[0, 4]: 0 False runs then 4 True runs — all-True 2x2 mask.
# Image is 4x4, so cv2.resize must expand the decoded 2x2 to 4x4.
roboflow_result = _result(
_pred(
yx=(2.0, 2.0),
size=(4.0, 4.0),
class_name="cat",
rle_mask={"size": [2, 2], "counts": [0, 4]},
)
)
_, _, _, masks, _, _ = process_roboflow_result(roboflow_result, compact_masks=False)
assert masks is not None
assert masks.shape == (1, 4, 4)
assert masks[0].sum() > 0
@pytest.mark.parametrize(
("data_list", "expected_result", "exception"),
[
(
[],
{},
DoesNotRaise(),
), # empty data list
(
[{}],
{},
DoesNotRaise(),
), # single empty data dict
(
[{}, {}],
{},
DoesNotRaise(),
), # two empty data dicts
(
[
{"test_1": []},
],
{"test_1": []},
DoesNotRaise(),
), # single data dict with a single field name and empty list values
(
[
{"test_1": []},
{"test_1": []},
],
{"test_1": []},
DoesNotRaise(),
), # two data dicts with the same field name and empty list values
(
[
{"test_1": np.array([])},
],
{"test_1": np.array([])},
DoesNotRaise(),
), # single data dict with a single field name and empty np.array values
(
[
{"test_1": np.array([])},
{"test_1": np.array([])},
],
{"test_1": np.array([])},
DoesNotRaise(),
), # two data dicts with the same field name and empty np.array values
(
[
{"test_1": [1, 2, 3]},
],
{"test_1": [1, 2, 3]},
DoesNotRaise(),
), # single data dict with a single field name and list values
(
[
{"test_1": []},
{"test_1": [3, 2, 1]},
],
{"test_1": [3, 2, 1]},
DoesNotRaise(),
), # two data dicts with the same field name; one of with empty list as value
(
[
{"test_1": [1, 2, 3]},
{"test_1": [3, 2, 1]},
],
{"test_1": [1, 2, 3, 3, 2, 1]},
DoesNotRaise(),
), # two data dicts with the same field name and list values
(
[
{"test_1": [1, 2, 3]},
{"test_1": [3, 2, 1]},
{"test_1": [1, 2, 3]},
],
{"test_1": [1, 2, 3, 3, 2, 1, 1, 2, 3]},
DoesNotRaise(),
), # three data dicts with the same field name and list values
(
[
{"test_1": [1, 2, 3]},
{"test_2": [3, 2, 1]},
],
None,
pytest.raises(ValueError, match="same keys to merge"),
), # two data dicts with different field names
(
[
{"test_1": np.array([1, 2, 3])},
{"test_1": np.array([3, 2, 1])},
],
{"test_1": np.array([1, 2, 3, 3, 2, 1])},
DoesNotRaise(),
), # two data dicts with the same field name and np.array values as 1D arrays
(
[
{"test_1": np.array([[1, 2, 3]])},
{"test_1": np.array([[3, 2, 1]])},
],
{"test_1": np.array([[1, 2, 3], [3, 2, 1]])},
DoesNotRaise(),
), # two data dicts with the same field name and np.array values as 2D arrays
(
[
{"test_1": np.array([1, 2, 3]), "test_2": np.array(["a", "b", "c"])},
{"test_1": np.array([3, 2, 1]), "test_2": np.array(["c", "b", "a"])},
],
{
"test_1": np.array([1, 2, 3, 3, 2, 1]),
"test_2": np.array(["a", "b", "c", "c", "b", "a"]),
},
DoesNotRaise(),
), # two data dicts with the same field names and np.array values
(
[
{"test_1": [1, 2, 3], "test_2": np.array(["a", "b", "c"])},
{"test_1": [3, 2, 1], "test_2": np.array(["c", "b", "a"])},
],
{
"test_1": [1, 2, 3, 3, 2, 1],
"test_2": np.array(["a", "b", "c", "c", "b", "a"]),
},
DoesNotRaise(),
), # two data dicts with the same field names and mixed values
(
[
{"test_1": np.array([1, 2, 3])},
{"test_1": np.array([[3, 2, 1]])},
],
None,
pytest.raises(ValueError, match="same number of dimensions"),
), # two data dicts with the same field name and 1D and 2D arrays values
(
[
{"test_1": np.array([1, 2, 3]), "test_2": np.array(["a", "b"])},
{"test_1": np.array([3, 2, 1]), "test_2": np.array(["c", "b", "a"])},
],
None,
pytest.raises(ValueError, match="equal length"),
), # two data dicts with the same field name and different length arrays values
(
[{}, {"test_1": [1, 2, 3]}],
None,
pytest.raises(ValueError, match="same keys to merge"),
), # two data dicts; one empty and one non-empty dict
(
[{"test_1": [], "test_2": []}, {"test_1": [1, 2, 3], "test_2": [1, 2, 3]}],
{"test_1": [1, 2, 3], "test_2": [1, 2, 3]},
DoesNotRaise(),
), # two data dicts; one empty and one non-empty dict; same keys
(
[{"test_1": []}, {"test_1": [1, 2, 3], "test_2": [4, 5, 6]}],
None,
pytest.raises(ValueError, match="same keys to merge"),
), # two data dicts; one empty and one non-empty dict; different keys
(
[
{
"test_1": [1, 2, 3],
"test_2": [4, 5, 6],
"test_3": [7, 8, 9],
},
{"test_1": [1, 2, 3], "test_2": [4, 5, 6]},
],
None,
pytest.raises(ValueError, match="same keys to merge"),
), # two data dicts; one with three keys, one with two keys
(
[
{"test_1": [1, 2, 3]},
{"test_1": [1, 2, 3], "test_2": [1, 2, 3]},
],
None,
pytest.raises(ValueError, match="same keys to merge"),
), # some keys missing in one dict
(
[
{"test_1": [1, 2, 3], "test_2": ["a", "b"]},
{"test_1": [4, 5], "test_2": ["c", "d", "e"]},
],
None,
pytest.raises(ValueError, match="equal length"),
), # different value lengths for the same key
],
)
def test_merge_data(
data_list: list[dict[str, Any]],
expected_result: dict[str, Any] | None,
exception: Exception,
) -> None:
with exception:
result = merge_data(data_list=data_list)
if expected_result is None:
pytest.fail(f"Expected an error, but got result {result}")
for key in result:
if isinstance(result[key], np.ndarray):
assert np.array_equal(result[key], expected_result[key]), (
f"Mismatch in arrays for key {key}"
)
else:
assert result[key] == expected_result[key], (
f"Mismatch in non-array data for key {key}"
)
@pytest.mark.parametrize(
("data", "index", "expected_result", "exception"),
[
({}, 0, {}, DoesNotRaise()), # empty data dict
(
{
"test_1": [1, 2, 3],
},
0,
{
"test_1": [1],
},
DoesNotRaise(),
), # data dict with a single list field and integer index
(
{
"test_1": np.array([1, 2, 3]),
},
0,
{
"test_1": np.array([1]),
},
DoesNotRaise(),
), # data dict with a single np.array field and integer index
(
{
"test_1": [1, 2, 3],
},
slice(0, 2),
{
"test_1": [1, 2],
},
DoesNotRaise(),
), # data dict with a single list field and slice index
(
{
"test_1": np.array([1, 2, 3]),
},
slice(0, 2),
{
"test_1": np.array([1, 2]),
},
DoesNotRaise(),
), # data dict with a single np.array field and slice index
(
{
"test_1": [1, 2, 3],
},
-1,
{
"test_1": [3],
},
DoesNotRaise(),
), # data dict with a single list field and negative integer index
(
{
"test_1": np.array([1, 2, 3]),
},
-1,
{
"test_1": np.array([3]),
},
DoesNotRaise(),
), # data dict with a single np.array field and negative integer index
(
{
"test_1": [1, 2, 3],
},
[0, 2],
{
"test_1": [1, 3],
},
DoesNotRaise(),
), # data dict with a single list field and integer list index
(
{
"test_1": np.array([1, 2, 3]),
},
[0, 2],
{
"test_1": np.array([1, 3]),
},
DoesNotRaise(),
), # data dict with a single np.array field and integer list index
(
{
"test_1": [1, 2, 3],
},
np.array([0, 2]),
{
"test_1": [1, 3],
},
DoesNotRaise(),
), # data dict with a single list field and integer np.array index
(
{
"test_1": np.array([1, 2, 3]),
},
np.array([0, 2]),
{
"test_1": np.array([1, 3]),
},
DoesNotRaise(),
), # data dict with a single np.array field and integer np.array index
(
{
"test_1": np.array([1, 2, 3]),
},
np.array([True, True, True]),
{
"test_1": np.array([1, 2, 3]),
},
DoesNotRaise(),
), # data dict with a single np.array field and all-true bool np.array index
(
{
"test_1": np.array([1, 2, 3]),
},
np.array([False, False, False]),
{
"test_1": np.array([]),
},
DoesNotRaise(),
), # data dict with a single np.array field and all-false bool np.array index
(
{
"test_1": np.array([1, 2, 3]),
},
np.array([False, True, False]),
{
"test_1": np.array([2]),
},
DoesNotRaise(),
), # data dict with a single np.array field and mixed bool np.array index
(
{"test_1": np.array([1, 2, 3]), "test_2": ["a", "b", "c"]},
0,
{"test_1": np.array([1]), "test_2": ["a"]},
DoesNotRaise(),
), # data dict with two fields and integer index
(
{"test_1": np.array([1, 2, 3]), "test_2": ["a", "b", "c"]},
-1,
{"test_1": np.array([3]), "test_2": ["c"]},
DoesNotRaise(),
), # data dict with two fields and negative integer index
(
{"test_1": np.array([1, 2, 3]), "test_2": ["a", "b", "c"]},
np.array([False, True, False]),
{"test_1": np.array([2]), "test_2": ["b"]},
DoesNotRaise(),
), # data dict with two fields and mixed bool np.array index
],
)
def test_get_data_item(
data: dict[str, Any],
index: Any,
expected_result: dict[str, Any] | None,
exception: Exception,
) -> None:
with exception:
result = get_data_item(data=data, index=index)
for key in result:
if isinstance(result[key], np.ndarray):
assert np.array_equal(result[key], expected_result[key]), (
f"Mismatch in arrays for key {key}"
)
else:
assert result[key] == expected_result[key], (
f"Mismatch in non-array data for key {key}"
)
@pytest.mark.parametrize(
("metadata_list", "expected_result", "exception"),
[
# Identical metadata with a single key
([{"key1": "value1"}, {"key1": "value1"}], {"key1": "value1"}, DoesNotRaise()),
# Identical metadata with multiple keys
(
[
{"key1": "value1", "key2": "value2"},
{"key1": "value1", "key2": "value2"},
],
{"key1": "value1", "key2": "value2"},
DoesNotRaise(),
),
# Conflicting values for the same key
(
[{"key1": "value1"}, {"key1": "value2"}],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1'\\."),
),
# Different sets of keys across dictionaries
(
[{"key1": "value1"}, {"key2": "value2"}],
None,
pytest.raises(ValueError, match="same keys to merge"),
),
# Empty metadata list
([], {}, DoesNotRaise()),
# Empty metadata dictionaries
([{}, {}], {}, DoesNotRaise()),
# Different declaration order for keys
(
[
{"key1": "value1", "key2": "value2"},
{"key2": "value2", "key1": "value1"},
],
{"key1": "value1", "key2": "value2"},
DoesNotRaise(),
),
# Nested metadata dictionaries
(
[{"key1": {"sub_key": "sub_value"}}, {"key1": {"sub_key": "sub_value"}}],
{"key1": {"sub_key": "sub_value"}},
DoesNotRaise(),
),
# Large metadata dictionaries with many keys
(
[
{f"key{i}": f"value{i}" for i in range(100)},
{f"key{i}": f"value{i}" for i in range(100)},
],
{f"key{i}": f"value{i}" for i in range(100)},
DoesNotRaise(),
),
# Mixed types in list metadata values
(
[{"key1": ["value1", 2, True]}, {"key1": ["value1", 2, True]}],
{"key1": ["value1", 2, True]},
DoesNotRaise(),
),
# Identical lists across metadata dictionaries
(
[{"key1": [1, 2, 3]}, {"key1": [1, 2, 3]}],
{"key1": [1, 2, 3]},
DoesNotRaise(),
),
# Identical numpy arrays across metadata dictionaries
(
[{"key1": np.array([1, 2, 3])}, {"key1": np.array([1, 2, 3])}],
{"key1": np.array([1, 2, 3])},
DoesNotRaise(),
),
# Identical numpy arrays across metadata dictionaries, different datatype
(
[
{"key1": np.array([1, 2, 3], dtype=np.int32)},
{"key1": np.array([1, 2, 3], dtype=np.int64)},
],
{"key1": np.array([1, 2, 3])},
DoesNotRaise(),
),
# Conflicting lists for the same key
(
[{"key1": [1, 2, 3]}, {"key1": [4, 5, 6]}],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1'\\."),
),
# Conflicting numpy arrays for the same key
(
[{"key1": np.array([1, 2, 3])}, {"key1": np.array([4, 5, 6])}],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1':"),
),
# Mixed data types: list and numpy array for the same key
(
[{"key1": [1, 2, 3]}, {"key1": np.array([1, 2, 3])}],
None,
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=(
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
(
[{"key1": [[1, 2], [3, 4]]}, {"key1": [[1, 2], [3, 4]]}],
{"key1": [[1, 2], [3, 4]]},
DoesNotRaise(),
),
# Identical multi-dimensional numpy arrays across metadata dictionaries
(
[
{"key1": np.arange(4).reshape(2, 2)},
{"key1": np.arange(4).reshape(2, 2)},
],
{"key1": np.arange(4).reshape(2, 2)},
DoesNotRaise(),
),
# Conflicting multi-dimensional lists for the same key
(
[{"key1": [[1, 2], [3, 4]]}, {"key1": [[5, 6], [7, 8]]}],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1'\\."),
),
# Conflicting multi-dimensional numpy arrays for the same key
(
[
{"key1": np.arange(4).reshape(2, 2)},
{"key1": np.arange(4, 8).reshape(2, 2)},
],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1':"),
),
# Mixed types with multi-dimensional list and array for the same key
(
[{"key1": [[1, 2], [3, 4]]}, {"key1": np.arange(4).reshape(2, 2)}],
None,
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
(
[
{"key1": np.arange(8).reshape(2, 2, 2)},
{"key1": np.arange(8).reshape(2, 2, 2)},
],
{"key1": np.arange(8).reshape(2, 2, 2)},
DoesNotRaise(),
),
# Differently-shaped higher-dimensional (3D) numpy arrays
# across metadata dictionaries
(
[
{"key1": np.arange(8).reshape(2, 2, 2)},
{"key1": np.arange(8).reshape(4, 1, 2)},
],
None,
pytest.raises(ValueError, match="Conflicting metadata for key: 'key1':"),
),
],
)
def test_merge_metadata(metadata_list, expected_result, exception) -> None:
with exception:
result = merge_metadata(metadata_list)
if expected_result is None:
assert result is None, f"Expected an error, but got a result {result}"
for key, value in result.items():
assert key in expected_result
if isinstance(value, np.ndarray):
np.testing.assert_array_equal(value, expected_result[key])
else:
assert value == expected_result[key]
def test_process_roboflow_result_compact_masks_batch_retry_logs_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Batch RLE failure triggers per-prediction retry; mixed-modality drops masks."""
import logging
roboflow_result = _result(
_pred(rle={"size": [4, 4], "counts": "52203"}),
_pred(rle={"size": [4, 4], "counts": "52203"}),
# sum=6 != 16 — triggers batch failure → per-prediction retry
_pred(rle={"size": [4, 4], "counts": [1, 2, 3]}),
)
with caplog.at_level(logging.WARNING):
result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
# Batch call fails, per-prediction retry produces 2/3 decoded → mixed-modality drop.
assert "Batch compact RLE decode failed" in caplog.text
assert result[3] is None
assert result[0].shape == (3, 4)
def test_process_roboflow_result_compact_masks_rle_mask_size_mismatch() -> None:
"""rle_mask key + size mismatch triggers resize fallback with compact_masks=True."""
# RLE is 2x2; image is 4x4 — size mismatch triggers resize fallback.
roboflow_result = _result(
_pred(
yx=(2.0, 2.0),
size=(4.0, 4.0),
rle_mask={"size": [2, 2], "counts": [0, 4]},
)
)
dense_result = process_roboflow_result(roboflow_result=roboflow_result)
compact_result = process_roboflow_result(
roboflow_result=roboflow_result, compact_masks=True
)
assert isinstance(compact_result[3], CompactMask)
np.testing.assert_array_equal(compact_result[3].to_dense(), dense_result[3])
# ---------------------------------------------------------------------------
# cross_product — regression + unit tests (GitHub #2384)
# ---------------------------------------------------------------------------
import warnings # noqa: E402
from supervision.detection.utils.internal import cross_product # noqa: E402
from supervision.geometry.core import Point, Vector # noqa: E402
def test_cross_product_no_deprecation_warning() -> None:
"""Regression for #2384: cross_product must not fire DeprecationWarning."""
anchors = np.array([[[5.0, 5.0]]])
v = Vector(start=Point(0, 0), end=Point(10, 0))
with warnings.catch_warnings():
warnings.simplefilter("error", DeprecationWarning)
cross_product(anchors, v)
@pytest.mark.parametrize(
("anchors", "vector", "expected_sign"),
[
pytest.param(
np.array([[[5.0, 5.0]]]),
Vector(Point(0, 0), Point(10, 0)),
1,
id="above",
),
pytest.param(
np.array([[[5.0, -5.0]]]),
Vector(Point(0, 0), Point(10, 0)),
-1,
id="below",
),
pytest.param(
np.array([[[5.0, 0.0]]]),
Vector(Point(0, 0), Point(10, 0)),
0,
id="on-line",
),
pytest.param(
np.array([[[3.0, 3.0]]]),
Vector(Point(1, 1), Point(5, 1)),
1,
id="offset-start",
),
],
)
def test_cross_product_sign(
anchors: np.ndarray, vector: Vector, expected_sign: int
) -> None:
"""Verify cross_product returns correct sign for known anchor/vector pairs."""
result = cross_product(anchors, vector)
assert int(np.sign(result[0, 0])) == expected_sign