supervision/src/supervision/detection/compact_mask.py

1333 lines
48 KiB
Python

"""Crop-RLE compact mask storage for memory-efficient instance segmentation.
Dense ``(N, H, W)`` boolean masks use O(N·H·W) memory, which becomes
prohibitive for aerial imagery (e.g. 1000 objects x 4K image ~ 8.3 GB).
:class:`CompactMask` stores each mask as a run-length encoding of its
bounding-box crop, reducing typical usage to tens of MB.
The bounding boxes (``xyxy``) already present in ``Detections`` serve as the
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 cast, overload
import numpy as np
import numpy.typing as npt
from supervision.detection.utils.converters import (
_mask_to_rle_counts,
_rle_counts_to_mask,
)
def _rle_area(rle: npt.NDArray[np.int32]) -> int:
"""Return the number of ``True`` pixels in a run-length encoded mask.
Args:
rle: int32 array of run lengths as produced by :func:`_mask_to_rle_counts`.
Returns:
Total number of ``True`` pixels.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import _rle_area
>>> rle = np.array([1, 2, 1, 1, 1], dtype=np.int32)
>>> _rle_area(rle)
3
```
"""
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:
src_c_int = int(src_c)
if src_c_int not in col_cache:
col_cache[src_c_int] = _rle_scale_col(per_col[src_c_int], crop_h, row_map)
scaled_cols.append(col_cache[src_c_int])
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.
Instead of storing N full ``(H, W)`` boolean arrays, :class:`CompactMask`
encodes each mask as a run-length sequence of its bounding-box crop. This
reduces memory from O(N·H·W) to roughly O(N·bbox_area), which is orders of
magnitude smaller for sparse masks on high-resolution images.
The class exposes a duck-typed interface compatible with ``np.ndarray``
masks used elsewhere in ``supervision``:
* ``mask[int]`` → dense ``(H, W)`` bool array (annotators, converters).
* ``mask[slice | list | ndarray]`` → new :class:`CompactMask` (filtering).
* ``np.asarray(mask)`` → dense ``(N, H, W)`` bool array (numpy interop).
* ``mask.shape``, ``mask.dtype``, ``mask.area`` — match the dense API.
:class:`CompactMask` is **not** a drop-in ``np.ndarray`` replacement.
When you need to call arbitrary ndarray methods (``astype``, ``reshape``,
``ravel``, ``any``, ``all``, …) call :meth:`to_dense` first:
``cm.to_dense().astype(np.uint8)``. :meth:`to_dense` is the single
explicit materialisation boundary.
.. note:: **RLE encoding — COCO / pycocotools pixel-scan order**
:class:`CompactMask` uses **column-major (Fortran-order, F-order)**
run-lengths scoped to each mask's bounding-box crop, matching the
pixel-scan order used by the COCO API (pycocotools). The crop scope
still differs from the full-image scope used by pycocotools, so a
:class:`CompactMask` RLE cannot be passed directly to
``maskUtils.iou()`` or ``maskUtils.decode()`` without re-scoping to
the full canvas. Use :meth:`to_dense` to obtain a standard boolean
array for pycocotools interop.
This scan order is part of CompactMask's internal RLE representation.
Switching from row-major (C-order) to column-major (F-order) is a
backward-incompatible format change for any persisted or serialized
:class:`CompactMask` state, including pickled objects and any
external storage of ``._rles``. Older stored RLE arrays will decode
incorrectly under the new convention.
Migration note: load or decode legacy masks with the older version,
materialize them to dense boolean arrays, and then re-encode them
with the current version (for example via :meth:`to_dense` followed
by :meth:`from_dense`) before persisting them again.
Args:
rles: List of N int32 run-length arrays.
crop_shapes: Array of shape ``(N, 2)`` — ``(crop_h, crop_w)`` per mask.
offsets: Array of shape ``(N, 2)`` — ``(x1, y1)`` bounding-box origins.
image_shape: ``(H, W)`` of the full image.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((2, 100, 100), dtype=bool)
>>> masks[0, 10:20, 10:20] = True
>>> masks[1, 50:70, 50:80] = True
>>> xyxy = np.array([[10, 10, 19, 19], [50, 50, 79, 69]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(100, 100))
>>> len(cm)
2
>>> cm.shape
(2, 100, 100)
```
"""
__slots__ = ("_crop_shapes", "_image_shape", "_offsets", "_rles")
def __init__(
self,
rles: list[npt.NDArray[np.int32]],
crop_shapes: npt.NDArray[np.int32],
offsets: npt.NDArray[np.int32],
image_shape: tuple[int, int],
) -> None:
self._rles: list[npt.NDArray[np.int32]] = rles
self._crop_shapes: npt.NDArray[np.int32] = crop_shapes # (N,2): (h,w)
self._offsets: npt.NDArray[np.int32] = offsets # (N,2): (x1,y1)
self._image_shape: tuple[int, int] = image_shape # (H, W)
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
@classmethod
def from_dense(
cls,
masks: npt.NDArray[np.bool_],
xyxy: npt.NDArray[np.number],
image_shape: tuple[int, int],
) -> CompactMask:
"""Create a :class:`CompactMask` from a dense ``(N, H, W)`` bool array.
Bounding boxes are clipped to image bounds and interpreted in the
supervision ``xyxy`` convention (inclusive max coordinates). A
box with invalid ordering (``x2 < x1`` or ``y2 < y1``) is replaced by
a ``1x1`` all-False crop to avoid degenerate RLE.
Args:
masks: Dense boolean mask array of shape ``(N, H, W)``.
xyxy: Bounding boxes of shape ``(N, 4)`` in ``[x1, y1, x2, y2]``
format.
image_shape: ``(H, W)`` of the full image.
Returns:
A new :class:`CompactMask` instance.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 100, 100), dtype=bool)
>>> masks[0, 10:20, 10:20] = True
>>> xyxy = np.array([[10, 10, 19, 19]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(100, 100))
>>> cm.shape
(1, 100, 100)
```
"""
img_h, img_w = image_shape
num_masks = len(masks)
if num_masks == 0:
return cls(
[],
np.empty((0, 2), dtype=np.int32),
np.empty((0, 2), dtype=np.int32),
image_shape,
)
rles: list[npt.NDArray[np.int32]] = []
crop_shapes_list: list[tuple[int, int]] = []
offsets_list: list[tuple[int, int]] = []
for mask_idx in range(num_masks):
x1, y1, x2, y2 = xyxy[mask_idx]
x1c = int(max(0, min(int(x1), img_w - 1)))
y1c = int(max(0, min(int(y1), img_h - 1)))
x2c = int(max(0, min(int(x2), img_w - 1)))
y2c = int(max(0, min(int(y2), img_h - 1)))
crop: npt.NDArray[np.bool_]
# supervision xyxy uses inclusive max coords, so slicing must add +1.
if x2c < x1c or y2c < y1c:
crop = np.zeros((1, 1), dtype=bool)
x2c, y2c = x1c, y1c
else:
crop = masks[mask_idx, y1c : y2c + 1, x1c : x2c + 1]
crop_h = y2c - y1c + 1
crop_w = x2c - x1c + 1
rles.append(_mask_to_rle_counts(crop))
crop_shapes_list.append((crop_h, crop_w))
offsets_list.append((x1c, y1c))
crop_shapes = np.array(crop_shapes_list, dtype=np.int32)
offsets = np.array(offsets_list, dtype=np.int32)
return cls(rles, crop_shapes, offsets, image_shape)
# ------------------------------------------------------------------
# Materialisation
# ------------------------------------------------------------------
def to_dense(self) -> npt.NDArray[np.bool_]:
"""Materialise all masks as a dense ``(N, H, W)`` boolean array.
Returns:
Boolean array of shape ``(N, H, W)``.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 50, 50), dtype=bool)
>>> masks[0, 10:20, 10:30] = True
>>> xyxy = np.array([[10, 10, 29, 19]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(50, 50))
>>> cm.to_dense().shape
(1, 50, 50)
```
"""
num_masks = len(self._rles)
img_h, img_w = self._image_shape
result: npt.NDArray[np.bool_] = np.zeros((num_masks, img_h, img_w), dtype=bool)
for mask_idx in range(num_masks):
crop_h, crop_w = (
int(self._crop_shapes[mask_idx, 0]),
int(self._crop_shapes[mask_idx, 1]),
)
x1, y1 = int(self._offsets[mask_idx, 0]), int(self._offsets[mask_idx, 1])
crop = _rle_counts_to_mask(self._rles[mask_idx], crop_h, crop_w)
result[mask_idx, y1 : y1 + crop_h, x1 : x1 + crop_w] = crop
return result
def crop(self, index: int) -> npt.NDArray[np.bool_]:
"""Decode a single mask crop without allocating the full image array.
This is an O(crop_area) operation — ideal for annotators that only
need the cropped region.
Args:
index: Index of the mask to decode.
Returns:
Boolean array of shape ``(crop_h, crop_w)``.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 100, 100), dtype=bool)
>>> masks[0, 20:30, 10:40] = True
>>> xyxy = np.array([[10, 20, 39, 29]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(100, 100))
>>> cm.crop(0).shape
(10, 30)
```
"""
crop_h = int(self._crop_shapes[index, 0])
crop_w = int(self._crop_shapes[index, 1])
return _rle_counts_to_mask(self._rles[index], crop_h, crop_w)
# ------------------------------------------------------------------
# Sequence / array protocol
# ------------------------------------------------------------------
def __len__(self) -> int:
"""Return the number of masks.
Returns:
Number of masks N.
Examples:
```pycon
>>> from supervision.detection.compact_mask import CompactMask
>>> import numpy as np
>>> cm = CompactMask(
... [], np.empty((0, 2), dtype=np.int32),
... np.empty((0, 2), dtype=np.int32), (100, 100))
>>> len(cm)
0
```
"""
return len(self._rles)
def __iter__(self) -> Iterator[npt.NDArray[np.bool_]]:
"""Iterate over masks as dense ``(H, W)`` boolean arrays."""
for mask_idx in range(len(self)):
yield self[mask_idx]
@property
def shape(self) -> tuple[int, int, int]:
"""Return ``(N, H, W)`` matching the dense mask convention.
Returns:
Tuple ``(N, H, W)``.
Examples:
```pycon
>>> from supervision.detection.compact_mask import CompactMask
>>> import numpy as np
>>> cm = CompactMask(
... [], np.empty((0, 2), dtype=np.int32),
... np.empty((0, 2), dtype=np.int32), (480, 640))
>>> cm.shape
(0, 480, 640)
```
"""
img_h, img_w = self._image_shape
return (len(self), img_h, img_w)
@property
def offsets(self) -> npt.NDArray[np.int32]:
"""Return per-mask crop origins as ``(x1, y1)`` integer offsets.
Returns:
Array of shape ``(N, 2)`` with ``int32`` offsets.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> masks[0, 2:4, 3:5] = True
>>> xyxy = np.array([[3, 2, 4, 3]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> cm.offsets.tolist()
[[3, 2]]
```
"""
return self._offsets
@property
def bbox_xyxy(self) -> npt.NDArray[np.int32]:
"""Return per-mask inclusive bounding boxes in ``xyxy`` format.
Boxes are derived from crop metadata:
``x2 = x1 + crop_w - 1``, ``y2 = y1 + crop_h - 1``.
Returns:
Array of shape ``(N, 4)`` with ``int32`` boxes
``[x1, y1, x2, y2]``.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> masks[0, 2:5, 3:7] = True
>>> xyxy = np.array([[3, 2, 6, 4]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> cm.bbox_xyxy.tolist()
[[3, 2, 6, 4]]
```
"""
if len(self) == 0:
return np.empty((0, 4), dtype=np.int32)
x1: npt.NDArray[np.int32] = self._offsets[:, 0]
y1: npt.NDArray[np.int32] = self._offsets[:, 1]
x2: npt.NDArray[np.int32] = x1 + self._crop_shapes[:, 1] - 1
y2: npt.NDArray[np.int32] = y1 + self._crop_shapes[:, 0] - 1
return np.column_stack((x1, y1, x2, y2)).astype(np.int32, copy=False)
@property
def dtype(self) -> np.dtype[np.bool_]:
"""Return ``np.dtype(bool)`` — always.
Returns:
``np.dtype(bool)``.
Examples:
```pycon
>>> from supervision.detection.compact_mask import CompactMask
>>> import numpy as np
>>> cm = CompactMask(
... [], np.empty((0, 2), dtype=np.int32),
... np.empty((0, 2), dtype=np.int32), (100, 100))
>>> cm.dtype
dtype('bool')
```
"""
return np.dtype(bool)
@property
def area(self) -> npt.NDArray[np.int64]:
"""Compute the area (``True`` pixel count) of each mask.
Returns:
int64 array of shape ``(N,)`` with per-mask pixel counts.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((2, 100, 100), dtype=bool)
>>> masks[0, 0:10, 0:10] = True # 100 pixels
>>> masks[1, 0:5, 0:5] = True # 25 pixels
>>> xyxy = np.array([[0, 0, 9, 9], [0, 0, 4, 4]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(100, 100))
>>> cm.area.tolist()
[100, 25]
```
"""
return np.array([_rle_area(rle) for rle in self._rles], dtype=np.int64)
def sum(
self, axis: int | tuple[int, ...] | None = None
) -> npt.NDArray[np.int64] | np.int64:
"""NumPy-compatible sum with a fast path for per-mask area.
When ``axis=(1, 2)``, returns the per-mask True-pixel count via
:attr:`area` without materialising the full dense array.
Args:
axis: Axis or axes to sum over.
Returns:
Sum result matching NumPy semantics.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> masks[0, 0:3, 0:3] = True
>>> xyxy = np.array([[0, 0, 2, 2]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> cm.sum(axis=(1, 2)).tolist()
[9]
```
"""
if axis == (1, 2):
return self.area
return cast(npt.NDArray[np.int64] | np.int64, self.to_dense().sum(axis=axis))
@overload
def __getitem__(self, index: int | np.integer) -> npt.NDArray[np.bool_]: ...
@overload
def __getitem__(
self,
index: slice
| list[int]
| list[bool]
| npt.NDArray[np.int_]
| npt.NDArray[np.bool_],
) -> CompactMask: ...
def __getitem__(
self,
index: (
int
| np.integer
| slice
| list[int]
| list[bool]
| npt.NDArray[np.int_]
| npt.NDArray[np.bool_]
),
) -> npt.NDArray[np.bool_] | CompactMask:
"""Index into the mask collection.
* ``int`` → dense ``(H, W)`` bool array (for annotators, iterators).
* ``slice | list | ndarray`` → new :class:`CompactMask` (for filtering).
Args:
index: An integer returns a dense ``(H, W)`` mask. Any other
supported index type returns a new :class:`CompactMask`.
Returns:
Dense ``(H, W)`` ``np.ndarray`` for integer index, or a new
:class:`CompactMask` for all other index types.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((3, 20, 20), dtype=bool)
>>> xyxy = np.array(
... [[0,0,5,5],[5,5,10,10],[10,10,15,15]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(20, 20))
>>> cm[0].shape # int → dense (H, W)
(20, 20)
>>> len(cm[[0, 2]]) # list → CompactMask
2
```
"""
if isinstance(index, (int, np.integer)):
idx = int(index)
img_h, img_w = self._image_shape
result: npt.NDArray[np.bool_] = np.zeros((img_h, img_w), dtype=bool)
crop_h = int(self._crop_shapes[idx, 0])
crop_w = int(self._crop_shapes[idx, 1])
x1 = int(self._offsets[idx, 0])
y1 = int(self._offsets[idx, 1])
crop = _rle_counts_to_mask(self._rles[idx], crop_h, crop_w)
result[y1 : y1 + crop_h, x1 : x1 + crop_w] = crop
return result
# Slice: use direct Python list slice and numpy view — O(k), no arange.
if isinstance(index, slice):
return CompactMask(
self._rles[index],
self._crop_shapes[index],
self._offsets[index],
self._image_shape,
)
# Boolean selectors and fancy index → convert to integer positions first.
if isinstance(index, np.ndarray) and index.dtype == bool:
idx_arr = np.where(index)[0]
elif isinstance(index, list) and all(
isinstance(item, (bool, np.bool_)) for item in index
):
idx_arr = np.flatnonzero(np.asarray(index, dtype=bool))
else:
idx_arr = np.asarray(list(index), dtype=np.intp)
new_rles = [self._rles[int(mask_idx)] for mask_idx in idx_arr]
new_crop_shapes: npt.NDArray[np.int32] = self._crop_shapes[idx_arr]
new_offsets: npt.NDArray[np.int32] = self._offsets[idx_arr]
return CompactMask(new_rles, new_crop_shapes, new_offsets, self._image_shape)
def __array__(
self, dtype: np.dtype[np.generic] | None = None
) -> npt.NDArray[np.generic]:
"""NumPy interop: materialise as a dense ``(N, H, W)`` array.
Called by ``np.asarray(compact_mask)`` and similar NumPy functions.
Args:
dtype: Optional dtype to cast the result to.
Returns:
Dense boolean array of shape ``(N, H, W)``.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> xyxy = np.array([[0, 0, 5, 5]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> np.asarray(cm).shape
(1, 10, 10)
```
"""
result = self.to_dense()
if dtype is not None:
return result.astype(dtype)
return result
def __eq__(self, other: object) -> bool:
"""Element-wise equality with another :class:`CompactMask` or ndarray.
Args:
other: Another :class:`CompactMask` or ``np.ndarray``.
Returns:
``True`` if all masks are pixel-identical.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> xyxy = np.array([[0, 0, 5, 5]], dtype=np.float32)
>>> cm1 = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> cm2 = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> cm1 == cm2
True
```
"""
if isinstance(other, CompactMask):
return bool(np.array_equal(self.to_dense(), other.to_dense()))
if isinstance(other, np.ndarray):
return bool(np.array_equal(self.to_dense(), other))
return NotImplemented
# ------------------------------------------------------------------
# Collection utilities
# ------------------------------------------------------------------
@staticmethod
def merge(masks_list: list[CompactMask]) -> CompactMask:
"""Concatenate multiple :class:`CompactMask` objects into one.
All inputs must have the same ``image_shape``.
Args:
masks_list: Non-empty list of :class:`CompactMask` objects.
Returns:
A new :class:`CompactMask` containing every mask from the inputs,
in order.
Raises:
ValueError: If ``masks_list`` is empty or image shapes differ.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks1 = np.zeros((2, 50, 50), dtype=bool)
>>> masks2 = np.zeros((3, 50, 50), dtype=bool)
>>> xyxy1 = np.array([[0,0,10,10],[10,10,20,20]], dtype=np.float32)
>>> xyxy2 = np.array(
... [[0,0,5,5],[5,5,10,10],[10,10,15,15]], dtype=np.float32)
>>> cm1 = CompactMask.from_dense(masks1, xyxy1, image_shape=(50, 50))
>>> cm2 = CompactMask.from_dense(masks2, xyxy2, image_shape=(50, 50))
>>> len(CompactMask.merge([cm1, cm2]))
5
```
"""
if not masks_list:
raise ValueError("Cannot merge an empty list of CompactMask objects.")
image_shape = masks_list[0]._image_shape
for cm in masks_list[1:]:
if cm._image_shape != image_shape:
raise ValueError(
f"Cannot merge CompactMask objects with different image shapes: "
f"{image_shape} vs {cm._image_shape}"
)
# list.extend is a C-level call and avoids the per-element Python
# bytecode overhead of a flat list comprehension. This matters under
# GIL contention when multiple threads call merge concurrently.
new_rles: list[npt.NDArray[np.int32]] = []
for cm in masks_list:
new_rles.extend(cm._rles)
# np.concatenate handles (0, 2) arrays correctly.
# No .astype() needed — _crop_shapes and _offsets are already int32.
new_crop_shapes: npt.NDArray[np.int32] = np.concatenate(
[cm._crop_shapes for cm in masks_list], axis=0
)
new_offsets: npt.NDArray[np.int32] = np.concatenate(
[cm._offsets for cm in masks_list], axis=0
)
return CompactMask(new_rles, new_crop_shapes, new_offsets, image_shape)
def repack(self) -> CompactMask:
"""Re-encode all masks using tight bounding boxes.
When the original ``xyxy`` boxes are padded or loose — common with
object-detector outputs and full-image boxes used in tests — each RLE
crop encodes more background (``False``) pixels than necessary. This
method decodes every crop, trims it to the minimal rectangle that
contains all ``True`` pixels, and re-encodes. All-``False`` masks are
normalised to a ``1x1`` all-``False`` crop.
The call is O(sum of crop areas) — suitable as a one-time cleanup
after accumulating many merges (e.g. after
:class:`~supervision.detection.tools.inference_slicer.InferenceSlicer`
tiles are merged).
Returns:
A new :class:`CompactMask` with minimal-area crops and updated
offsets.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 10, 10), dtype=bool)
>>> masks[0, 3:7, 3:7] = True
>>> # Deliberately loose bbox: covers the full image.
>>> xyxy = np.array([[0, 0, 9, 9]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(10, 10))
>>> repacked = cm.repack()
>>> repacked.offsets.tolist() # tight origin: x1=3, y1=3
[[3, 3]]
```
"""
num_masks = len(self._rles)
if num_masks == 0:
return CompactMask(
[],
np.empty((0, 2), dtype=np.int32),
np.empty((0, 2), dtype=np.int32),
self._image_shape,
)
new_rles: list[npt.NDArray[np.int32]] = []
new_crop_shapes_list: list[tuple[int, int]] = []
new_offsets_list: list[tuple[int, int]] = []
for mask_idx in range(num_masks):
crop = self.crop(mask_idx)
x1_off = int(self._offsets[mask_idx, 0])
y1_off = int(self._offsets[mask_idx, 1])
rows_any = np.any(crop, axis=1)
cols_any = np.any(crop, axis=0)
if not rows_any.any():
# All-False: normalise to 1x1 to avoid zero-sized arrays.
new_rles.append(_mask_to_rle_counts(np.zeros((1, 1), dtype=bool)))
new_crop_shapes_list.append((1, 1))
new_offsets_list.append((x1_off, y1_off))
continue
y_indices = np.where(rows_any)[0]
x_indices = np.where(cols_any)[0]
y_min, y_max = int(y_indices[0]), int(y_indices[-1])
x_min, x_max = int(x_indices[0]), int(x_indices[-1])
tight = crop[y_min : y_max + 1, x_min : x_max + 1]
new_rles.append(_mask_to_rle_counts(tight))
new_crop_shapes_list.append((y_max - y_min + 1, x_max - x_min + 1))
new_offsets_list.append((x1_off + x_min, y1_off + y_min))
return CompactMask(
new_rles,
np.array(new_crop_shapes_list, dtype=np.int32),
np.array(new_offsets_list, dtype=np.int32),
self._image_shape,
)
# ------------------------------------------------------------------
# Slicer support
# ------------------------------------------------------------------
def with_offset(
self,
dx: int,
dy: int,
new_image_shape: tuple[int, int],
) -> CompactMask:
"""Return a new :class:`CompactMask` with adjusted offsets and image shape.
Used by :class:`~supervision.detection.tools.inference_slicer.InferenceSlicer`
to relocate tile-local masks into full-image coordinates without
materialising the dense ``(N, H, W)`` array.
Args:
dx: Pixels to add to every mask's ``x1`` offset.
dy: Pixels to add to every mask's ``y1`` offset.
new_image_shape: ``(H, W)`` of the full (destination) image.
Returns:
New :class:`CompactMask` with updated offsets and image shape.
Crops are clipped to stay inside ``new_image_shape``; masks fully
outside are represented as ``1x1`` all-False crops.
Examples:
```pycon
>>> import numpy as np
>>> from supervision.detection.compact_mask import CompactMask
>>> masks = np.zeros((1, 20, 20), dtype=bool)
>>> xyxy = np.array([[5, 5, 15, 15]], dtype=np.float32)
>>> cm = CompactMask.from_dense(masks, xyxy, image_shape=(20, 20))
>>> cm2 = cm.with_offset(100, 200, new_image_shape=(400, 400))
>>> cm2.offsets[0].tolist()
[105, 205]
```
"""
new_h, new_w = new_image_shape
if new_h <= 0 or new_w <= 0:
raise ValueError("new_image_shape must contain positive dimensions")
num_masks = len(self)
if num_masks == 0:
return CompactMask(
[],
np.empty((0, 2), dtype=np.int32),
np.empty((0, 2), dtype=np.int32),
new_image_shape,
)
# Vectorised bounds check: compute every new [x1,y1,x2,y2] at once.
# For the common case (InferenceSlicer tiles that fit fully inside the
# new canvas) this catches the "no clipping needed" path in O(N) numpy
# without touching any RLE data.
new_offsets: npt.NDArray[np.int32] = self._offsets + np.array(
[dx, dy], dtype=np.int32
)
x1s = new_offsets[:, 0]
y1s = new_offsets[:, 1]
x2s = x1s + self._crop_shapes[:, 1] - 1
y2s = y1s + self._crop_shapes[:, 0] - 1
needs_clip: npt.NDArray[np.bool_] = (
(x1s < 0) | (y1s < 0) | (x2s >= new_w) | (y2s >= new_h)
)
if not needs_clip.any():
# Fast path: pure offset arithmetic, no decode/re-encode needed.
return CompactMask(
list(self._rles),
self._crop_shapes.copy(),
new_offsets,
new_image_shape,
)
# Slow path: only decode+clip+re-encode the masks that actually overflow.
out_rles: list[npt.NDArray[np.int32]] = []
out_crop_shapes: list[tuple[int, int]] = []
out_offsets_list: list[tuple[int, int]] = []
for mask_idx in range(num_masks):
x1 = int(x1s[mask_idx])
y1 = int(y1s[mask_idx])
x2 = int(x2s[mask_idx])
y2 = int(y2s[mask_idx])
if not needs_clip[mask_idx]:
out_rles.append(self._rles[mask_idx])
out_crop_shapes.append(
(
int(self._crop_shapes[mask_idx, 0]),
int(self._crop_shapes[mask_idx, 1]),
)
)
out_offsets_list.append((x1, y1))
continue
ix1 = max(0, x1)
iy1 = max(0, y1)
ix2 = min(new_w - 1, x2)
iy2 = min(new_h - 1, y2)
if ix1 > ix2 or iy1 > iy2:
anchor_x = min(max(x1, 0), new_w - 1)
anchor_y = min(max(y1, 0), new_h - 1)
out_rles.append(_mask_to_rle_counts(np.zeros((1, 1), dtype=bool)))
out_crop_shapes.append((1, 1))
out_offsets_list.append((anchor_x, anchor_y))
continue
crop = self.crop(mask_idx)
clipped = crop[iy1 - y1 : iy2 - y1 + 1, ix1 - x1 : ix2 - x1 + 1]
out_rles.append(_mask_to_rle_counts(clipped))
out_crop_shapes.append((iy2 - iy1 + 1, ix2 - ix1 + 1))
out_offsets_list.append((ix1, iy1))
return CompactMask(
out_rles,
np.array(out_crop_shapes, dtype=np.int32),
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)