Add out-of-bounds detection warning to InferenceSlicer (#2186)
When a user's callback accidentally runs inference on the full image instead of the provided slice, detections get incorrect offsets applied, causing a repeating grid pattern. Add a validation check in _run_callback that emits a SupervisionWarnings warning when any detection coordinate exceeds the slice dimensions or is negative. An instance flag prevents repeated warnings across many slices. - Wrap _out_of_slice_bounds_warned check-and-set in threading.Lock to prevent duplicate warnings under ThreadPoolExecutor with thread_workers > 1 - Change stacklevel=2 to stacklevel=1 — under executor.submit the stacklevel=2 frame points into concurrent.futures internals, not user code - Assert exactly 1 warning fires with thread_workers=4 (validates Lock fix) - Assert no warning for detection touching but not exceeding slice boundary (pins > vs >= semantics) - Assert second slicer call does not re-warn (documents once-per-instance semantic) - Extract warning message into `msg` variable to satisfy E501 line-length limit --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: Borda <6035284+Borda@users.noreply.github.com> Co-authored-by: Claude Code <noreply@anthropic.com> Co-authored-by: Copilot <175728472+Copilot@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:
Copilot
GitHub
parent
8b60ee1a0e
commit
e6fab4b7fa
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import warnings
|
||||
from collections.abc import Callable
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
|
|
@ -135,6 +136,8 @@ class InferenceSlicer:
|
|||
self.overlap_filter = OverlapFilter.from_value(overlap_filter)
|
||||
self.callback: Callable[[ImageType], Detections] = callback
|
||||
self.thread_workers = thread_workers
|
||||
self._out_of_slice_bounds_warned: bool = False
|
||||
self._out_of_slice_bounds_lock = threading.Lock()
|
||||
|
||||
def __call__(self, image: ImageType) -> Detections:
|
||||
"""
|
||||
|
|
@ -198,6 +201,33 @@ class InferenceSlicer:
|
|||
detections = self.callback(image_slice)
|
||||
resolution_wh = get_image_resolution_wh(image)
|
||||
|
||||
# Fast-path: skip locking and bounds checking when the warning has already
|
||||
# been emitted or when there are no detections to inspect.
|
||||
needs_warning_check = (
|
||||
not self._out_of_slice_bounds_warned and len(detections) > 0
|
||||
)
|
||||
|
||||
if needs_warning_check:
|
||||
with self._out_of_slice_bounds_lock:
|
||||
# Re-check under the lock to ensure correctness with multiple threads.
|
||||
if not self._out_of_slice_bounds_warned and len(detections) > 0:
|
||||
slice_width = offset[2] - offset[0]
|
||||
slice_height = offset[3] - offset[1]
|
||||
x_exceeds = np.any(detections.xyxy[:, [0, 2]] > slice_width)
|
||||
y_exceeds = np.any(detections.xyxy[:, [1, 3]] > slice_height)
|
||||
x_negative = np.any(detections.xyxy[:, [0, 2]] < 0)
|
||||
y_negative = np.any(detections.xyxy[:, [1, 3]] < 0)
|
||||
if x_exceeds or y_exceeds or x_negative or y_negative:
|
||||
self._out_of_slice_bounds_warned = True
|
||||
msg = (
|
||||
"Detections returned by the callback have coordinates "
|
||||
"outside the slice bounds. This may be caused by the "
|
||||
"callback running inference on the full image instead of "
|
||||
"the provided image slice. Ensure your callback uses the "
|
||||
"input slice for inference, not the original "
|
||||
"full-resolution image."
|
||||
)
|
||||
warnings.warn(msg, category=SupervisionWarnings, stacklevel=2)
|
||||
detections = move_detections(
|
||||
detections=detections,
|
||||
offset=offset[:2],
|
||||
|
|
|
|||
|
|
@ -1,10 +1,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import warnings
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from supervision.detection.core import Detections
|
||||
from supervision.detection.tools.inference_slicer import InferenceSlicer
|
||||
from supervision.utils.internal import SupervisionWarnings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -195,3 +198,187 @@ def test_generate_offset(
|
|||
assert np.array_equal(offsets, expected_offsets), (
|
||||
f"Expected {expected_offsets}, got {offsets}"
|
||||
)
|
||||
|
||||
|
||||
def test_run_callback_warns_when_detections_outside_slice_bounds() -> None:
|
||||
"""Test that a warning is emitted when callback returns detections with
|
||||
coordinates outside the slice bounds."""
|
||||
|
||||
def out_of_bounds_callback(_: np.ndarray) -> Detections:
|
||||
# Return detections with coordinates exceeding the 64x64 slice size
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 128, 128]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((128, 128, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(callback=out_of_bounds_callback, slice_wh=64, overlap_wh=0)
|
||||
|
||||
with pytest.warns(SupervisionWarnings, match="outside the slice bounds"):
|
||||
slicer(image)
|
||||
|
||||
|
||||
def test_run_callback_warns_only_once_for_out_of_bounds_detections() -> None:
|
||||
"""Test that the out-of-bounds warning is only emitted once even across
|
||||
multiple slices."""
|
||||
|
||||
def out_of_bounds_callback(_: np.ndarray) -> Detections:
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 128, 128]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((256, 256, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(callback=out_of_bounds_callback, slice_wh=64, overlap_wh=0)
|
||||
|
||||
with warnings.catch_warnings(record=True) as recorded_warnings:
|
||||
warnings.simplefilter("always")
|
||||
slicer(image)
|
||||
|
||||
out_of_bounds_warnings = [
|
||||
w
|
||||
for w in recorded_warnings
|
||||
if issubclass(w.category, SupervisionWarnings)
|
||||
and "outside the slice bounds" in str(w.message)
|
||||
]
|
||||
assert len(out_of_bounds_warnings) == 1
|
||||
|
||||
|
||||
def test_run_callback_no_warning_when_detections_inside_slice_bounds() -> None:
|
||||
"""Test that no warning is emitted when callback returns detections within
|
||||
the slice bounds."""
|
||||
|
||||
def in_bounds_callback(_: np.ndarray) -> Detections:
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 10, 10]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((128, 128, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(callback=in_bounds_callback, slice_wh=64, overlap_wh=0)
|
||||
|
||||
with warnings.catch_warnings(record=True) as recorded_warnings:
|
||||
warnings.simplefilter("always")
|
||||
slicer(image)
|
||||
|
||||
out_of_bounds_warnings = [
|
||||
w
|
||||
for w in recorded_warnings
|
||||
if issubclass(w.category, SupervisionWarnings)
|
||||
and "outside the slice bounds" in str(w.message)
|
||||
]
|
||||
assert len(out_of_bounds_warnings) == 0
|
||||
|
||||
|
||||
def test_run_callback_warns_when_detections_have_negative_coordinates() -> None:
|
||||
"""Test that a warning is emitted when callback returns detections with
|
||||
negative coordinates, indicating wrong reference frame."""
|
||||
|
||||
def negative_coords_callback(_: np.ndarray) -> Detections:
|
||||
# Return detections with negative coordinates (e.g., returned in full-image
|
||||
# coordinates that are to the left/top of this slice's origin)
|
||||
return Detections(
|
||||
xyxy=np.array([[-10, -10, 10, 10]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((128, 128, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(
|
||||
callback=negative_coords_callback, slice_wh=64, overlap_wh=0
|
||||
)
|
||||
|
||||
with pytest.warns(SupervisionWarnings, match="outside the slice bounds"):
|
||||
slicer(image)
|
||||
|
||||
|
||||
def test_run_callback_warns_only_once_with_multiple_threads() -> None:
|
||||
"""Test that exactly one warning fires even with thread_workers > 1, validating
|
||||
that the threading.Lock makes the check-and-set atomic."""
|
||||
|
||||
def out_of_bounds_callback(_: np.ndarray) -> Detections:
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 128, 128]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
# 512x512 / 64 slice -> 64 slices; all 4 threads will see out-of-bounds detections
|
||||
image = np.zeros((512, 512, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(
|
||||
callback=out_of_bounds_callback,
|
||||
slice_wh=64,
|
||||
overlap_wh=0,
|
||||
thread_workers=4,
|
||||
)
|
||||
|
||||
with warnings.catch_warnings(record=True) as recorded_warnings:
|
||||
warnings.simplefilter("always")
|
||||
slicer(image)
|
||||
|
||||
out_of_bounds_warnings = [
|
||||
w
|
||||
for w in recorded_warnings
|
||||
if issubclass(w.category, SupervisionWarnings)
|
||||
and "outside the slice bounds" in str(w.message)
|
||||
]
|
||||
assert len(out_of_bounds_warnings) == 1
|
||||
|
||||
|
||||
def test_run_callback_no_warning_for_detection_exactly_at_slice_boundary() -> None:
|
||||
"""Test that a detection whose coordinates exactly equal the slice dimensions
|
||||
does not trigger the warning (boundary is exclusive: > not >=)."""
|
||||
|
||||
def at_boundary_callback(_: np.ndarray) -> Detections:
|
||||
# x2=64, y2=64 on a 64x64 slice — touching the edge but not exceeding it
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 64, 64]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((128, 128, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(callback=at_boundary_callback, slice_wh=64, overlap_wh=0)
|
||||
|
||||
with warnings.catch_warnings(record=True) as recorded_warnings:
|
||||
warnings.simplefilter("always")
|
||||
slicer(image)
|
||||
|
||||
out_of_bounds_warnings = [
|
||||
w
|
||||
for w in recorded_warnings
|
||||
if issubclass(w.category, SupervisionWarnings)
|
||||
and "outside the slice bounds" in str(w.message)
|
||||
]
|
||||
assert len(out_of_bounds_warnings) == 0
|
||||
|
||||
|
||||
def test_run_callback_does_not_rewarn_on_second_call() -> None:
|
||||
"""Test that a second call to the same slicer instance does not re-emit
|
||||
the out-of-bounds warning even when detections are still out of bounds."""
|
||||
|
||||
def out_of_bounds_callback(_: np.ndarray) -> Detections:
|
||||
return Detections(
|
||||
xyxy=np.array([[0, 0, 128, 128]], dtype=float),
|
||||
confidence=np.array([0.9]),
|
||||
class_id=np.array([0]),
|
||||
)
|
||||
|
||||
image = np.zeros((128, 128, 3), dtype=np.uint8)
|
||||
slicer = InferenceSlicer(callback=out_of_bounds_callback, slice_wh=64, overlap_wh=0)
|
||||
|
||||
with warnings.catch_warnings(record=True) as recorded_warnings:
|
||||
warnings.simplefilter("always")
|
||||
slicer(image) # first call — warning fires
|
||||
slicer(image) # second call — must not re-warn
|
||||
|
||||
out_of_bounds_warnings = [
|
||||
w
|
||||
for w in recorded_warnings
|
||||
if issubclass(w.category, SupervisionWarnings)
|
||||
and "outside the slice bounds" in str(w.message)
|
||||
]
|
||||
assert len(out_of_bounds_warnings) == 1
|
||||
|
|
|
|||
Loading…
Reference in New Issue