feat(compact_mask): add `resize()` method and benchmark stage (#2227)
* feat: add resize() method and benchmark stage * perf: optimise resize() — vectorised coords, L3 direct RLE * refactor: split _rle_resize and extract _resize_crop * test: expand resize() tests for scaling and edge cases * refactor: switch resize helpers to F-order (column-major) RLE * fix: merge True/True RLE junctions in _rle_join_cols * perf: vectorize _rle_scale_col RLE re-encoding * test: add density-dispatch and parallel-path resize tests * refactor: harden resize() threading and RLE invariants * fix: accurate resize timing and exact nearest-neighbour parity * type: add explicit numpy typing to ndarray declarations --------- Co-authored-by: Claude Code <noreply@anthropic.com> Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Jirka Borovec
GitHub
parent
de87030a21
commit
153cde5854
|
|
@ -115,6 +115,9 @@ class ScenarioResult:
|
|||
merge_ok: bool | None
|
||||
offset_ok: bool | None
|
||||
centroids_ok: bool | None
|
||||
dense_resize_s: float # nan when dense_skipped
|
||||
compact_resize_s: float
|
||||
resize_ok: bool | None
|
||||
# skip flags
|
||||
dense_skipped: bool = field(default=False)
|
||||
iou_dense_skipped: bool = field(default=False)
|
||||
|
|
@ -327,6 +330,20 @@ def stage_build(
|
|||
return xyxy, masks_dense, class_ids, compact_mask
|
||||
|
||||
|
||||
def _resize_dense_to_shape(masks: np.ndarray, new_h: int, new_w: int) -> np.ndarray:
|
||||
"""Nearest-neighbour resize of (N, H, W) bool masks to (N, new_h, new_w).
|
||||
|
||||
Uses floor-division indexing (``arange * src // dst``) to match the
|
||||
strategy in ``_rle_resize``, ensuring pixel-exact parity for correctness
|
||||
comparisons in :func:`stage_resize`.
|
||||
"""
|
||||
orig_h, orig_w = masks.shape[1], masks.shape[2]
|
||||
x = np.arange(new_w) * orig_w // new_w
|
||||
y = np.arange(new_h) * orig_h // new_h
|
||||
xv, yv = np.meshgrid(x, y)
|
||||
return masks[:, yv, xv]
|
||||
|
||||
|
||||
def stage_encode(
|
||||
masks_dense: np.ndarray,
|
||||
xyxy: np.ndarray,
|
||||
|
|
@ -563,6 +580,44 @@ def stage_centroids(
|
|||
return dense_centroids_s, compact_centroids_s, centroids_ok
|
||||
|
||||
|
||||
def stage_resize(
|
||||
masks_dense: np.ndarray,
|
||||
compact_mask: CompactMask,
|
||||
image_height: int,
|
||||
image_width: int,
|
||||
dense_skipped: bool,
|
||||
) -> tuple[float, float, bool | None]:
|
||||
"""Time resize to half resolution; check pixel-level correctness.
|
||||
|
||||
Dense path uses numpy fancy-indexing via ``_resize_dense_to_shape``.
|
||||
Compact path times ``CompactMask.resize()``, which uses direct RLE
|
||||
arithmetic for sparse masks (below ``_L3_DENSITY_THRESHOLD``) and
|
||||
falls back to ``cv2.INTER_NEAREST`` decode/resize/re-encode for dense
|
||||
masks. The two nearest-neighbour strategies can differ by 1 px at
|
||||
bbox boundaries, so correctness is checked with 1-pixel tolerance.
|
||||
"""
|
||||
new_h, new_w = image_height // 2, image_width // 2
|
||||
new_shape = (new_h, new_w)
|
||||
|
||||
# Use parallel=1 to avoid nested ThreadPoolExecutor contention:
|
||||
# CompactMask.resize() itself spawns a thread pool for N >= _PARALLEL_THRESHOLD,
|
||||
# and time_reps' own parallel outer loop would cause oversubscription.
|
||||
compact_resize_s = time_reps(lambda: compact_mask.resize(new_shape), parallel=1)
|
||||
if dense_skipped:
|
||||
return math.nan, compact_resize_s, None
|
||||
|
||||
resized_dense = _resize_dense_to_shape(masks_dense, new_h, new_w)
|
||||
resized_compact = compact_mask.resize(new_shape).to_dense()
|
||||
resize_ok = bool(
|
||||
np.abs(resized_dense.astype(np.int8) - resized_compact.astype(np.int8)).max()
|
||||
<= 1
|
||||
)
|
||||
dense_resize_s = time_reps(
|
||||
lambda: _resize_dense_to_shape(masks_dense, new_h, new_w)
|
||||
)
|
||||
return dense_resize_s, compact_resize_s, resize_ok
|
||||
|
||||
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
# Scenario runner — orchestrates stages
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
|
@ -681,6 +736,9 @@ def run_scenario(
|
|||
dense_centroids_s, compact_centroids_s, centroids_ok = stage_centroids(
|
||||
masks_dense, compact_mask, dense_skipped
|
||||
)
|
||||
dense_resize_s, compact_resize_s, resize_ok = stage_resize(
|
||||
masks_dense, compact_mask, image_height, image_width, dense_skipped
|
||||
)
|
||||
|
||||
def _timing_line(label: str, dense_s: float, compact_s: float) -> str:
|
||||
compact_ms = f"{compact_s * 1e3:.2f} ms"
|
||||
|
|
@ -704,6 +762,7 @@ def run_scenario(
|
|||
console.print(_timing_line("merge ", dense_merge_s, compact_merge_s))
|
||||
console.print(_timing_line("nms ", dense_nms_s, compact_nms_s))
|
||||
console.print(_timing_line("offset ", dense_offset_s, compact_offset_s))
|
||||
console.print(_timing_line("resize ", dense_resize_s, compact_resize_s))
|
||||
|
||||
checks = {
|
||||
"pixel-perfect": pixel_perfect,
|
||||
|
|
@ -714,6 +773,7 @@ def run_scenario(
|
|||
"merge": merge_ok,
|
||||
"offset": offset_ok,
|
||||
"centroids": centroids_ok,
|
||||
"resize": resize_ok,
|
||||
}
|
||||
parts = []
|
||||
for k, v in checks.items():
|
||||
|
|
@ -777,6 +837,9 @@ def run_scenario(
|
|||
merge_ok=merge_ok,
|
||||
offset_ok=offset_ok,
|
||||
centroids_ok=centroids_ok,
|
||||
dense_resize_s=dense_resize_s,
|
||||
compact_resize_s=compact_resize_s,
|
||||
resize_ok=resize_ok,
|
||||
dense_skipped=dense_skipped,
|
||||
iou_dense_skipped=iou_dense_skipped,
|
||||
)
|
||||
|
|
@ -803,7 +866,17 @@ def _time_compact_annotate(scene: np.ndarray, det_compact: sv.Detections) -> flo
|
|||
# Rich summary table
|
||||
# ══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
_OPS = ("area", "filter", "annot", "iou", "nms", "merge", "offset", "centroids")
|
||||
_OPS = (
|
||||
"area",
|
||||
"filter",
|
||||
"annot",
|
||||
"iou",
|
||||
"nms",
|
||||
"merge",
|
||||
"offset",
|
||||
"centroids",
|
||||
"resize",
|
||||
)
|
||||
|
||||
|
||||
def _build_summary_df(results: list[ScenarioResult]) -> pd.DataFrame:
|
||||
|
|
@ -835,6 +908,7 @@ def _build_summary_df(results: list[ScenarioResult]) -> pd.DataFrame:
|
|||
"merge_ok",
|
||||
"offset_ok",
|
||||
"centroids_ok",
|
||||
"resize_ok",
|
||||
]
|
||||
df["ok"] = df.apply(
|
||||
lambda row: (
|
||||
|
|
@ -899,6 +973,7 @@ def print_summary(results: list[ScenarioResult]) -> None:
|
|||
table.add_column("NMS\nop.", justify="right", min_width=6)
|
||||
table.add_column("Merge\nop.", justify="right", min_width=6)
|
||||
table.add_column("Offset\nop.", justify="right", min_width=6)
|
||||
table.add_column("Resize\nop.", justify="right", min_width=6)
|
||||
table.add_column("Centr\nop.", justify="right", min_width=6)
|
||||
table.add_column("OK?", justify="center", min_width=4)
|
||||
|
||||
|
|
@ -940,6 +1015,7 @@ def print_summary(results: list[ScenarioResult]) -> None:
|
|||
_fmt_speedup(row["dense_nms_s"], row["compact_nms_s"]),
|
||||
_fmt_speedup(row["dense_merge_s"], row["compact_merge_s"]),
|
||||
_fmt_speedup(row["dense_offset_s"], row["compact_offset_s"]),
|
||||
_fmt_speedup(row["dense_resize_s"], row["compact_resize_s"]),
|
||||
_fmt_speedup(row["dense_centroids_s"], row["compact_centroids_s"]),
|
||||
ok_cell,
|
||||
)
|
||||
|
|
@ -967,6 +1043,7 @@ def print_summary(results: list[ScenarioResult]) -> None:
|
|||
"NMS x — mask_non_max_suppression speedup",
|
||||
"Merge x — Detections.merge speedup",
|
||||
"Offset x — move_masks vs with_offset speedup",
|
||||
"Resize x — resize-to-half speedup",
|
||||
"Centroids x — calculate_masks_centroids speedup",
|
||||
"dim ms — dense skipped, compact absolute time shown",
|
||||
]
|
||||
|
|
@ -1020,6 +1097,7 @@ def save_results_csv(results: list[ScenarioResult], path: Path) -> None:
|
|||
"encode_ms_per_mask": (df["encode_s"] * 1e3).round(4),
|
||||
"decode_ms_per_mask": (df["decode_s"] * 1e3).round(4),
|
||||
**{f"{op}_speedup": df[f"{op}_speedup"].round(2) for op in _OPS},
|
||||
"resize_ok": df["resize_ok"],
|
||||
"ok": df["ok"],
|
||||
}
|
||||
).to_csv(path, index=False)
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ crop boundaries, so no extra metadata is required from the caller.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
from typing import Any
|
||||
|
||||
|
|
@ -45,6 +46,346 @@ def _rle_area(rle: npt.NDArray[np.int32]) -> int:
|
|||
return int(np.sum(rle[1::2]))
|
||||
|
||||
|
||||
def _rle_split_cols(
|
||||
rle: npt.NDArray[np.int32],
|
||||
crop_h: int,
|
||||
crop_w: int,
|
||||
) -> list[list[int]]:
|
||||
"""Split a flat F-order RLE into per-column run lists.
|
||||
|
||||
With F-order (column-major) RLE the flat pixel sequence visits all rows
|
||||
of column 0, then all rows of column 1, etc. Each column therefore
|
||||
contains ``crop_h`` contiguous pixels.
|
||||
|
||||
Runs that cross column boundaries are split at the boundary. Each
|
||||
returned list starts with a ``False``-run count (possibly 0), matching
|
||||
the convention of :func:`_mask_to_rle_counts`.
|
||||
|
||||
Args:
|
||||
rle: int32 run-length array as produced by
|
||||
:func:`~supervision.detection.utils.converters._mask_to_rle_counts`.
|
||||
crop_h: Number of rows (pixels per column).
|
||||
crop_w: Number of columns.
|
||||
|
||||
Returns:
|
||||
List of ``crop_w`` run lists, one per column. Each list sums to
|
||||
``crop_h``.
|
||||
|
||||
Examples:
|
||||
```pycon
|
||||
>>> import numpy as np
|
||||
>>> from supervision.detection.compact_mask import _rle_split_cols
|
||||
>>> from supervision.detection.utils.converters import _mask_to_rle_counts
|
||||
>>> mask = np.array([[True, False], [True, True]], dtype=bool)
|
||||
>>> rle = _mask_to_rle_counts(mask)
|
||||
>>> rle.tolist()
|
||||
[0, 2, 1, 1]
|
||||
>>> _rle_split_cols(rle, 2, 2)
|
||||
[[0, 2], [1, 1]]
|
||||
|
||||
```
|
||||
"""
|
||||
per_col: list[list[int]] = [[] for _ in range(crop_w)]
|
||||
col = 0
|
||||
row = 0
|
||||
|
||||
for run_idx, run_len in enumerate(rle):
|
||||
is_true = run_idx % 2 == 1
|
||||
remaining = int(run_len)
|
||||
while remaining > 0:
|
||||
space_in_col = crop_h - row
|
||||
take = min(remaining, space_in_col)
|
||||
if len(per_col[col]) == 0:
|
||||
if is_true:
|
||||
per_col[col].append(0) # leading False count = 0
|
||||
# Check if last run has same parity (True/False) as current chunk.
|
||||
# Last element's parity: index (len-1) odd → True, even → False.
|
||||
elif is_true == ((len(per_col[col]) - 1) % 2 == 1):
|
||||
per_col[col][-1] += take
|
||||
remaining -= take
|
||||
row += take
|
||||
if row >= crop_h:
|
||||
row = 0
|
||||
col += 1
|
||||
continue
|
||||
per_col[col].append(take)
|
||||
remaining -= take
|
||||
row += take
|
||||
if row >= crop_h:
|
||||
row = 0
|
||||
col += 1
|
||||
if col >= crop_w:
|
||||
break
|
||||
|
||||
# Fill any empty columns (all-False).
|
||||
for c in range(crop_w):
|
||||
if not per_col[c]:
|
||||
per_col[c] = [crop_h]
|
||||
|
||||
return per_col
|
||||
|
||||
|
||||
def _rle_scale_col(
|
||||
col_runs: list[int],
|
||||
src_h: int,
|
||||
row_map: npt.NDArray[np.int32],
|
||||
) -> list[int]:
|
||||
"""Scale one column's run list to a new height using a precomputed row map.
|
||||
|
||||
Each output row is mapped to a source row via ``row_map``, which
|
||||
implements nearest-neighbour resampling in the vertical direction.
|
||||
|
||||
Args:
|
||||
col_runs: Per-column run list starting with a ``False``-run count.
|
||||
src_h: Height of the source column (sum of ``col_runs``).
|
||||
row_map: int32 array of length ``new_crop_h``; ``row_map[r']`` is the
|
||||
source row index for output row ``r'``. Use
|
||||
``(np.arange(new_crop_h) * src_h // new_crop_h)`` for
|
||||
``cv2.INTER_NEAREST``-compatible mapping.
|
||||
|
||||
Returns:
|
||||
Scaled run list of total length ``len(row_map)``, always starting
|
||||
with a ``False``-run count.
|
||||
|
||||
Examples:
|
||||
```pycon
|
||||
>>> import numpy as np
|
||||
>>> from supervision.detection.compact_mask import _rle_scale_col
|
||||
>>> col_runs = [0, 2, 2] # F=0, T=2, F=2 → [T, T, F, F]
|
||||
>>> row_map = np.array([0, 1, 2, 3, 0, 1, 2, 3], dtype=np.int32)
|
||||
>>> _rle_scale_col(col_runs, 4, row_map)
|
||||
[0, 2, 2, 2, 2]
|
||||
|
||||
```
|
||||
"""
|
||||
new_crop_h = len(row_map)
|
||||
if new_crop_h == 0:
|
||||
return [0]
|
||||
|
||||
# Reconstruct per-source-row boolean values from run list.
|
||||
src_values: npt.NDArray[np.bool_] = np.empty(src_h, dtype=np.bool_)
|
||||
pos = 0
|
||||
for ri, rl in enumerate(col_runs):
|
||||
src_values[pos : pos + rl] = ri % 2 == 1 # odd index → True
|
||||
pos += rl
|
||||
if pos < src_h:
|
||||
src_values[pos:] = False # pad truncated RLE
|
||||
|
||||
# Map output rows to source values.
|
||||
out_values = src_values[row_map]
|
||||
|
||||
# RLE-encode the output column; vectorised via np.diff on bool view.
|
||||
out_uint8 = out_values.view(np.uint8)
|
||||
boundaries = np.flatnonzero(np.diff(out_uint8))
|
||||
run_starts: npt.NDArray[np.int64] = np.empty(len(boundaries) + 1, dtype=np.int64)
|
||||
run_ends: npt.NDArray[np.int64] = np.empty(len(boundaries) + 1, dtype=np.int64)
|
||||
run_starts[0] = 0
|
||||
run_starts[1:] = boundaries + 1
|
||||
run_ends[:-1] = boundaries + 1
|
||||
run_ends[-1] = new_crop_h
|
||||
result_runs: list[int] = (run_ends - run_starts).tolist()
|
||||
# RLE starts with a False count; prepend 0 if output begins with True.
|
||||
if bool(out_values[0]):
|
||||
result_runs.insert(0, 0)
|
||||
return result_runs
|
||||
|
||||
|
||||
def _rle_join_cols(
|
||||
scaled_cols: list[list[int]],
|
||||
new_total: int,
|
||||
) -> npt.NDArray[np.int32]:
|
||||
"""Concatenate per-column run lists into a flat RLE, merging junctions.
|
||||
|
||||
Each column run list starts with a ``False``-run count. Two junction types
|
||||
can be merged across column boundaries:
|
||||
|
||||
* ``False``/``False``: the trailing False run merges with the leading False
|
||||
run of the next column (leading count may be zero).
|
||||
* ``True``/``True``: when the accumulated output ends on a True run and the
|
||||
next column's leading False count is zero (column starts with True), the
|
||||
two True runs are merged to avoid inserting a zero-length False run that
|
||||
would inflate ``len(rle)`` and skew the density metric in
|
||||
:func:`_resize_crop`.
|
||||
|
||||
Args:
|
||||
scaled_cols: List of per-column run lists, each starting with a
|
||||
``False``-run count.
|
||||
new_total: Total pixel count of the output (fallback for empty input).
|
||||
|
||||
Returns:
|
||||
Flat int32 RLE array starting with a ``False``-run count.
|
||||
|
||||
Examples:
|
||||
```pycon
|
||||
>>> import numpy as np
|
||||
>>> from supervision.detection.compact_mask import _rle_join_cols
|
||||
>>> cols = [[1, 2], [1, 2]] # each col: F=1, T=2
|
||||
>>> _rle_join_cols(cols, 6).tolist()
|
||||
[1, 2, 1, 2]
|
||||
|
||||
```
|
||||
"""
|
||||
output_runs: list[int] = []
|
||||
for col_runs in scaled_cols:
|
||||
if not output_runs:
|
||||
output_runs.extend(col_runs)
|
||||
else:
|
||||
last_is_true = (len(output_runs) - 1) % 2 == 1
|
||||
# col_runs always starts with a False count → first_is_true=False
|
||||
if not last_is_true: # last == False == first → merge
|
||||
output_runs[-1] += col_runs[0]
|
||||
output_runs.extend(col_runs[1:])
|
||||
elif col_runs[0] == 0 and len(col_runs) > 1:
|
||||
# last run = True; column also starts True (leading False = 0)
|
||||
# → merge to avoid a zero-length False run at the junction.
|
||||
output_runs[-1] += col_runs[1]
|
||||
output_runs.extend(col_runs[2:])
|
||||
else:
|
||||
output_runs.extend(col_runs)
|
||||
|
||||
return np.array(output_runs if output_runs else [new_total], dtype=np.int32)
|
||||
|
||||
|
||||
def _rle_resize(
|
||||
rle: npt.NDArray[np.int32],
|
||||
crop_h: int,
|
||||
crop_w: int,
|
||||
new_crop_h: int,
|
||||
new_crop_w: int,
|
||||
) -> npt.NDArray[np.int32]:
|
||||
"""Resize an F-order RLE-encoded crop via nearest-neighbour resampling.
|
||||
|
||||
Manipulates run lengths directly without decoding to a full 2D boolean
|
||||
array. Delegates to :func:`_rle_split_cols`, :func:`_rle_scale_col`,
|
||||
and :func:`_rle_join_cols`.
|
||||
|
||||
The nearest-neighbour mapping ``src = floor(dst * src_size / dst_size)``
|
||||
is bit-exact with ``cv2.INTER_NEAREST``.
|
||||
|
||||
Args:
|
||||
rle: int32 array of F-order run lengths as produced by
|
||||
:func:`~supervision.detection.utils.converters._mask_to_rle_counts`.
|
||||
Starts with a ``False``-run count (may be 0).
|
||||
crop_h: Height of the original crop.
|
||||
crop_w: Width of the original crop.
|
||||
new_crop_h: Height of the resized crop.
|
||||
new_crop_w: Width of the resized crop.
|
||||
|
||||
Returns:
|
||||
int32 array of F-order run lengths for the resized crop, starting
|
||||
with the ``False``-run count.
|
||||
|
||||
Examples:
|
||||
Upscale a 3x3 mask with a diagonal True stripe to 6x6:
|
||||
|
||||
```pycon
|
||||
>>> import numpy as np
|
||||
>>> from supervision.detection.compact_mask import _rle_resize
|
||||
>>> from supervision.detection.utils.converters import (
|
||||
... _mask_to_rle_counts, _rle_counts_to_mask,
|
||||
... )
|
||||
>>> mask = np.array([
|
||||
... [True, False, False],
|
||||
... [False, True, False],
|
||||
... [False, False, True ],
|
||||
... ], dtype=bool)
|
||||
>>> rle = _mask_to_rle_counts(mask)
|
||||
>>> resized_rle = _rle_resize(rle, 3, 3, 6, 6)
|
||||
>>> result = _rle_counts_to_mask(resized_rle, 6, 6)
|
||||
>>> result.astype(int)
|
||||
array([[1, 1, 0, 0, 0, 0],
|
||||
[1, 1, 0, 0, 0, 0],
|
||||
[0, 0, 1, 1, 0, 0],
|
||||
[0, 0, 1, 1, 0, 0],
|
||||
[0, 0, 0, 0, 1, 1],
|
||||
[0, 0, 0, 0, 1, 1]])
|
||||
|
||||
```
|
||||
"""
|
||||
new_total = new_crop_h * new_crop_w
|
||||
|
||||
if crop_h * crop_w == 0 or new_total == 0:
|
||||
return np.array([0], dtype=np.int32)
|
||||
if len(rle) == 1 or int(np.sum(rle[1::2])) == 0:
|
||||
return np.array([new_total], dtype=np.int32)
|
||||
if len(rle) == 2 and rle[0] == 0:
|
||||
return np.array([0, new_total], dtype=np.int32)
|
||||
|
||||
per_col = _rle_split_cols(rle, crop_h, crop_w)
|
||||
|
||||
# cv2.INTER_NEAREST column mapping: src = floor(dst * src_w / dst_w)
|
||||
col_map = (np.arange(new_crop_w) * crop_w // new_crop_w).astype(np.int32)
|
||||
|
||||
# cv2.INTER_NEAREST row mapping: src = floor(dst * src_h / dst_h)
|
||||
row_map = (np.arange(new_crop_h) * crop_h // new_crop_h).astype(np.int32)
|
||||
|
||||
# Scale each unique source column once; reuse via cache for repeated cols.
|
||||
col_cache: dict[int, list[int]] = {}
|
||||
scaled_cols = []
|
||||
for src_c in col_map:
|
||||
if src_c not in col_cache:
|
||||
col_cache[src_c] = _rle_scale_col(per_col[src_c], crop_h, row_map)
|
||||
scaled_cols.append(col_cache[src_c])
|
||||
|
||||
return _rle_join_cols(scaled_cols, new_total)
|
||||
|
||||
|
||||
# Fraction of (run_count / pixel_count) below which _rle_resize is used
|
||||
# instead of the decode → cv2 → re-encode path. Sparse masks have few long
|
||||
# runs; dense/complex masks approach 1 run per 2 pixels.
|
||||
_L3_DENSITY_THRESHOLD: float = 0.25
|
||||
# Thread overhead outweighs gains below this mask count.
|
||||
_PARALLEL_THRESHOLD: int = 8
|
||||
|
||||
|
||||
def _resize_crop(
|
||||
rle: npt.NDArray[np.int32],
|
||||
orig_h: int,
|
||||
orig_w: int,
|
||||
new_h: int,
|
||||
new_w: int,
|
||||
) -> npt.NDArray[np.int32]:
|
||||
"""Resize one RLE crop to ``(new_h, new_w)``, choosing the fastest path.
|
||||
|
||||
Dispatch order:
|
||||
|
||||
1. **All-False fast path** — returns a single False run; no decode.
|
||||
2. **L3 direct RLE path** — used when run density is below
|
||||
:data:`_L3_DENSITY_THRESHOLD`; manipulates run lengths without
|
||||
allocating a 2D array.
|
||||
3. **cv2 fallback** — decodes to ``uint8``, calls
|
||||
``cv2.resize(INTER_NEAREST)``, re-encodes; used for dense masks.
|
||||
|
||||
Args:
|
||||
rle: int32 run-length array for the source crop.
|
||||
orig_h: Height of the source crop.
|
||||
orig_w: Width of the source crop.
|
||||
new_h: Target height.
|
||||
new_w: Target width.
|
||||
|
||||
Returns:
|
||||
int32 RLE array for the resized crop.
|
||||
"""
|
||||
import cv2
|
||||
|
||||
# All-False: skip decode entirely.
|
||||
if _rle_area(rle) == 0:
|
||||
return np.array([new_h * new_w], dtype=np.int32)
|
||||
|
||||
# L3: direct RLE arithmetic for sparse masks.
|
||||
if len(rle) / max(1, orig_h * orig_w) < _L3_DENSITY_THRESHOLD:
|
||||
return _rle_resize(rle, orig_h, orig_w, new_h, new_w)
|
||||
|
||||
# cv2 fallback for dense masks.
|
||||
crop = _rle_counts_to_mask(rle, orig_h, orig_w)
|
||||
resized = cv2.resize(
|
||||
crop.view(np.uint8),
|
||||
(new_w, new_h),
|
||||
interpolation=cv2.INTER_NEAREST,
|
||||
).astype(bool)
|
||||
return _mask_to_rle_counts(resized)
|
||||
|
||||
|
||||
class CompactMask:
|
||||
"""Memory-efficient crop-RLE mask storage for instance segmentation.
|
||||
|
||||
|
|
@ -843,3 +1184,123 @@ class CompactMask:
|
|||
np.array(out_offsets_list, dtype=np.int32),
|
||||
new_image_shape,
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Resize
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def resize(self, new_image_shape: tuple[int, int]) -> CompactMask:
|
||||
"""Return a new CompactMask scaled to a different image resolution.
|
||||
|
||||
Each crop mask is resized with nearest-neighbour interpolation.
|
||||
Sparse masks use direct RLE arithmetic (:func:`_rle_resize`); dense
|
||||
masks fall back to ``cv2.resize(INTER_NEAREST)``. Offsets and crop
|
||||
dimensions are scaled proportionally to the new image size.
|
||||
|
||||
Performance notes:
|
||||
|
||||
* Coordinate arithmetic is fully vectorised (no Python loop over N).
|
||||
* All-``False`` crops skip decode/resize entirely.
|
||||
* For N >= 8, resize runs in a thread pool — NumPy and OpenCV
|
||||
release the GIL so crops execute in parallel on multi-core CPUs.
|
||||
|
||||
Args:
|
||||
new_image_shape: ``(H, W)`` of the target image.
|
||||
|
||||
Returns:
|
||||
New :class:`CompactMask` with updated ``image_shape``, scaled
|
||||
offsets, scaled crop shapes, and re-encoded RLE crops.
|
||||
|
||||
Raises:
|
||||
ValueError: If any dimension in *new_image_shape* is ``<= 0``.
|
||||
|
||||
Examples:
|
||||
```pycon
|
||||
>>> import numpy as np
|
||||
>>> from supervision.detection.compact_mask import CompactMask
|
||||
>>> masks = np.zeros((1, 100, 100), dtype=bool)
|
||||
>>> masks[0, 20:40, 30:60] = True
|
||||
>>> xyxy = np.array([[30, 20, 59, 39]], dtype=np.float32)
|
||||
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(100, 100))
|
||||
>>> small = cm.resize((50, 50))
|
||||
>>> small.shape
|
||||
(1, 50, 50)
|
||||
>>> small.offsets[0].tolist()
|
||||
[15, 10]
|
||||
|
||||
```
|
||||
"""
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
new_h, new_w = new_image_shape
|
||||
if new_h <= 0 or new_w <= 0:
|
||||
raise ValueError("new_image_shape must contain positive dimensions")
|
||||
|
||||
# fast path — identity resize; list() creates a new container but the
|
||||
# individual RLE numpy arrays are shared (shallow copy). Callers must
|
||||
# not mutate returned RLE arrays in-place.
|
||||
if (new_h, new_w) == self._image_shape:
|
||||
return CompactMask(
|
||||
list(self._rles),
|
||||
self._crop_shapes.copy(),
|
||||
self._offsets.copy(),
|
||||
new_image_shape,
|
||||
)
|
||||
|
||||
# empty guard
|
||||
if len(self) == 0:
|
||||
return CompactMask(
|
||||
[],
|
||||
np.empty((0, 2), dtype=np.int32),
|
||||
np.empty((0, 2), dtype=np.int32),
|
||||
new_image_shape,
|
||||
)
|
||||
|
||||
img_h, img_w = self._image_shape
|
||||
sx = new_w / img_w
|
||||
sy = new_h / img_h
|
||||
|
||||
# L1 — vectorised coordinate arithmetic; no Python loop over N masks.
|
||||
x1s = self._offsets[:, 0].astype(np.float64)
|
||||
y1s = self._offsets[:, 1].astype(np.float64)
|
||||
x2s = x1s + self._crop_shapes[:, 1] - 1 # inclusive right edge
|
||||
y2s = y1s + self._crop_shapes[:, 0] - 1 # inclusive bottom edge
|
||||
|
||||
new_x1s = np.clip(np.round(x1s * sx), 0, new_w - 1).astype(np.int32)
|
||||
new_y1s = np.clip(np.round(y1s * sy), 0, new_h - 1).astype(np.int32)
|
||||
new_x2s = np.clip(np.round(x2s * sx), 0, new_w - 1).astype(np.int32)
|
||||
new_y2s = np.clip(np.round(y2s * sy), 0, new_h - 1).astype(np.int32)
|
||||
new_crop_ws: npt.NDArray[np.int32] = np.maximum(
|
||||
1, new_x2s - new_x1s + 1
|
||||
).astype(np.int32)
|
||||
new_crop_hs: npt.NDArray[np.int32] = np.maximum(
|
||||
1, new_y2s - new_y1s + 1
|
||||
).astype(np.int32)
|
||||
|
||||
# L2b — parallel per-crop resize; NumPy and OpenCV release the GIL.
|
||||
orig_crop_hs = self._crop_shapes[:, 0]
|
||||
orig_crop_ws = self._crop_shapes[:, 1]
|
||||
|
||||
args = [
|
||||
(
|
||||
self._rles[i],
|
||||
int(orig_crop_hs[i]),
|
||||
int(orig_crop_ws[i]),
|
||||
int(new_crop_hs[i]),
|
||||
int(new_crop_ws[i]),
|
||||
)
|
||||
for i in range(len(self))
|
||||
]
|
||||
|
||||
n = len(self)
|
||||
if n >= _PARALLEL_THRESHOLD:
|
||||
with ThreadPoolExecutor(max_workers=min(n, os.cpu_count() or 4)) as pool:
|
||||
new_rles: list[npt.NDArray[np.int32]] = list(
|
||||
pool.map(lambda a: _resize_crop(*a), args)
|
||||
)
|
||||
else:
|
||||
new_rles = [_resize_crop(*a) for a in args]
|
||||
|
||||
new_crop_shapes = np.column_stack((new_crop_hs, new_crop_ws)).astype(np.int32)
|
||||
new_offsets = np.column_stack((new_x1s, new_y1s)).astype(np.int32)
|
||||
return CompactMask(new_rles, new_crop_shapes, new_offsets, new_image_shape)
|
||||
|
|
|
|||
|
|
@ -977,3 +977,615 @@ class TestCompactMaskWithOffsetRandom:
|
|||
expected,
|
||||
err_msg=f"Larger canvas offset mismatch for seed={seed}",
|
||||
)
|
||||
|
||||
|
||||
class TestRleSplitCols:
|
||||
"""Tests for _rle_split_cols: splitting F-order RLE into per-column lists."""
|
||||
|
||||
def test_all_true_2x2(self) -> None:
|
||||
"""All-True 2x2 splits into two columns each [0, 2]."""
|
||||
from supervision.detection.compact_mask import _rle_split_cols
|
||||
|
||||
mask = np.ones((2, 2), dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result = _rle_split_cols(rle, 2, 2)
|
||||
assert result == [[0, 2], [0, 2]]
|
||||
|
||||
def test_all_false_3x3(self) -> None:
|
||||
"""All-False 3x3 splits into three columns each [3]."""
|
||||
from supervision.detection.compact_mask import _rle_split_cols
|
||||
|
||||
mask = np.zeros((3, 3), dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result = _rle_split_cols(rle, 3, 3)
|
||||
assert result == [[3], [3], [3]]
|
||||
|
||||
def test_mixed_2x2(self) -> None:
|
||||
"""Mixed mask splits correctly per column."""
|
||||
from supervision.detection.compact_mask import _rle_split_cols
|
||||
|
||||
mask = np.array([[True, False], [True, True]], dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result = _rle_split_cols(rle, 2, 2)
|
||||
assert result == [[0, 2], [1, 1]]
|
||||
|
||||
@pytest.mark.parametrize("seed", list(range(20)))
|
||||
def test_round_trip_random(self, seed: int) -> None:
|
||||
"""Split then rejoin must reconstruct original mask for random inputs."""
|
||||
from supervision.detection.compact_mask import (
|
||||
_rle_join_cols,
|
||||
_rle_split_cols,
|
||||
)
|
||||
|
||||
rng = np.random.default_rng(seed + 8000)
|
||||
crop_h = int(rng.integers(1, 30))
|
||||
crop_w = int(rng.integers(1, 30))
|
||||
mask = rng.random((crop_h, crop_w)) < 0.4
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
per_col = _rle_split_cols(rle, crop_h, crop_w)
|
||||
|
||||
assert len(per_col) == crop_w
|
||||
for c in range(crop_w):
|
||||
assert sum(per_col[c]) == crop_h, f"col {c} sum mismatch"
|
||||
|
||||
# Rejoin and verify pixel equality.
|
||||
rejoined = _rle_join_cols(per_col, crop_h * crop_w)
|
||||
decoded = _rle_counts_to_mask(rejoined, crop_h, crop_w)
|
||||
np.testing.assert_array_equal(
|
||||
decoded,
|
||||
mask,
|
||||
err_msg=f"Split→join round-trip failed for seed={seed}",
|
||||
)
|
||||
|
||||
def test_join_true_true_junction_no_zero_run(self) -> None:
|
||||
"""_rle_join_cols merges True/True boundary; no zero-length False run inserted.
|
||||
|
||||
When column A ends True and column B starts True (leading False count = 0),
|
||||
the junction must produce a single merged True run, not a zero-length False
|
||||
run between two True runs. A zero-length run would inflate len(rle) and
|
||||
misroute density-based dispatch in _resize_crop.
|
||||
"""
|
||||
from supervision.detection.compact_mask import _rle_join_cols
|
||||
|
||||
# col A: [0, 3] → T=3 (height=3, all True)
|
||||
# col B: [0, 3] → T=3 (height=3, all True)
|
||||
# Merged: should be [0, 6], NOT [0, 3, 0, 3].
|
||||
cols = [[0, 3], [0, 3]]
|
||||
result = _rle_join_cols(cols, 6).tolist()
|
||||
assert result == [0, 6], (
|
||||
f"Expected [0, 6] (merged True runs), got {result}; "
|
||||
"zero-length False run would inflate density metric"
|
||||
)
|
||||
assert 0 not in result[1:], "Zero-length run found after junction merge"
|
||||
|
||||
|
||||
class TestCompactMaskResize:
|
||||
"""Tests for CompactMask.resize method.
|
||||
|
||||
Verifies scaling behaviour, coordinate arithmetic, identity resize,
|
||||
empty collections, invalid dimensions, and dense parity with cv2.
|
||||
"""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_shape", "mask_slice", "target_shape", "description"),
|
||||
[
|
||||
(
|
||||
(10, 10),
|
||||
(slice(2, 5), slice(2, 5)),
|
||||
(100, 100),
|
||||
"10x upscale 10x10 to 100x100",
|
||||
),
|
||||
(
|
||||
(480, 640),
|
||||
(slice(100, 200), slice(150, 300)),
|
||||
(240, 320),
|
||||
"HD halve 480x640 to 240x320",
|
||||
),
|
||||
(
|
||||
(100, 200),
|
||||
(slice(20, 40), slice(50, 100)),
|
||||
(50, 400),
|
||||
"asymmetric: shrink H, grow W",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_scale_shape_and_offsets(
|
||||
self,
|
||||
src_shape: tuple[int, int],
|
||||
mask_slice: tuple[slice, slice],
|
||||
target_shape: tuple[int, int],
|
||||
description: str,
|
||||
) -> None:
|
||||
"""Resize scales shape and offsets proportionally."""
|
||||
img_h, img_w = src_shape
|
||||
masks = np.zeros((1, img_h, img_w), dtype=bool)
|
||||
masks[0, mask_slice[0], mask_slice[1]] = True
|
||||
xyxy = mask_to_xyxy(masks)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=src_shape)
|
||||
|
||||
resized = cm.resize(target_shape)
|
||||
|
||||
assert resized.shape == (1, target_shape[0], target_shape[1]), description
|
||||
|
||||
sx = target_shape[1] / src_shape[1]
|
||||
sy = target_shape[0] / src_shape[0]
|
||||
orig_offset_x = int(cm.offsets[0, 0])
|
||||
orig_offset_y = int(cm.offsets[0, 1])
|
||||
expected_x = round(orig_offset_x * sx)
|
||||
expected_y = round(orig_offset_y * sy)
|
||||
assert abs(int(resized.offsets[0, 0]) - expected_x) <= 1, description
|
||||
assert abs(int(resized.offsets[0, 1]) - expected_y) <= 1, description
|
||||
|
||||
def test_identity_preserves_rle(self) -> None:
|
||||
"""Resize to same shape returns identical RLE, offsets, and crop shapes."""
|
||||
masks = np.zeros((1, 80, 80), dtype=bool)
|
||||
masks[0, 10:30, 15:45] = True
|
||||
xyxy = mask_to_xyxy(masks)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(80, 80))
|
||||
|
||||
resized = cm.resize((80, 80))
|
||||
|
||||
assert resized.shape == cm.shape
|
||||
np.testing.assert_array_equal(resized.offsets, cm.offsets)
|
||||
np.testing.assert_array_equal(resized._crop_shapes, cm._crop_shapes)
|
||||
for orig_rle, new_rle in zip(cm._rles, resized._rles):
|
||||
np.testing.assert_array_equal(orig_rle, new_rle)
|
||||
|
||||
def test_empty_n0(self) -> None:
|
||||
"""Resize of an empty CompactMask returns empty with new image_shape."""
|
||||
masks = np.zeros((0, 50, 50), dtype=bool)
|
||||
xyxy = np.empty((0, 4), dtype=np.float32)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(50, 50))
|
||||
|
||||
resized = cm.resize((100, 200))
|
||||
|
||||
assert len(resized) == 0
|
||||
assert resized.shape == (0, 100, 200)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"bad_shape",
|
||||
[
|
||||
(0, 50),
|
||||
(-1, 50),
|
||||
(50, 0),
|
||||
(50, -1),
|
||||
],
|
||||
)
|
||||
def test_invalid_dimensions_raises(self, bad_shape: tuple[int, int]) -> None:
|
||||
"""Resize with non-positive dimensions raises ValueError."""
|
||||
masks = np.zeros((1, 50, 50), dtype=bool)
|
||||
masks[0, 10:20, 10:20] = True
|
||||
xyxy = mask_to_xyxy(masks)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(50, 50))
|
||||
|
||||
with pytest.raises(ValueError, match="positive"):
|
||||
cm.resize(bad_shape)
|
||||
|
||||
def test_multi_mask_each_scales_independently(self) -> None:
|
||||
"""N=4 masks at different positions all scale correctly after resize."""
|
||||
img_h, img_w = 100, 100
|
||||
target_h, target_w = 50, 50
|
||||
masks = np.zeros((4, img_h, img_w), dtype=bool)
|
||||
masks[0, 10:20, 10:20] = True
|
||||
masks[1, 30:50, 30:50] = True
|
||||
masks[2, 60:80, 60:80] = True
|
||||
masks[3, 5:10, 80:90] = True
|
||||
xyxy = mask_to_xyxy(masks)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(img_h, img_w))
|
||||
|
||||
resized = cm.resize((target_h, target_w))
|
||||
|
||||
assert resized.shape == (4, target_h, target_w)
|
||||
sx = target_w / img_w
|
||||
sy = target_h / img_h
|
||||
for i in range(4):
|
||||
expected_x = round(int(cm.offsets[i, 0]) * sx)
|
||||
expected_y = round(int(cm.offsets[i, 1]) * sy)
|
||||
assert abs(int(resized.offsets[i, 0]) - expected_x) <= 1, f"mask {i} x"
|
||||
assert abs(int(resized.offsets[i, 1]) - expected_y) <= 1, f"mask {i} y"
|
||||
|
||||
def test_zero_extent_extreme_downscale(self) -> None:
|
||||
"""Extreme downscale that collapses a 1px bbox returns valid 1x1 crop."""
|
||||
masks = np.zeros((1, 1000, 1000), dtype=bool)
|
||||
masks[0, 500, 500] = True
|
||||
xyxy = mask_to_xyxy(masks)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(1000, 1000))
|
||||
|
||||
resized = cm.resize((2, 2))
|
||||
|
||||
assert resized.shape == (1, 2, 2)
|
||||
assert int(resized._crop_shapes[0, 0]) >= 1
|
||||
assert int(resized._crop_shapes[0, 1]) >= 1
|
||||
dense = resized.to_dense()
|
||||
assert dense.shape == (1, 2, 2)
|
||||
|
||||
@pytest.mark.parametrize("seed", list(range(10)))
|
||||
def test_dense_parity_roundtrip(self, seed: int) -> None:
|
||||
"""Resized CompactMask matches OpenCV-resized dense masks within 1px."""
|
||||
import cv2
|
||||
|
||||
rng = np.random.default_rng(seed + 500)
|
||||
img_h, img_w = 80, 120
|
||||
target_h, target_w = 40, 60
|
||||
num_masks = int(rng.integers(1, 5))
|
||||
masks, xyxy = _random_masks_and_xyxy(rng, num_masks, img_h, img_w)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(img_h, img_w))
|
||||
|
||||
resized = cm.resize((target_h, target_w))
|
||||
resized_dense = resized.to_dense()
|
||||
|
||||
for i in range(num_masks):
|
||||
expected = cv2.resize(
|
||||
masks[i].astype(np.uint8),
|
||||
(target_w, target_h),
|
||||
interpolation=cv2.INTER_NEAREST,
|
||||
).astype(bool)
|
||||
actual = resized_dense[i]
|
||||
diff = np.abs(actual.astype(int) - expected.astype(int)).max()
|
||||
assert int(diff) <= 1, (
|
||||
f"Dense parity mismatch for seed={seed}, mask={i}: "
|
||||
f"max pixel diff={diff}"
|
||||
)
|
||||
|
||||
|
||||
class TestRleResize:
|
||||
"""Tests for _rle_resize direct F-order RLE resizing.
|
||||
|
||||
Verifies that _rle_resize produces identical results to the decode ->
|
||||
cv2.resize(INTER_NEAREST) -> encode path for identity, upscale, downscale,
|
||||
non-square, all-False, all-True, single-pixel, and random masks.
|
||||
"""
|
||||
|
||||
def test_identity_4x4(self) -> None:
|
||||
"""Identity resize (same dimensions) preserves the decoded mask."""
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.array(
|
||||
[
|
||||
[False, True, True, False],
|
||||
[True, True, False, False],
|
||||
[False, False, True, True],
|
||||
[True, False, False, True],
|
||||
],
|
||||
dtype=bool,
|
||||
)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, 4, 4, 4, 4)
|
||||
result = _rle_counts_to_mask(result_rle, 4, 4)
|
||||
np.testing.assert_array_equal(result, mask)
|
||||
|
||||
def test_2x_upscale(self) -> None:
|
||||
"""2x upscale of a 2x2 mask doubles each pixel."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.array(
|
||||
[
|
||||
[True, False],
|
||||
[False, True],
|
||||
],
|
||||
dtype=bool,
|
||||
)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, 2, 2, 4, 4)
|
||||
result = _rle_counts_to_mask(result_rle, 4, 4)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8), (4, 4), interpolation=cv2.INTER_NEAREST
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
def test_2x_downscale(self) -> None:
|
||||
"""2x downscale of a 4x4 block mask halves dimensions."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.array(
|
||||
[
|
||||
[True, True, False, False],
|
||||
[True, True, False, False],
|
||||
[False, False, True, True],
|
||||
[False, False, True, True],
|
||||
],
|
||||
dtype=bool,
|
||||
)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, 4, 4, 2, 2)
|
||||
result = _rle_counts_to_mask(result_rle, 2, 2)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8), (2, 2), interpolation=cv2.INTER_NEAREST
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
def test_non_square_scale(self) -> None:
|
||||
"""Non-square resize: 4x6 to 2x3 with independent axis scaling."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.zeros((4, 6), dtype=bool)
|
||||
mask[0:2, 0:3] = True
|
||||
mask[2:4, 3:6] = True
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, 4, 6, 2, 3)
|
||||
result = _rle_counts_to_mask(result_rle, 2, 3)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8), (3, 2), interpolation=cv2.INTER_NEAREST
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_shape", "dst_shape"),
|
||||
[
|
||||
((3, 3), (6, 6)),
|
||||
((5, 5), (2, 2)),
|
||||
((4, 6), (8, 12)),
|
||||
((10, 10), (3, 3)),
|
||||
],
|
||||
)
|
||||
def test_all_false(
|
||||
self, src_shape: tuple[int, int], dst_shape: tuple[int, int]
|
||||
) -> None:
|
||||
"""All-False mask resizes to all-False regardless of dimensions."""
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.zeros(src_shape, dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, *src_shape, *dst_shape)
|
||||
result = _rle_counts_to_mask(result_rle, *dst_shape)
|
||||
assert not result.any()
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_shape", "dst_shape"),
|
||||
[
|
||||
((3, 3), (6, 6)),
|
||||
((5, 5), (2, 2)),
|
||||
((4, 6), (8, 12)),
|
||||
((10, 10), (3, 3)),
|
||||
],
|
||||
)
|
||||
def test_all_true(
|
||||
self, src_shape: tuple[int, int], dst_shape: tuple[int, int]
|
||||
) -> None:
|
||||
"""All-True mask resizes to all-True regardless of dimensions."""
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.ones(src_shape, dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, *src_shape, *dst_shape)
|
||||
result = _rle_counts_to_mask(result_rle, *dst_shape)
|
||||
assert result.all()
|
||||
|
||||
def test_single_pixel_true_upscale(self) -> None:
|
||||
"""Single True pixel in a 3x3 mask upscaled preserves position."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.zeros((3, 3), dtype=bool)
|
||||
mask[1, 1] = True
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, 3, 3, 6, 6)
|
||||
result = _rle_counts_to_mask(result_rle, 6, 6)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8), (6, 6), interpolation=cv2.INTER_NEAREST
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
@pytest.mark.parametrize("seed", list(range(45)))
|
||||
def test_roundtrip_parity_with_cv2(self, seed: int) -> None:
|
||||
"""_rle_resize matches cv2.resize(INTER_NEAREST) within 1-pixel tolerance."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
rng = np.random.default_rng(seed + 7000)
|
||||
crop_h = int(rng.integers(1, 50))
|
||||
crop_w = int(rng.integers(1, 50))
|
||||
new_crop_h = int(rng.integers(1, 100))
|
||||
new_crop_w = int(rng.integers(1, 100))
|
||||
|
||||
mask = rng.random((crop_h, crop_w)) < 0.3
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, crop_h, crop_w, new_crop_h, new_crop_w)
|
||||
result = _rle_counts_to_mask(result_rle, new_crop_h, new_crop_w)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8),
|
||||
(new_crop_w, new_crop_h),
|
||||
interpolation=cv2.INTER_NEAREST,
|
||||
).astype(bool)
|
||||
diff = np.abs(result.astype(int) - expected.astype(int)).max()
|
||||
assert diff <= 1, (
|
||||
f"Parity mismatch >1px for seed={seed}, "
|
||||
f"src=({crop_h},{crop_w}), dst=({new_crop_h},{new_crop_w}): "
|
||||
f"max diff={diff}"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_shape", "dst_shape"),
|
||||
[
|
||||
((1, 10), (1, 5)),
|
||||
((10, 1), (5, 1)),
|
||||
((1, 20), (1, 40)),
|
||||
((20, 1), (40, 1)),
|
||||
],
|
||||
)
|
||||
def test_tall_and_wide_crops(
|
||||
self, src_shape: tuple[int, int], dst_shape: tuple[int, int]
|
||||
) -> None:
|
||||
"""Single-row and single-col crops scale correctly with cv2 parity."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
rng = np.random.default_rng(src_shape[0] * 31 + dst_shape[1] * 17)
|
||||
mask = rng.random(src_shape) < 0.5
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, *src_shape, *dst_shape)
|
||||
result = _rle_counts_to_mask(result_rle, *dst_shape)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8),
|
||||
(dst_shape[1], dst_shape[0]),
|
||||
interpolation=cv2.INTER_NEAREST,
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_shape", "dst_shape"),
|
||||
[
|
||||
((7, 11), (5, 13)),
|
||||
((13, 7), (17, 3)),
|
||||
((3, 5), (11, 7)),
|
||||
((11, 13), (7, 17)),
|
||||
],
|
||||
)
|
||||
def test_prime_sized_crops(
|
||||
self, src_shape: tuple[int, int], dst_shape: tuple[int, int]
|
||||
) -> None:
|
||||
"""Prime-sized crops with non-integer scale ratios match cv2 exactly."""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
rng = np.random.default_rng(src_shape[0] * 101 + dst_shape[1] * 53)
|
||||
mask = rng.random(src_shape) < 0.4
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, *src_shape, *dst_shape)
|
||||
result = _rle_counts_to_mask(result_rle, *dst_shape)
|
||||
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8),
|
||||
(dst_shape[1], dst_shape[0]),
|
||||
interpolation=cv2.INTER_NEAREST,
|
||||
).astype(bool)
|
||||
np.testing.assert_array_equal(result, expected)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("src_val", "src_shape", "dst_shape"),
|
||||
[
|
||||
(True, (1, 1), (32, 32)),
|
||||
(False, (1, 1), (32, 32)),
|
||||
],
|
||||
)
|
||||
def test_large_scale_ratio(
|
||||
self,
|
||||
src_val: bool,
|
||||
src_shape: tuple[int, int],
|
||||
dst_shape: tuple[int, int],
|
||||
) -> None:
|
||||
"""1x1 source resized to large shape fills entirely True or False."""
|
||||
from supervision.detection.compact_mask import _rle_resize
|
||||
|
||||
mask = np.full(src_shape, src_val, dtype=bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
result_rle = _rle_resize(rle, *src_shape, *dst_shape)
|
||||
result = _rle_counts_to_mask(result_rle, *dst_shape)
|
||||
|
||||
if src_val:
|
||||
assert result.all(), "1x1 True -> large shape must be all True"
|
||||
else:
|
||||
assert not result.any(), "1x1 False -> large shape must be all False"
|
||||
|
||||
def test_resize_dispatch_uses_l3_for_sparse(self) -> None:
|
||||
"""resize() dispatches to _rle_resize for sparse masks."""
|
||||
img_h, img_w = 100, 100
|
||||
masks = np.zeros((1, img_h, img_w), dtype=bool)
|
||||
masks[0, 50, 50] = True
|
||||
xyxy = mask_to_xyxy(masks).astype(np.float32)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(img_h, img_w))
|
||||
|
||||
resized = cm.resize((200, 200))
|
||||
|
||||
assert resized.shape == (1, 200, 200)
|
||||
dense = resized.to_dense()
|
||||
assert dense.sum() > 0
|
||||
|
||||
def test_resize_dispatch_uses_cv2_for_dense(self) -> None:
|
||||
"""_resize_crop falls back to cv2 for dense masks (above _L3_DENSITY_THRESHOLD).
|
||||
|
||||
Checkerboard yields ~1 run per pixel, far above the 0.25 threshold.
|
||||
Result must match cv2.resize(INTER_NEAREST) within 1 pixel.
|
||||
"""
|
||||
import cv2
|
||||
|
||||
from supervision.detection.compact_mask import (
|
||||
_L3_DENSITY_THRESHOLD,
|
||||
_resize_crop,
|
||||
)
|
||||
from supervision.detection.utils.converters import _mask_to_rle_counts
|
||||
|
||||
h, w = 20, 20
|
||||
# Checkerboard: alternates True/False → very dense RLE.
|
||||
rows, cols = np.meshgrid(np.arange(h), np.arange(w), indexing="ij")
|
||||
mask = ((rows + cols) % 2).astype(bool)
|
||||
rle = _mask_to_rle_counts(mask)
|
||||
density = len(rle) / max(1, h * w)
|
||||
assert density >= _L3_DENSITY_THRESHOLD, (
|
||||
f"Test precondition failed: density {density:.3f} < threshold "
|
||||
f"{_L3_DENSITY_THRESHOLD}; checkerboard should be dense"
|
||||
)
|
||||
|
||||
result_rle = _resize_crop(rle, h, w, h // 2, w // 2)
|
||||
result = _rle_counts_to_mask(result_rle, h // 2, w // 2)
|
||||
expected = cv2.resize(
|
||||
mask.astype(np.uint8), (w // 2, h // 2), interpolation=cv2.INTER_NEAREST
|
||||
).astype(bool)
|
||||
diff = np.abs(result.astype(int) - expected.astype(int)).max()
|
||||
assert int(diff) <= 1, f"Dense-path cv2 parity failed; max pixel diff={diff}"
|
||||
|
||||
|
||||
class TestResizeParallelPath:
|
||||
"""Tests for CompactMask.resize() thread-pool code path (N >= 8 masks)."""
|
||||
|
||||
def test_parallel_resize_correctness(self) -> None:
|
||||
"""resize() with N=10 masks exercises ThreadPoolExecutor; output is correct."""
|
||||
img_h, img_w = 80, 80
|
||||
n = 10 # above _PARALLEL_THRESHOLD = 8
|
||||
masks = np.zeros((n, img_h, img_w), dtype=bool)
|
||||
for i in range(n):
|
||||
r = 10 + i * 3
|
||||
masks[i, r : r + 8, r : r + 8] = True
|
||||
xyxy = mask_to_xyxy(masks).astype(np.float32)
|
||||
cm = CompactMask.from_dense(masks, xyxy, image_shape=(img_h, img_w))
|
||||
|
||||
target = (40, 40)
|
||||
resized = cm.resize(target)
|
||||
|
||||
assert resized.shape == (n, target[0], target[1])
|
||||
assert len(resized) == n
|
||||
# Each resized mask must be non-empty (the small squares survive downscale).
|
||||
for i in range(n):
|
||||
assert resized[i].any(), f"Mask {i} is empty after parallel resize"
|
||||
|
||||
def test_parallel_matches_sequential(self) -> None:
|
||||
"""Thread-pool path produces the same result as the sequential path."""
|
||||
img_h, img_w = 60, 60
|
||||
n_parallel = 10 # triggers thread pool
|
||||
n_sequential = 4 # stays sequential
|
||||
rng = np.random.default_rng(0)
|
||||
|
||||
def _make_masks(n: int) -> CompactMask:
|
||||
masks = np.zeros((n, img_h, img_w), dtype=bool)
|
||||
for i in range(n):
|
||||
r, c = int(rng.integers(5, 30)), int(rng.integers(5, 30))
|
||||
masks[i, r : r + 10, c : c + 10] = True
|
||||
xyxy = mask_to_xyxy(masks).astype(np.float32)
|
||||
return CompactMask.from_dense(masks, xyxy, image_shape=(img_h, img_w))
|
||||
|
||||
cm_par = _make_masks(n_parallel)
|
||||
cm_seq = _make_masks(n_sequential)
|
||||
|
||||
target = (30, 30)
|
||||
resized_par = cm_par.resize(target)
|
||||
resized_seq = cm_seq.resize(target)
|
||||
|
||||
# Both return correct shapes.
|
||||
assert resized_par.shape == (n_parallel, target[0], target[1])
|
||||
assert resized_seq.shape == (n_sequential, target[0], target[1])
|
||||
|
|
|
|||
Loading…
Reference in New Issue