Skip to content

Mask Builders

Mask builders are tools for assembling feature masks from neural network predictions or other tile-level data. They handle the complexity of combining overlapping tiles, scaling between coordinate spaces, and managing memory for large output masks using a flexible strategy-based architecture.

Overview

When processing whole-slide images with neural networks, you often need to:

  1. Extract tiles from a slide
  2. Run inference to get predictions or features for each tile
  3. Assemble these predictions back into a full-resolution mask

Mask builders automate step 3, handling:

  • Coordinate scaling: Converting from source WSI coordinates to mask coordinates — including automatic GCD-based compression when tiles and strides share common factors.
  • Overlap handling: Averaging or taking the maximum when tiles overlap.
  • Memory management: Using in-memory arrays or memory-mapped files for large masks.
  • Scalar expansion: Broadcasting scalar per-tile predictions (B, C) into spatial tiles automatically.
  • Edge clipping: Removing border artifacts from model output tiles at update time.

MaskBuilder

Builder for assembling large masks from tiled data with automatic scaling and clipping.

This class coordinates: - Coordinate scaling: Maps source WSI coordinates to mask resolution. - Edge clipping: Removes artifacts from model output tiles. - Pluggable storage: Allocates memory (RAM vs disk-backed). - Pluggable aggregation: Merges overlapping tiles (e.g., mean, max).

Source code in ratiopath/masks/mask_builders/mask_builder.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
class MaskBuilder[DType: np.generic, AggregatorR]:
    """Builder for assembling large masks from tiled data with automatic scaling and clipping.

    This class coordinates:
    - Coordinate scaling: Maps source WSI coordinates to mask resolution.
    - Edge clipping: Removes artifacts from model output tiles.
    - Pluggable storage: Allocates memory (RAM vs disk-backed).
    - Pluggable aggregation: Merges overlapping tiles (e.g., mean, max).
    """

    storage: NDArray[DType]
    aggregator: Aggregator[DType, AggregatorR]

    def __init__(
        self,
        source_extents: tuple[int, ...],
        source_tile_extent: int | tuple[int, ...],
        output_tile_extent: int | tuple[int, ...],
        stride: int | tuple[int, ...],
        n_channels: int = 1,
        storage: Literal["inmemory", "memmap"]
        | Callable[..., NDArray[DType]] = "inmemory",
        aggregation: type[Aggregator[DType, AggregatorR]] = MeanAggregator,  # type: ignore[assignment]
        dtype: type[DType] = np.float32,  # type: ignore[assignment]
        **kwargs: Any,
    ) -> None:
        """Initialize the mask builder.

        Args:
            source_extents: Spatial dimensions (H, W, ...) of the source WSI.
            source_tile_extent: Spatial dimensions of the model input tiles.
            output_tile_extent: Spatial dimensions of the model output tiles.
            stride: Stride between tiles in source resolution.
            n_channels: Number of channels in the output mask.
            storage: Strategy for allocating memory ("inmemory", "memmap", or a class).
            aggregation: Strategy for combining tiles ("mean", "max", or a class).
            dtype: Data type for the accumulator.
            kwargs: Extra arguments passed to storage and aggregation initialization.
        """
        # Normalize spatial dimensions to arrays
        self.source_extents = np.asarray(source_extents, dtype=np.int64)
        self.source_tile_extent = np.broadcast_to(
            source_tile_extent, len(source_extents)
        )
        self.output_tile_extent = np.broadcast_to(
            output_tile_extent, len(source_extents)
        )
        self.stride = np.broadcast_to(stride, len(source_extents))

        # Calculate how many tiles are required to fully cover the WSI (implicitly padding the edges)
        # Using np.ceil to ensure the right/bottom edges are fully covered.
        num_tiles = (
            np.ceil(
                np.maximum(0, self.source_extents - self.source_tile_extent)
                / self.stride
            ).astype(np.int64)
            + 1
        )

        # Calculate the actual spatial span these tiles cover in the original WSI space
        self.span = (num_tiles - 1) * self.stride + self.source_tile_extent

        # Find the Greatest Common Divisor for the alignment math
        gcd = np.gcd(self.source_tile_extent, self.stride * self.output_tile_extent)

        # Calculate the required MaskBuilder properties
        self.mask_extents = (self.span * self.output_tile_extent) // gcd
        self.upscale_factor = self.source_tile_extent // gcd
        self.mask_stride = (self.stride * self.output_tile_extent) // gcd

        # Initialize Storage
        if isinstance(storage, str):
            if storage == "inmemory":
                from ratiopath.masks.mask_builders.storage import InMemory

                storage = InMemory
            elif storage == "memmap":
                from ratiopath.masks.mask_builders.storage import MemMap

                storage = MemMap
            else:
                raise ValueError(f"Unknown storage type: {storage}")

        self.storage = storage(
            shape=(n_channels, *self.mask_extents), dtype=dtype, **kwargs
        )

        # Initialize Aggregator
        self.aggregator = aggregation(storage=self.storage, **kwargs)

    def update_batch(
        self,
        batch: NDArray[DType],
        coords: NDArray[np.int64],
        edge_clipping: EdgeClipping = 0,
    ) -> None:
        """Update the accumulator with a batch of tiles.

        Args:
            batch: Array of shape (B, C, *SpatialDims) or (B, C) containing B tiles.
            coords: Array of shape (B, N) containing top-left coordinates in source resolution.
            edge_clipping: Pixels to clip from tile edges.
                Supports an int (symmetric for all dims), a tuple of N ints
                (symmetric per dim), or a tuple of N (start, end) pairs.
        """
        # Scale coordinates from source to mask resolution
        # Find which tile index this corresponds to and multiply by the mask stride
        # to get the starting pixel in the mask array
        coords = (coords // self.stride) * self.mask_stride

        # Handle scalar data expansion
        # Broadcast scalar (B, C) to (B, C, *output_tile_extent)
        # If already dense, this is a fast view operation
        num_missing_dims = len(self.mask_extents) + 2 - batch.ndim
        batch = np.broadcast_to(
            batch[..., *[np.newaxis] * num_missing_dims],
            (*batch.shape[:2], *self.output_tile_extent),
        )

        # Apply Edge Clipping
        slices, shift = _prepare_clipping(
            len(self.mask_extents), tuple(self.output_tile_extent), edge_clipping
        )
        batch = batch[slices]
        coords += shift * self.upscale_factor

        # Upscale the batch to match the mask resolution if needed
        for axis_idx, factor in enumerate(self.upscale_factor, start=2):
            if factor > 1:
                batch = np.repeat(batch, factor, axis=axis_idx)

        for sample, coord in zip(batch, coords, strict=True):
            self.aggregator.update(self.storage, sample, coord)

    def finalize(self) -> AggregatorR:
        return self.aggregator.finalize(self.storage)

    def resize_to_source(self, image: NDArray[DType], **kwargs: Any) -> pyvips.Image:
        """Resize a mask array to the original source resolution using pyvips.

        Args:
            image: Array of shape (C, H_mask, W_mask) to be resized to (H_source, W_source, C).
            kwargs: Additional arguments passed to pyvips resize (e.g., kernel="nearest").
        """
        import pyvips

        vips_image = pyvips.Image.new_from_array(image.transpose(1, 2, 0))

        scale_factors = self.source_extents / self.mask_extents
        vips_image = vips_image.resize(
            scale_factors[1], vscale=scale_factors[0], **kwargs
        )
        vips_image = vips_image.crop(
            0, 0, self.source_extents[1], self.source_extents[0]
        )

        return vips_image

    def cleanup(self) -> None:
        if hasattr(self, "storage"):
            if hasattr(self.storage, "close"):
                self.storage.close()
            del self.storage

        self.aggregator.cleanup()

aggregator = aggregation(storage=(self.storage), **kwargs) instance-attribute

mask_extents = self.span * self.output_tile_extent // gcd instance-attribute

mask_stride = self.stride * self.output_tile_extent // gcd instance-attribute

output_tile_extent = np.broadcast_to(output_tile_extent, len(source_extents)) instance-attribute

source_extents = np.asarray(source_extents, dtype=(np.int64)) instance-attribute

source_tile_extent = np.broadcast_to(source_tile_extent, len(source_extents)) instance-attribute

span = (num_tiles - 1) * self.stride + self.source_tile_extent instance-attribute

storage = storage(shape=(n_channels, *(self.mask_extents)), dtype=dtype, **kwargs) instance-attribute

stride = np.broadcast_to(stride, len(source_extents)) instance-attribute

upscale_factor = self.source_tile_extent // gcd instance-attribute

__init__(source_extents, source_tile_extent, output_tile_extent, stride, n_channels=1, storage='inmemory', aggregation=MeanAggregator, dtype=np.float32, **kwargs)

Initialize the mask builder.

Parameters:

Name Type Description Default
source_extents tuple[int, ...]

Spatial dimensions (H, W, ...) of the source WSI.

 required
source_tile_extent int | tuple[int, ...]

Spatial dimensions of the model input tiles.

 required
output_tile_extent int | tuple[int, ...]

Spatial dimensions of the model output tiles.

 required
stride int | tuple[int, ...]

Stride between tiles in source resolution.

 required
n_channels int

Number of channels in the output mask.

 1
storage Literal['inmemory', 'memmap'] | Callable[..., NDArray[DType]]

Strategy for allocating memory ("inmemory", "memmap", or a class).

 'inmemory'
aggregation type[Aggregator[DType, AggregatorR]]

Strategy for combining tiles ("mean", "max", or a class).

 MeanAggregator
dtype type[DType]

Data type for the accumulator.

 float32
kwargs Any

Extra arguments passed to storage and aggregation initialization.

 {}
Source code in ratiopath/masks/mask_builders/mask_builder.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def __init__(
    self,
    source_extents: tuple[int, ...],
    source_tile_extent: int | tuple[int, ...],
    output_tile_extent: int | tuple[int, ...],
    stride: int | tuple[int, ...],
    n_channels: int = 1,
    storage: Literal["inmemory", "memmap"]
    | Callable[..., NDArray[DType]] = "inmemory",
    aggregation: type[Aggregator[DType, AggregatorR]] = MeanAggregator,  # type: ignore[assignment]
    dtype: type[DType] = np.float32,  # type: ignore[assignment]
    **kwargs: Any,
) -> None:
    """Initialize the mask builder.

    Args:
        source_extents: Spatial dimensions (H, W, ...) of the source WSI.
        source_tile_extent: Spatial dimensions of the model input tiles.
        output_tile_extent: Spatial dimensions of the model output tiles.
        stride: Stride between tiles in source resolution.
        n_channels: Number of channels in the output mask.
        storage: Strategy for allocating memory ("inmemory", "memmap", or a class).
        aggregation: Strategy for combining tiles ("mean", "max", or a class).
        dtype: Data type for the accumulator.
        kwargs: Extra arguments passed to storage and aggregation initialization.
    """
    # Normalize spatial dimensions to arrays
    self.source_extents = np.asarray(source_extents, dtype=np.int64)
    self.source_tile_extent = np.broadcast_to(
        source_tile_extent, len(source_extents)
    )
    self.output_tile_extent = np.broadcast_to(
        output_tile_extent, len(source_extents)
    )
    self.stride = np.broadcast_to(stride, len(source_extents))

    # Calculate how many tiles are required to fully cover the WSI (implicitly padding the edges)
    # Using np.ceil to ensure the right/bottom edges are fully covered.
    num_tiles = (
        np.ceil(
            np.maximum(0, self.source_extents - self.source_tile_extent)
            / self.stride
        ).astype(np.int64)
        + 1
    )

    # Calculate the actual spatial span these tiles cover in the original WSI space
    self.span = (num_tiles - 1) * self.stride + self.source_tile_extent

    # Find the Greatest Common Divisor for the alignment math
    gcd = np.gcd(self.source_tile_extent, self.stride * self.output_tile_extent)

    # Calculate the required MaskBuilder properties
    self.mask_extents = (self.span * self.output_tile_extent) // gcd
    self.upscale_factor = self.source_tile_extent // gcd
    self.mask_stride = (self.stride * self.output_tile_extent) // gcd

    # Initialize Storage
    if isinstance(storage, str):
        if storage == "inmemory":
            from ratiopath.masks.mask_builders.storage import InMemory

            storage = InMemory
        elif storage == "memmap":
            from ratiopath.masks.mask_builders.storage import MemMap

            storage = MemMap
        else:
            raise ValueError(f"Unknown storage type: {storage}")

    self.storage = storage(
        shape=(n_channels, *self.mask_extents), dtype=dtype, **kwargs
    )

    # Initialize Aggregator
    self.aggregator = aggregation(storage=self.storage, **kwargs)

cleanup()

Source code in ratiopath/masks/mask_builders/mask_builder.py 
228
229
230
231
232
233
234
def cleanup(self) -> None:
    if hasattr(self, "storage"):
        if hasattr(self.storage, "close"):
            self.storage.close()
        del self.storage

    self.aggregator.cleanup()

finalize()

Source code in ratiopath/masks/mask_builders/mask_builder.py 
204
205
def finalize(self) -> AggregatorR:
    return self.aggregator.finalize(self.storage)

resize_to_source(image, **kwargs)

Resize a mask array to the original source resolution using pyvips.

Parameters:

Name Type Description Default
image NDArray[DType]

Array of shape (C, H_mask, W_mask) to be resized to (H_source, W_source, C).

 required
kwargs Any

Additional arguments passed to pyvips resize (e.g., kernel="nearest").

 {}
Source code in ratiopath/masks/mask_builders/mask_builder.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def resize_to_source(self, image: NDArray[DType], **kwargs: Any) -> pyvips.Image:
    """Resize a mask array to the original source resolution using pyvips.

    Args:
        image: Array of shape (C, H_mask, W_mask) to be resized to (H_source, W_source, C).
        kwargs: Additional arguments passed to pyvips resize (e.g., kernel="nearest").
    """
    import pyvips

    vips_image = pyvips.Image.new_from_array(image.transpose(1, 2, 0))

    scale_factors = self.source_extents / self.mask_extents
    vips_image = vips_image.resize(
        scale_factors[1], vscale=scale_factors[0], **kwargs
    )
    vips_image = vips_image.crop(
        0, 0, self.source_extents[1], self.source_extents[0]
    )

    return vips_image

update_batch(batch, coords, edge_clipping=0)

Update the accumulator with a batch of tiles.

Parameters:

Name Type Description Default
batch NDArray[DType]

Array of shape (B, C, *SpatialDims) or (B, C) containing B tiles.

 required
coords NDArray[int64]

Array of shape (B, N) containing top-left coordinates in source resolution.

 required
edge_clipping EdgeClipping

Pixels to clip from tile edges. Supports an int (symmetric for all dims), a tuple of N ints (symmetric per dim), or a tuple of N (start, end) pairs.

 0
Source code in ratiopath/masks/mask_builders/mask_builder.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
def update_batch(
    self,
    batch: NDArray[DType],
    coords: NDArray[np.int64],
    edge_clipping: EdgeClipping = 0,
) -> None:
    """Update the accumulator with a batch of tiles.

    Args:
        batch: Array of shape (B, C, *SpatialDims) or (B, C) containing B tiles.
        coords: Array of shape (B, N) containing top-left coordinates in source resolution.
        edge_clipping: Pixels to clip from tile edges.
            Supports an int (symmetric for all dims), a tuple of N ints
            (symmetric per dim), or a tuple of N (start, end) pairs.
    """
    # Scale coordinates from source to mask resolution
    # Find which tile index this corresponds to and multiply by the mask stride
    # to get the starting pixel in the mask array
    coords = (coords // self.stride) * self.mask_stride

    # Handle scalar data expansion
    # Broadcast scalar (B, C) to (B, C, *output_tile_extent)
    # If already dense, this is a fast view operation
    num_missing_dims = len(self.mask_extents) + 2 - batch.ndim
    batch = np.broadcast_to(
        batch[..., *[np.newaxis] * num_missing_dims],
        (*batch.shape[:2], *self.output_tile_extent),
    )

    # Apply Edge Clipping
    slices, shift = _prepare_clipping(
        len(self.mask_extents), tuple(self.output_tile_extent), edge_clipping
    )
    batch = batch[slices]
    coords += shift * self.upscale_factor

    # Upscale the batch to match the mask resolution if needed
    for axis_idx, factor in enumerate(self.upscale_factor, start=2):
        if factor > 1:
            batch = np.repeat(batch, factor, axis=axis_idx)

    for sample, coord in zip(batch, coords, strict=True):
        self.aggregator.update(self.storage, sample, coord)

The MaskBuilder is the central orchestrator. You configure it by providing:

  • source_extents: Spatial dimensions of the source WSI (H, W, ...).
  • source_tile_extent: Spatial dimensions of the model input tiles.
  • output_tile_extent: Spatial dimensions of the model output tiles (can differ from input due to pooling/stride).
  • stride: Stride between tiles in source resolution.
  • storage: Where the mask is stored — "inmemory" (RAM) or "memmap" (disk-backed).
  • aggregation: How overlapping tiles are merged — MeanAggregator (default) or MaxAggregator.

The mask shape is computed automatically from the source extents, tile extents, and stride using GCD-based compression for efficient memory use.

Components

Storage Strategies

Bases: ndarray

Storage allocator that uses in-memory NumPy arrays.

Source code in ratiopath/masks/mask_builders/storage.py 
13
14
15
16
17
class InMemory[DType: np.generic](np.ndarray):
    """Storage allocator that uses in-memory NumPy arrays."""

    def __new__(cls, shape: tuple[int, ...], dtype: type[DType], **kwargs) -> Self:
        return np.zeros(shape=shape, dtype=dtype).view(cls)

__new__(shape, dtype, **kwargs)

Source code in ratiopath/masks/mask_builders/storage.py 
16
17
def __new__(cls, shape: tuple[int, ...], dtype: type[DType], **kwargs) -> Self:
    return np.zeros(shape=shape, dtype=dtype).view(cls)

Bases: memmap

Storage allocator that uses numpy memory-mapped files (memmaps).

This class provides disk-backed storage for large masks that exceed available RAM. Memory mapping allows the OS to manage paging between disk and memory transparently, enabling processing of masks that would otherwise cause out-of-memory errors.

Temporary Files (default behavior when filename=None): A temporary file is created and used as backing storage. The file is deleted when the memmap is closed or garbage collected. Disk space is consumed during processing but automatically reclaimed afterward.

Explicit Files (when filename is provided): The memmap is backed by the specified file path, which persists after processing. This is useful for caching results or processing masks too large for temporary storage. If the file already exists, a FileExistsError is raised to prevent accidental data loss.

This class uses NumPy's NPY format version 3.0 for compatibility with large arrays (>4GB).

Source code in ratiopath/masks/mask_builders/storage.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class MemMap[DType: np.generic](np.memmap):
    """Storage allocator that uses numpy memory-mapped files (memmaps).

    This class provides disk-backed storage for large masks that exceed available RAM.
    Memory mapping allows the OS to manage paging between disk and memory transparently,
    enabling processing of masks that would otherwise cause out-of-memory errors.

    **Temporary Files (default behavior when `filename=None`):**
    A temporary file is created and used as backing storage. The file is deleted when
    the memmap is closed or garbage collected. Disk space is consumed during processing
    but automatically reclaimed afterward.

    **Explicit Files (when `filename` is provided):**
    The memmap is backed by the specified file path, which persists after processing.
    This is useful for caching results or processing masks too large for temporary storage.
    If the file already exists, a FileExistsError is raised to prevent accidental data loss.

    This class uses NumPy's NPY format version 3.0 for compatibility with large arrays (>4GB).
    """

    def __new__(
        cls,
        shape: tuple[int, ...],
        dtype: type[DType],
        filename: str | PathLike[Any] | None = None,
        **kwargs,
    ) -> Self:
        temp_path = None
        if filename is None:
            # delete=False ensures Windows doesn't lock the file prematurely
            with tempfile.NamedTemporaryFile(delete=False) as file:
                temp_path = filename = file.name
        elif Path(filename).exists():
            raise FileExistsError(f"Memmap {filename} already exists.")

        obj = np.lib.format.open_memmap(
            filename, mode="w+", shape=shape, dtype=dtype, version=(3, 0)
        ).view(cls)

        obj._tempfile = temp_path
        return obj

    def __array_finalize__(self, obj: Any) -> None:
        """Called automatically when a view or slice of the array is created."""
        if obj is None:
            return

        # CRITICAL: If someone slices this array (e.g., subset = my_memmap[:5]),
        # the slice should NOT own the tempfile. Only the original object should
        # delete the file during garbage collection.
        self._tempfile = None

    def close(self) -> None:
        if hasattr(self, "_mmap") and self._mmap is not None:
            try:
                self._mmap.close()
            except Exception as e:
                logger.warning(f"Failed to close memmap: {e}")

        if self._tempfile is not None:
            try:
                Path(self._tempfile).unlink(missing_ok=True)
            except Exception as e:
                logger.warning(f"Failed to delete memmap file {self._tempfile}: {e}")
            self._tempfile = None

    def __del__(self) -> None:
        self.close()

__array_finalize__(obj)

Called automatically when a view or slice of the array is created.

Source code in ratiopath/masks/mask_builders/storage.py 
62
63
64
65
66
67
68
69
70
def __array_finalize__(self, obj: Any) -> None:
    """Called automatically when a view or slice of the array is created."""
    if obj is None:
        return

    # CRITICAL: If someone slices this array (e.g., subset = my_memmap[:5]),
    # the slice should NOT own the tempfile. Only the original object should
    # delete the file during garbage collection.
    self._tempfile = None

__del__()

Source code in ratiopath/masks/mask_builders/storage.py 
86
87
def __del__(self) -> None:
    self.close()

__new__(shape, dtype, filename=None, **kwargs)

Source code in ratiopath/masks/mask_builders/storage.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __new__(
    cls,
    shape: tuple[int, ...],
    dtype: type[DType],
    filename: str | PathLike[Any] | None = None,
    **kwargs,
) -> Self:
    temp_path = None
    if filename is None:
        # delete=False ensures Windows doesn't lock the file prematurely
        with tempfile.NamedTemporaryFile(delete=False) as file:
            temp_path = filename = file.name
    elif Path(filename).exists():
        raise FileExistsError(f"Memmap {filename} already exists.")

    obj = np.lib.format.open_memmap(
        filename, mode="w+", shape=shape, dtype=dtype, version=(3, 0)
    ).view(cls)

    obj._tempfile = temp_path
    return obj

close()

Source code in ratiopath/masks/mask_builders/storage.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
def close(self) -> None:
    if hasattr(self, "_mmap") and self._mmap is not None:
        try:
            self._mmap.close()
        except Exception as e:
            logger.warning(f"Failed to close memmap: {e}")

    if self._tempfile is not None:
        try:
            Path(self._tempfile).unlink(missing_ok=True)
        except Exception as e:
            logger.warning(f"Failed to delete memmap file {self._tempfile}: {e}")
        self._tempfile = None

Aggregation Strategies

Bases: Aggregator[DType, MeanAggregatorResults[DType]]

Aggregator that implements averaging aggregation for overlapping tiles.

This aggregator accumulates tiles by addition and tracks the overlap count at each pixel. During finalization, the accumulated values are divided by the overlap count to compute the average value at each position. This is useful for: - Smoothly blending overlapping tile predictions - Reducing edge artifacts in sliding window processing - Computing ensemble averages from multiple passes

The aggregator allocates an additional overlap_counter accumulator with shape (1, *SpatialDims) to track how many tiles contributed to each pixel position.

Source code in ratiopath/masks/mask_builders/aggregation.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class MeanAggregator[DType: np.generic](
    Aggregator[DType, MeanAggregatorResults[DType]]
):
    """Aggregator that implements averaging aggregation for overlapping tiles.

    This aggregator accumulates tiles by addition and tracks the overlap count at each pixel.
    During finalization, the accumulated values are divided by the overlap count to compute
    the average value at each position. This is useful for:
    - Smoothly blending overlapping tile predictions
    - Reducing edge artifacts in sliding window processing
    - Computing ensemble averages from multiple passes

    The aggregator allocates an additional `overlap_counter` accumulator with shape (1, *SpatialDims)
    to track how many tiles contributed to each pixel position.
    """

    def __init__(
        self,
        storage: NDArray[DType],
        filename: Path | str | None = None,
        overlap_counter_filename: Path | str | None = None,
        **kwargs: Any,
    ) -> None:
        overlap_filename = overlap_counter_filename
        if overlap_filename is None and filename is not None:
            path = Path(filename)
            overlap_filename = path.with_suffix(f".overlaps{path.suffix}")

        storage_cls = cast("Callable[..., NDArray[np.uint16]]", type(storage))
        self.overlap_counter = storage_cls(
            filename=overlap_filename,
            shape=(1, *storage.shape[1:]),
            dtype=np.uint16,
            **kwargs,
        )

    def update(
        self, accumulator: NDArray[DType], sample: np.ndarray, coords: NDArray[np.int64]
    ) -> None:
        mask_tile_extents = np.asarray(sample.shape[1:], dtype=np.int64)
        acc_slices = self._get_acc_slices(coords, mask_tile_extents)
        accumulator[:, *acc_slices] += sample  # type: ignore[misc]
        self.overlap_counter[:, *acc_slices] += 1

    def finalize(self, accumulator: NDArray[DType]) -> MeanAggregatorResults[DType]:
        accumulator /= self.overlap_counter.clip(min=1)  # type: ignore[misc]
        return {
            "mask": accumulator,
            "overlap_counter": self.overlap_counter,
        }

    def cleanup(self) -> None:
        if hasattr(self, "overlap_counter"):
            if hasattr(self.overlap_counter, "close"):
                self.overlap_counter.close()
            del self.overlap_counter

overlap_counter = storage_cls(filename=overlap_filename, shape=(1, *(storage.shape[1:])), dtype=(np.uint16), **kwargs) instance-attribute

__init__(storage, filename=None, overlap_counter_filename=None, **kwargs)

Source code in ratiopath/masks/mask_builders/aggregation.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(
    self,
    storage: NDArray[DType],
    filename: Path | str | None = None,
    overlap_counter_filename: Path | str | None = None,
    **kwargs: Any,
) -> None:
    overlap_filename = overlap_counter_filename
    if overlap_filename is None and filename is not None:
        path = Path(filename)
        overlap_filename = path.with_suffix(f".overlaps{path.suffix}")

    storage_cls = cast("Callable[..., NDArray[np.uint16]]", type(storage))
    self.overlap_counter = storage_cls(
        filename=overlap_filename,
        shape=(1, *storage.shape[1:]),
        dtype=np.uint16,
        **kwargs,
    )

cleanup()

Source code in ratiopath/masks/mask_builders/aggregation.py 
110
111
112
113
114
def cleanup(self) -> None:
    if hasattr(self, "overlap_counter"):
        if hasattr(self.overlap_counter, "close"):
            self.overlap_counter.close()
        del self.overlap_counter

finalize(accumulator)

Source code in ratiopath/masks/mask_builders/aggregation.py 
103
104
105
106
107
108
def finalize(self, accumulator: NDArray[DType]) -> MeanAggregatorResults[DType]:
    accumulator /= self.overlap_counter.clip(min=1)  # type: ignore[misc]
    return {
        "mask": accumulator,
        "overlap_counter": self.overlap_counter,
    }

update(accumulator, sample, coords)

Source code in ratiopath/masks/mask_builders/aggregation.py 
 95
 96
 97
 98
 99
100
101
def update(
    self, accumulator: NDArray[DType], sample: np.ndarray, coords: NDArray[np.int64]
) -> None:
    mask_tile_extents = np.asarray(sample.shape[1:], dtype=np.int64)
    acc_slices = self._get_acc_slices(coords, mask_tile_extents)
    accumulator[:, *acc_slices] += sample  # type: ignore[misc]
    self.overlap_counter[:, *acc_slices] += 1

Bases: Aggregator[DType, NDArray[DType]]

Aggregator that implements maximum aggregation for overlapping tiles.

This aggregator keeps only the maximum value at each pixel position when tiles overlap. No additional storage is required, and finalization is a no-op since the accumulator already contains the final max values. This is useful for: - Maximum intensity projection - Keeping the highest confidence prediction across overlapping tiles - Peak detection across multiple scales

Source code in ratiopath/masks/mask_builders/aggregation.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class MaxAggregator[DType: np.generic](Aggregator[DType, NDArray[DType]]):
    """Aggregator that implements maximum aggregation for overlapping tiles.

    This aggregator keeps only the maximum value at each pixel position when tiles overlap.
    No additional storage is required, and finalization is a no-op since the accumulator
    already contains the final max values. This is useful for:
    - Maximum intensity projection
    - Keeping the highest confidence prediction across overlapping tiles
    - Peak detection across multiple scales
    """

    def update(
        self, accumulator: NDArray[DType], sample: np.ndarray, coords: NDArray[np.int64]
    ) -> None:
        mask_tile_extents = np.asarray(sample.shape[1:], dtype=np.int64)
        acc_slices = self._get_acc_slices(coords, mask_tile_extents)
        accumulator[:, *acc_slices] = np.maximum(accumulator[:, *acc_slices], sample)

    def finalize(self, accumulator: NDArray[DType]) -> NDArray[DType]:
        return accumulator

finalize(accumulator)

Source code in ratiopath/masks/mask_builders/aggregation.py 
135
136
def finalize(self, accumulator: NDArray[DType]) -> NDArray[DType]:
    return accumulator

update(accumulator, sample, coords)

Source code in ratiopath/masks/mask_builders/aggregation.py 
128
129
130
131
132
133
def update(
    self, accumulator: NDArray[DType], sample: np.ndarray, coords: NDArray[np.int64]
) -> None:
    mask_tile_extents = np.asarray(sample.shape[1:], dtype=np.int64)
    acc_slices = self._get_acc_slices(coords, mask_tile_extents)
    accumulator[:, *acc_slices] = np.maximum(accumulator[:, *acc_slices], sample)

Examples

Averaging Scalar Predictions

Use case: You have scalar predictions (e.g., class probabilities) for each tile. Each prediction is uniformly expanded to fill the tile's footprint, and overlapping regions are averaged.

import numpy as np
import openslide
from ratiopath.masks.mask_builders import MaskBuilder, MeanAggregator
import matplotlib.pyplot as plt

# Set up tiling parameters
LEVEL = 3
tile_extents = (512, 512)
tile_strides = (256, 256)
slide = openslide.OpenSlide("path/to/slide.mrxs")
slide_w, slide_h = slide.level_dimensions[LEVEL]

# output_tile_extent=(1, 1) means scalar data — the builder
# broadcasts (B, C) → (B, C, 1, 1) and upscales automatically.
mask_builder = MaskBuilder(
    source_extents=(slide_h, slide_w),
    source_tile_extent=tile_extents,
    output_tile_extent=(1, 1),
    stride=tile_strides,
    n_channels=1,
    storage="inmemory",
    aggregation=MeanAggregator,
    dtype=np.float32,
)

# Process tiles
for tiles, xs, ys in generate_tiles_from_slide(slide, LEVEL, tile_extents, tile_strides):
    features = model.predict(tiles)  # features shape: (B, 1)
    coords_batch = np.stack([ys, xs], axis=1)  # shape: (B, 2)
    mask_builder.update_batch(features, coords_batch)

# Finalize — MeanAggregator returns {"mask": ..., "overlap_counter": ...}
results = mask_builder.finalize()
assembled_mask = results["mask"]
overlap_counter = results["overlap_counter"]

plt.imshow(assembled_mask[0], cmap="gray")
plt.show()

# Always clean up to release storage resources
mask_builder.cleanup()

Max Aggregation with Edge Clipping (MemMap)

Use case: You have high-resolution feature maps. You want to preserve the maximum signal where tiles overlap, remove border pixels from each tile edge to avoid artifacts, and use disk storage because the mask is very large.

import numpy as np
from ratiopath.masks.mask_builders import MaskBuilder, MaxAggregator

# Dense output — output tiles match input tiles in spatial size
mask_builder = MaskBuilder(
    source_extents=(10000, 10000),
    source_tile_extent=(512, 512),
    output_tile_extent=(512, 512),
    stride=(256, 256),
    n_channels=3,
    storage="memmap",
    aggregation=MaxAggregator,
    dtype=np.float32,
    filename="large_mask.npy",  # persisted to disk
)

for tiles, coords in tile_generator:
    predictions = model.predict(tiles)  # (B, 3, 512, 512)
    # edge_clipping=4 removes 4px from each edge of every tile
    mask_builder.update_batch(predictions, coords, edge_clipping=4)

# MaxAggregator returns the accumulator NDArray directly
assembled_mask = mask_builder.finalize()
mask_builder.cleanup()

Auto-Scaling Coordinates (Different Input/Output Resolution)

Use case: Your model's output tiles have different spatial dimensions than the input tiles (e.g., due to stride or pooling). The builder auto-scales coordinates between source and mask resolution.

import numpy as np
from ratiopath.masks.mask_builders import MaskBuilder, MeanAggregator

# Model takes 512×512 input tiles, produces 128×128 output tiles (4× downsampled)
mask_builder = MaskBuilder(
    source_extents=(2000, 2000),
    source_tile_extent=(512, 512),
    output_tile_extent=(128, 128),
    stride=(256, 256),
    n_channels=1,
    storage="inmemory",
    aggregation=MeanAggregator,
    dtype=np.float32,
)

# Coordinates are always in SOURCE resolution — the builder
# handles the conversion to mask resolution internally.
for tiles, coords in tile_generator:
    predictions = model.predict(tiles)  # (B, 1, 128, 128)
    mask_builder.update_batch(predictions, coords)

results = mask_builder.finalize()
mask_builder.cleanup()

Coordinate System Notes

All mask builders expect coordinates in the format (B, N) where:

  • B is the batch size.
  • N is the number of spatial dimensions (typically 2 for height and width).

Note the order: [ys, xs] not [xs, ys], as the first dimension represents height (y) and the second represents width (x), matching the NumPy (C, H, W) convention used by the builder.

Lifecycle

Always call cleanup() when you are done with a MaskBuilder to release storage resources (especially important for MemMap storage which holds file handles):

mask_builder = MaskBuilder(...)
# ... update_batch calls ...
results = mask_builder.finalize()
mask_builder.cleanup()