"""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. CompactMask reduces memory footprint but does not improve computational speed. The ingestion path (base48 decode, column split, RLE trim) is Python-level and is typically slower than the dense NumPy path. The primary benefit is memory savings for large images with many sparse masks. """ from __future__ import annotations import os from collections.abc import Iterator, Mapping, Sequence from typing import Any, cast, overload import numpy as np import numpy.typing as npt # _base48_decode and _delta_decode are private to the COCO RLE codec. They # live in converters.py and are shared by public conversion helpers here and # in that module. from supervision.detection.utils.converters import ( _base48_decode, _delta_decode, _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, x_start: int = 0, x_stop: int | None = None, ) -> 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`. When ``x_start`` and ``x_stop`` are provided, only columns in the closed range ``[x_start, x_stop]`` are collected. Pixels in skipped columns are consumed without being stored, which avoids O(W) allocation when only a small crop of a wide image is needed. Note: ``x_start`` uses ``np.cumsum`` + ``np.searchsorted`` to jump directly to the first relevant run in O(log R) time, avoiding the O(pixel_prefix) walk that previously made right-edge crops on wide images expensive. 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. x_start: First column to collect (0-indexed, inclusive). Columns before ``x_start`` are skipped. Defaults to ``0``. x_stop: Last column to collect (0-indexed, inclusive). Processing stops after column ``x_stop`` is complete. Defaults to ``crop_w - 1`` (collect all columns). Returns: List of ``x_stop - x_start + 1`` run lists. Index ``i`` in the returned list corresponds to column ``x_start + i``. 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]] >>> _rle_split_cols(rle, 2, 2, x_start=1) [[1, 1]] ``` """ if x_stop is None: x_stop = crop_w - 1 # Convert numpy int32 array to Python ints to avoid scalar boxing overhead # in the inner loop (np.int32 boxing/unboxing slows pure-Python loops). rle_list: list[int] = rle.tolist() n_cols = x_stop - x_start + 1 per_col: list[list[int]] = [[] for _ in range(n_cols)] # Fast-forward to the first run that overlaps column x_start using O(log R) # searchsorted instead of an O(pixel_prefix) sequential walk. start_pixel = x_start * crop_h if start_pixel > 0 and len(rle_list) > 0: cumsum_ends = np.cumsum(rle, dtype=np.int64) first_run = int(np.searchsorted(cumsum_ends, start_pixel, side="right")) if first_run >= len(rle_list): for c in range(n_cols): per_col[c] = [crop_h] return per_col prefix = int(cumsum_ends[first_run - 1]) if first_run > 0 else 0 else: first_run = 0 prefix = 0 col = prefix // crop_h row = prefix % crop_h for run_idx in range(first_run, len(rle_list)): run_len = rle_list[run_idx] is_true = run_idx % 2 == 1 remaining = run_len while remaining > 0: # Past the requested range — stop early. if col > x_stop: remaining = 0 break space_in_col = crop_h - row take = min(remaining, space_in_col) if col >= x_start: local_col = col - x_start if len(per_col[local_col]) == 0: if is_true: per_col[local_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[local_col]) - 1) % 2 == 1): per_col[local_col][-1] += take remaining -= take row += take if row >= crop_h: row = 0 col += 1 continue per_col[local_col].append(take) remaining -= take row += take if row >= crop_h: row = 0 col += 1 if col > x_stop: break # Fill any empty columns (all-False). for c in range(n_cols): 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_trim_col_runs(col_runs: Sequence[int], y1: int, y2: int) -> list[int]: """Restrict one full-height column RLE to inclusive rows ``[y1, y2]``. Args: col_runs: Run lengths for one full-height column, starting with a ``False`` run. y1: Inclusive top row of the crop. y2: Inclusive bottom row of the crop. Returns: Run lengths for the cropped column, also starting with a ``False`` run. Examples: ```pycon >>> from supervision.detection.compact_mask import _rle_trim_col_runs >>> # Full column F=2, T=2, F=2 (height 6); crop to rows 1..4 inclusive >>> # yields rows [F, T, T, F] → F=1, T=2, F=1. >>> _rle_trim_col_runs([2, 2, 2], y1=1, y2=4) [1, 2, 1] ``` """ target_height = y2 - y1 + 1 # Sum invariant: returned list sums to target_height. Correctness depends # on the caller (from_coco_rle) having already validated that counts sum # equals img_h * img_w; no re-validation here. collected: list[tuple[bool, int]] = [] row = 0 for run_idx, run_len in enumerate(col_runs): is_true = run_idx % 2 == 1 start = row end = row + int(run_len) row = end crop_start = max(start, y1) crop_end = min(end, y2 + 1) if crop_end > crop_start: collected.append((is_true, crop_end - crop_start)) if row > y2: break if not collected: return [target_height] result: list[int] = [] if collected[0][0]: result.append(0) for is_true, length in collected: last_is_true = bool(result) and (len(result) - 1) % 2 == 1 if result and last_is_true == is_true: result[-1] += length else: result.append(length) return result def _coco_rle_counts_to_array(counts: Any) -> npt.NDArray[np.int32]: """Decode COCO RLE counts into absolute F-order run lengths. Args: counts: COCO compressed counts (``str`` or ``bytes``), or uncompressed integer run lengths. Returns: One-dimensional ``int32`` run-length array. Raises: ValueError: If counts cannot be decoded into non-negative run lengths. Examples: ```pycon >>> from supervision.detection.compact_mask import _coco_rle_counts_to_array >>> _coco_rle_counts_to_array([0, 2, 2, 2, 10]) array([ 0, 2, 2, 2, 10], dtype=int32) ``` """ try: if isinstance(counts, bytes): counts = counts.decode("utf-8") if isinstance(counts, str): decoded_counts = _delta_decode(_base48_decode(counts)) counts_arr = np.array(decoded_counts, dtype=np.int32) else: # Convert to int64 first, then range-check against int32 bounds before # narrowing. A direct int32 cast wraps silently on some numpy versions # and raises on others; this makes overflow detection deterministic. counts_arr64 = np.asarray(counts, dtype=np.int64) int32_info = np.iinfo(np.int32) if counts_arr64.size and ( counts_arr64.max() > int32_info.max or counts_arr64.min() < int32_info.min ): raise ValueError("COCO RLE counts exceed int32 range.") counts_arr = counts_arr64.astype(np.int32) except (TypeError, ValueError, OverflowError) as exc: raise ValueError("Invalid COCO RLE counts.") from exc if counts_arr.ndim != 1: raise ValueError("COCO RLE counts must be one-dimensional.") if counts_arr.size == 0: raise ValueError("COCO RLE counts cannot be empty.") if np.any(counts_arr < 0): raise ValueError("COCO RLE counts must be non-negative.") return counts_arr 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_col = int(src_c) if src_col not in col_cache: col_cache[src_col] = _rle_scale_col(per_col[src_col], crop_h, row_map) scaled_cols.append(col_cache[src_col]) 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 # Hard ceiling on each image dimension accepted by from_coco_rle, guarding # against crafted payloads that allocate O(H x W) column lists. _MAX_IMAGE_DIMENSION: int = 32768 # Images at or below this pixel count use a fully-vectorised numpy dense-decode # path inside from_coco_rle instead of the pure-Python column-split loop. # Crossover measured at ~8-16 K px (128x128); threshold set at 128x128 = 16 384. _SMALL_IMAGE_DENSE_THRESHOLD: int = 128 * 128 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. """ from supervision import _cv2 as 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] x1i, y1i, x2i, y2i = int(x1), int(y1), int(x2), int(y2) x1c = int(max(0, min(x1i, img_w - 1))) y1c = int(max(0, min(y1i, img_h - 1))) crop: npt.NDArray[np.bool_] # supervision xyxy uses inclusive max coords, so slicing must add +1. if ( x2i < x1i or y2i < y1i or x2i < 0 or y2i < 0 or x1i >= img_w or y1i >= img_h ): crop = np.zeros((1, 1), dtype=bool) crop_h = 1 crop_w = 1 else: x2c = int(max(0, min(x2i, img_w - 1))) y2c = int(max(0, min(y2i, img_h - 1))) 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) @classmethod def from_coco_rle( cls, rles: Sequence[Mapping[str, Any]], xyxy: npt.NDArray[np.floating], image_shape: tuple[int, int], ) -> CompactMask: """Create a :class:`CompactMask` from full-frame COCO RLE masks. Transcodes full-image COCO RLE payloads into the crop-scoped RLE format used by :class:`CompactMask`. The conversion uses run-length arithmetic scoped by ``xyxy`` boxes and does not materialise a dense ``(N, H, W)`` mask stack. Args: rles: Sequence of COCO RLE dictionaries. Each dictionary must contain ``"size"`` as ``[height, width]`` and ``"counts"`` as compressed counts (``str`` or ``bytes``) or uncompressed integer run lengths. xyxy: Bounding boxes of shape ``(N, 4)`` in ``[x1, y1, x2, y2]`` format. Max coordinates follow supervision's inclusive convention. image_shape: ``(H, W)`` of the full image. This must match every RLE ``"size"`` value. Returns: A new :class:`CompactMask` instance. Raises: ValueError: If the RLE payloads are malformed, are not aligned with ``xyxy``, or their sizes/counts do not match ``image_shape``. Examples: ```pycon >>> import numpy as np >>> from supervision.detection.compact_mask import CompactMask >>> # 4x4 image with a 2x2 True block at the top-left corner. >>> # Uncompressed F-order COCO counts: F=0, T=2, F=2, T=2, F=10 >>> # (column-major: col0=[T,T,F,F], col1=[T,T,F,F], cols2-3 all F). >>> rles = [{"size": [4, 4], "counts": [0, 2, 2, 2, 10]}] >>> xyxy = np.array([[0, 0, 3, 3]], dtype=np.float32) >>> cm = CompactMask.from_coco_rle(rles, xyxy, image_shape=(4, 4)) >>> cm.shape (1, 4, 4) >>> cm.area.tolist() [4] ``` """ img_h, img_w = (int(image_shape[0]), int(image_shape[1])) if img_h <= 0 or img_w <= 0: raise ValueError("image_shape must contain positive height and width.") if img_h > _MAX_IMAGE_DIMENSION or img_w > _MAX_IMAGE_DIMENSION: raise ValueError( f"image_shape {(img_h, img_w)} exceeds the maximum allowed dimension " f"of {_MAX_IMAGE_DIMENSION} pixels per side." ) xyxy_arr = np.asarray(xyxy) if xyxy_arr.shape != (len(rles), 4): raise ValueError( "xyxy must have shape (N, 4), where N matches the number of RLEs." ) if len(rles) == 0: return cls( [], np.empty((0, 2), dtype=np.int32), np.empty((0, 2), dtype=np.int32), (img_h, img_w), ) crop_rles: list[npt.NDArray[np.int32]] = [] crop_shapes_list: list[tuple[int, int]] = [] offsets_list: list[tuple[int, int]] = [] for mask_idx, rle in enumerate(rles): if not isinstance(rle, Mapping): raise ValueError("Each RLE payload must be a mapping.") if "size" not in rle or "counts" not in rle: raise ValueError("Each RLE payload must contain 'size' and 'counts'.") try: # COCO standard: size=[height, width] (h,w order per pycocotools spec) rle_h, rle_w = rle["size"] rle_h = int(rle_h) rle_w = int(rle_w) except (TypeError, ValueError) as exc: raise ValueError("RLE size must be [height, width].") from exc if (rle_h, rle_w) != (img_h, img_w): raise ValueError( f"RLE size {(rle_h, rle_w)} must match image_shape " f"{(img_h, img_w)}." ) counts = _coco_rle_counts_to_array(rle["counts"]) if int(np.sum(counts, dtype=np.int64)) != img_h * img_w: raise ValueError( "The sum of COCO RLE counts must match the image area." ) x1, y1, x2, y2 = xyxy_arr[mask_idx] x1i, y1i, x2i, y2i = int(x1), int(y1), int(x2), int(y2) x1c = max(0, min(x1i, img_w - 1)) y1c = max(0, min(y1i, img_h - 1)) if ( x2i < x1i or y2i < y1i or x2i < 0 or y2i < 0 or x1i >= img_w or y1i >= img_h ): crop_rles.append(np.array([1], dtype=np.int32)) crop_shapes_list.append((1, 1)) offsets_list.append((x1c, y1c)) continue x2c = max(0, min(x2i, img_w - 1)) y2c = max(0, min(y2i, img_h - 1)) crop_h = y2c - y1c + 1 crop_w = x2c - x1c + 1 if img_h * img_w <= _SMALL_IMAGE_DENSE_THRESHOLD: # Small image: vectorised numpy decode avoids the O(img_w)-column # Python loop. Decode RLE to flat F-order bool, extract crop, and # re-encode directly. ends = np.cumsum(counts, dtype=np.int64) starts = ends - counts.astype(np.int64) # Mark True runs (odd-indexed) via difference-array decode (O(R)). true_starts = starts[1::2] true_ends = ends[1::2] if true_starts.size > 0: indicator = np.zeros(img_h * img_w + 1, dtype=np.int32) np.add.at(indicator, true_starts, 1) np.add.at(indicator, true_ends, -1) # cumsum in int32 avoids int8 overflow; cast to uint8 (0/1). flat = np.cumsum(indicator[:-1], dtype=np.int32).astype(np.uint8) else: flat = np.zeros(img_h * img_w, dtype=np.uint8) # Extract crop: reshape to (img_w, img_h) F-order view, slice. flat_crop = flat.reshape(img_w, img_h)[ x1c : x2c + 1, y1c : y2c + 1 ].ravel() # RLE-encode the flat crop: find value-change positions. change_pos = np.where(np.diff(flat_crop.view(np.int8)))[0] + 1 boundaries = np.concatenate([[0], change_pos, [len(flat_crop)]]) run_lens = np.diff(boundaries) if flat_crop[0]: run_lens = np.concatenate([[0], run_lens]) crop_rle_arr = run_lens.astype(np.int32) else: cols = _rle_split_cols(counts, img_h, img_w, x_start=x1c, x_stop=x2c) selected_columns = [_rle_trim_col_runs(col, y1c, y2c) for col in cols] crop_rle_arr = _rle_join_cols(selected_columns, crop_h * crop_w) crop_rles.append(crop_rle_arr) 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(crop_rles, crop_shapes, offsets, (img_h, img_w)) # ------------------------------------------------------------------ # 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 image_shape(self) -> tuple[int, int]: """Return ``(H, W)`` of the full image this mask is scoped to. Returns: Tuple ``(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.image_shape (480, 640) ``` """ return self._image_shape @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. Note: The implementation iterates over the N individual RLE arrays in a Python loop (one :func:`_rle_area` call per mask). This is negligible for typical N, but callers processing thousands of detections per frame should be aware of the per-mask Python-level overhead. 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 if isinstance(index, slice): selected_rles = [rle.copy() for rle in self._rles[index]] return CompactMask( selected_rles, self._crop_shapes[index].copy(), self._offsets[index].copy(), 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)].copy() for mask_idx in idx_arr] new_crop_shapes: npt.NDArray[np.int32] = self._crop_shapes[idx_arr].copy() new_offsets: npt.NDArray[np.int32] = self._offsets[idx_arr].copy() return CompactMask(new_rles, new_crop_shapes, new_offsets, self._image_shape) def __array__( 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)