tiler

 10class Tiler:
 11    TILING_MODES = ["constant", "drop", "irregular", "reflect", "edge", "wrap"]
 12    r"""
 13    Supported tiling modes:
 14    - `constant` (default)
 15        If a tile is smaller than `tile_shape`, pad it with the constant value along each axis to match `tile_shape`.
 16        Set the value with the keyword `constant_value`.
 17    - `drop`
 18        If a tile is smaller than `tile_shape` in any of the dimensions, ignore it. Can result in zero tiles.
 19    - `irregular`
 20        Allow tiles to be smaller than `tile_shape`.
 21    - `reflect`
 22        If a tile is smaller than `tile_shape`,
 23        pad it with the reflection of values along each axis to match `tile_shape`.
 24    - `edge`
 25        If a tile is smaller than `tile_shape`,
 26        pad it with the edge values of data along each axis to match `tile_shape`.
 27    - `wrap`
 28        If a tile is smaller than `tile_shape`,
 29        pad it with the wrap of the vector along each axis to match `tile_shape`.
 30        The first values are used to pad the end and the end values are used to pad the beginning.
 31    """
 32
 33    def __init__(
 34        self,
 35        data_shape: Union[Tuple, List, np.ndarray],
 36        tile_shape: Union[Tuple, List, np.ndarray],
 37        overlap: Union[int, float, Tuple, List, np.ndarray] = 0,
 38        channel_dimension: Optional[int] = None,
 39        mode: str = "constant",
 40        constant_value: float = 0.0,
 41    ):
 42        """Tiler class precomputes everything for tiling with specified parameters, without actually slicing data.
 43        You can access tiles individually with `Tiler.get_tile()` or with an iterator, both individually and in batches,
 44        with `Tiler.iterate()` (or the alias `Tiler.__call__()`).
 45
 46        Args:
 47            data_shape (tuple, list or np.ndarray): Input data shape.
 48                If there is a channel dimension, it should be included in the shape.
 49                For example, (1920, 1080, 3), [512, 512, 512] or np.ndarray([3, 1024, 768]).
 50
 51            tile_shape (tuple, list or np.ndarray): Shape of a tile.
 52                Tile must have the same number of dimensions as data or less.
 53                If less, the shape will be automatically prepended with ones to match data_shape size.
 54                For example,  (256, 256, 3), [64, 64, 64] or np.ndarray([3, 128, 128]).
 55
 56            overlap (int, float, tuple, list or np.ndarray): Specifies overlap between tiles.
 57                If integer, the same overlap of overlap pixels applied in each dimension, except channel_dimension.
 58                If float, percentage of a tile_shape to overlap [0.0, 1.0), from 0% to 100% non-inclusive, except
 59                channel_dimension.
 60                If tuple, list or np.ndarray, explicit size of the overlap (must be smaller than tile_shape in each
 61                dimension).
 62                Default is `0`.
 63
 64            channel_dimension (int, optional): Specifies which axis is the channel dimension that will not be tiled.
 65                Usually it is the last or the first dimension of the array.
 66                Negative indexing (`-len(data_shape)` to `-1` inclusive) is allowed.
 67                Default is `None`, no channel dimension in the data.
 68
 69            mode (str): Defines how the data will be tiled.
 70                Must be one of the supported `Tiler.TILING_MODES`. Defaults to `"constant"`.
 71
 72            constant_value (float): Specifies the value of padding when `mode='constant'`.
 73                Default is `0.0`.
 74        """
 75
 76        self.recalculate(
 77            data_shape=data_shape,
 78            tile_shape=tile_shape,
 79            overlap=overlap,
 80            channel_dimension=channel_dimension,
 81            mode=mode,
 82            constant_value=constant_value,
 83        )
 84
 85    def recalculate(
 86        self,
 87        data_shape: Optional[Union[Tuple, List, np.ndarray]] = None,
 88        tile_shape: Optional[Union[Tuple, List, np.ndarray]] = None,
 89        overlap: Optional[Union[int, float, Tuple, List, np.ndarray]] = None,
 90        channel_dimension: Optional[int] = None,
 91        mode: Optional[str] = None,
 92        constant_value: Optional[float] = None,
 93    ) -> None:
 94        """Recalculates tiling for new given settings.
 95        If a passed value is None, use previously given value.
 96
 97        For more information about each argument see `Tiler.__init__()` documentation.
 98        """
 99
100        # Data and tile shapes
101        if data_shape is not None:
102            self.data_shape = np.asarray(data_shape, dtype=np.int64)
103        if tile_shape is not None:
104            self.tile_shape = np.atleast_1d(np.asarray(tile_shape, dtype=np.int64))
105
106            # Append ones to match data_shape size
107            if self.tile_shape.size < self.data_shape.size:
108                size_difference = self.data_shape.size - self.tile_shape.size
109                self.tile_shape = np.insert(arr=self.tile_shape, obj=0, values=np.ones(size_difference), axis=0)
110                warnings.warn(
111                    f"Tiler automatically adjusted tile_shape from {tuple(tile_shape)} to {tuple(self.tile_shape)}.",
112                    stacklevel=2,
113                )
114        self._n_dim: int = len(self.data_shape)
115        if (self.tile_shape <= 0).any() or (self.data_shape <= 0).any():
116            raise ValueError("Tile and data shapes must be tuple, list or ndarray of positive integers.")
117        if self.tile_shape.size != self.data_shape.size:
118            raise ValueError(
119                "Tile shape must have less or equal number of elements compared to the data shape. "
120                "If less, your tile shape will be prepended with ones to match the data shape, "
121                "e.g. data_shape=(28, 28), tile_shape=(28) -> tile_shape=(1, 28)."
122            )
123
124        # Tiling mode
125        if mode is not None:
126            self.mode = mode
127        if self.mode not in self.TILING_MODES:
128            raise ValueError(f"{self.mode} is an unsupported tiling mode, please check the documentation.")
129
130        # Constant value used for constant tiling mode
131        if constant_value is not None:
132            self.constant_value = constant_value
133
134        # Channel dimension can be None which means we need to check for init too
135        if not hasattr(self, "channel_dimension") or channel_dimension is not None:
136            self.channel_dimension = channel_dimension
137        if self.channel_dimension:
138            if (self.channel_dimension >= self._n_dim) or (self.channel_dimension < -self._n_dim):
139                raise ValueError(
140                    f"Specified channel dimension is out of bounds "
141                    f"(should be None or an integer from {-self._n_dim} to {self._n_dim - 1})."
142                )
143            if self.channel_dimension < 0:
144                # negative indexing
145                self.channel_dimension = self._n_dim + self.channel_dimension
146
147        # Overlap and step
148        if overlap is not None:
149            self.overlap = overlap
150        if isinstance(self.overlap, float):
151            if self.overlap < 0 or self.overlap >= 1.0:
152                raise ValueError("Float overlap must be in range of [0.0, 1.0) i.e. [0%, 100%).")
153
154            self._tile_overlap: np.ndarray = np.ceil(self.overlap * self.tile_shape).astype(int)
155            if self.channel_dimension is not None:
156                self._tile_overlap[self.channel_dimension] = 0
157
158        elif isinstance(self.overlap, int):
159            tile_shape_without_channel = self.tile_shape[np.arange(self._n_dim) != self.channel_dimension]
160            if self.overlap < 0 or np.any(self.overlap >= tile_shape_without_channel):
161                raise ValueError(f"Integer overlap must be in range of 0 to {np.max(tile_shape_without_channel)}")
162
163            self._tile_overlap: np.ndarray = np.array([self.overlap for _ in self.tile_shape])
164            if self.channel_dimension is not None:
165                self._tile_overlap[self.channel_dimension] = 0
166
167        elif isinstance(self.overlap, (list, tuple, np.ndarray)):
168            if np.any(np.array(self.overlap) < 0) or np.any(self.overlap >= self.tile_shape):
169                raise ValueError("Overlap size must be smaller than tile_shape.")
170
171            self._tile_overlap: np.ndarray = np.array(self.overlap).astype(int)
172
173        else:
174            raise ValueError("Unsupported overlap mode (not float, int, list, tuple or np.ndarray).")
175
176        self._tile_step: np.ndarray = (self.tile_shape - self._tile_overlap).astype(int)  # tile step
177
178        # Calculate mosaic (collection of tiles) shape
179        div, mod = np.divmod(
180            [self.data_shape[d] - self._tile_overlap[d] for d in range(self._n_dim)],
181            self._tile_step,
182        )
183        if self.mode == "drop":
184            self._indexing_shape = div
185        else:
186            self._indexing_shape = div + (mod != 0)
187        if self.channel_dimension is not None:
188            self._indexing_shape[self.channel_dimension] = 1
189
190        # Calculate new shape assuming tiles are padded
191        if self.mode == "irregular":
192            self._new_shape = self.data_shape
193        else:
194            self._new_shape = (self._indexing_shape * self._tile_step) + self._tile_overlap
195        self._shape_diff = self._new_shape - self.data_shape
196        if self.channel_dimension is not None:
197            self._shape_diff[self.channel_dimension] = 0
198
199        # If channel dimension is given, set tile_step of that dimension to 0
200        if self.channel_dimension is not None:
201            self._tile_step[self.channel_dimension] = 0
202
203        # Tile indexing
204        self._tile_index = np.vstack(np.meshgrid(*[np.arange(0, x) for x in self._indexing_shape], indexing="ij"))
205        self._tile_index = self._tile_index.reshape(self._n_dim, -1).T
206        self.n_tiles = len(self._tile_index)
207
208        if self.n_tiles == 0:
209            warnings.warn(
210                f"Tiler (mode={mode}, overlap={overlap}) will split data_shape {data_shape} "
211                f"into zero tiles (tile_shape={tile_shape}).",
212                stacklevel=2,
213            )
214
215    def __len__(self) -> int:
216        """
217        Returns:
218             int: Number of tiles in the mosaic.
219        """
220        return self.n_tiles
221
222    def __repr__(self) -> str:
223        """
224        Returns:
225            str: String representation of the object.
226        """
227        return (
228            f"Tiler split {self.data_shape.tolist()} data into {len(self)} tiles of {self.tile_shape.tolist()}."
229            f"\n\tMosaic shape: {self._indexing_shape.tolist()}"
230            f"\n\tPadded shape: {self._new_shape.tolist()}"
231            f"\n\tTile overlap: {self.overlap}"
232            f"\n\tElement step: {self._tile_step.tolist()}"
233            f"\n\tMode: {self.mode}"
234            f"\n\tChannel dimension: {self.channel_dimension}"
235        )
236
237    def __call__(
238        self,
239        data: Union[np.ndarray, Callable[..., np.ndarray]],
240        progress_bar: bool = False,
241        batch_size: int = 0,
242        drop_last: bool = False,
243        copy_data: bool = True,
244    ) -> Generator[Tuple[int, np.ndarray], None, None]:
245        """Alias for `Tiler.iterate()`"""
246        return self.iterate(data, progress_bar, batch_size, drop_last, copy_data)
247
248    def iterate(
249        self,
250        data: Union[np.ndarray, Callable[..., np.ndarray]],
251        progress_bar: bool = False,
252        batch_size: int = 0,
253        drop_last: bool = False,
254        copy_data: bool = True,
255    ) -> Generator[Tuple[int, np.ndarray], None, None]:
256        """Iterates through tiles of the given data array. This method can also be accessed by `Tiler.__call__()`.
257
258        Args:
259            data (np.ndarray or callable): The data array on which the tiling will be performed. A callable can be
260                supplied to load data into memory instead of slicing from an array. The callable should take integers
261                as input, the smallest tile corner coordinates and tile size in each dimension, and output numpy array.
262
263                e.g.
264                *python-bioformats*
265                ```python
266                >>> tileSize = 2000
267                >>> tiler = Tiler((sizeX, sizeY, sizeC), (tileSize, tileSize, sizeC))
268                >>> def reader_func(*args):
269                >>>     X, Y, W, H = args[0], args[1], args[3], args[4]
270                >>>     return reader.read(XYWH=[X, Y, W, H])
271                >>> for t_id, tile in tiler.iterate(reader_func):
272                >>>     pass
273                ```
274                *open-slide*
275                ```python
276                >>> reader_func = lambda *args: wsi.read_region([args[0], args[1]], 0, [args[3], args[4]])
277                >>> for t_id, tile in tiler.iterate(reader_func):
278                >>>     pass
279                ```
280
281            progress_bar (bool): Specifies whether to show the progress bar or not.
282                Uses `tqdm` package.
283                Default is `False`.
284
285            batch_size (int): Specifies returned batch size.
286                If `batch_size == 0`, return one tile at a time.
287                If `batch_size >= 1`, return in batches (returned shape: `[batch_size, *tile_shape]`).
288                Default is 0.
289
290            drop_last (bool): Specifies whether to drop last non-full batch.
291                Used only when batch_size > 0.
292                Default is False.
293
294            copy_data (bool): Specifies whether to copy the tile before returning it.
295                If `copy_data == False`, returns a view.
296                Default is True.
297
298        Yields:
299            (int, np.ndarray): Tuple with integer tile number and array tile data.
300        """
301
302        if batch_size < 0:
303            raise ValueError(f"Batch size must >= 0, not {batch_size}")
304
305        # return a tile at a time
306        if batch_size == 0:
307            for tile_i in tqdm(range(self.n_tiles), disable=not progress_bar, unit=" tiles"):
308                yield tile_i, self.get_tile(data, tile_i, copy_data=copy_data)
309
310        # return in batches
311        if batch_size > 0:
312            # check for drop_last
313            length = (self.n_tiles - (self.n_tiles % batch_size)) if drop_last else self.n_tiles
314
315            for tile_i in tqdm(range(0, length, batch_size), disable=not progress_bar, unit=" batches"):
316                tiles = np.stack(
317                    [
318                        self.get_tile(data, x, copy_data=copy_data)
319                        for x in range(tile_i, min(tile_i + batch_size, length))
320                    ]
321                )
322                yield tile_i // batch_size, tiles
323
324    def get_tile(
325        self,
326        data: Union[np.ndarray, Callable[..., np.ndarray]],
327        tile_id: int,
328        copy_data: bool = True,
329    ) -> np.ndarray:
330        """Returns an individual tile.
331
332        Args:
333            data (np.ndarray or callable): Data from which `tile_id`-th tile will be taken. A callable can be
334                supplied to load data into memory instead of slicing from an array. The callable should take integers
335                as input, the smallest tile corner coordinates and tile size in each dimension, and output numpy array.
336
337                e.g.
338                *python-bioformats*
339                ```python
340                >>> tileSize = 2000
341                >>> tiler = Tiler((sizeX, sizeY, sizeC), (tileSize, tileSize, sizeC))
342                >>> def reader_func(*args):
343                >>>     X, Y, W, H = args[0], args[1], args[3], args[4]
344                >>>     return reader.read(XYWH=[X, Y, W, H])
345                >>> tiler.get_tile(reader_func, 0)
346                ```
347                *open-slide*
348                ```python
349                >>> reader_func = lambda *args: wsi.read_region([args[0], args[1]], 0, [args[3], args[4]])
350                >>> tiler.get_tile(reader_func, 0)
351                ```
352
353            tile_id (int): Specifies which tile to return. Must be smaller than the total number of tiles.
354
355            copy_data (bool): Specifies whether returned tile is a copy.
356                If `copy_data == False` returns a view.
357                Default is True.
358
359        Returns:
360            np.ndarray: Content of tile number `tile_id`, padded if necessary.
361        """
362
363        if (tile_id < 0) or (tile_id >= self.n_tiles):
364            raise IndexError(
365                f"Out of bounds, there is no tile {tile_id}.There are {len(self) - 1} tiles, starting from index 0."
366            )
367
368        if isinstance(data, np.ndarray) and np.not_equal(data.shape, self.data_shape).any():
369            raise ValueError(
370                f"Shape of provided data array ({data.shape}) does not match "
371                f"same as Tiler's data_shape ({tuple(self.data_shape)})."
372            )
373
374        # get tile data
375        tile_corner = self._tile_index[tile_id] * self._tile_step
376        # take the lesser of the tile shape and the distance to the edge
377        sampling = [
378            slice(
379                tile_corner[d],
380                np.min([self.data_shape[d], tile_corner[d] + self.tile_shape[d]]),
381            )
382            for d in range(self._n_dim)
383        ]
384
385        if callable(data):
386            sampling = [x.stop - x.start for x in sampling]
387            tile_data = data(*tile_corner, *sampling)
388        else:
389            tile_data = data[tuple(sampling)]
390
391        if copy_data:
392            tile_data = tile_data.copy()
393
394        shape_diff = self.tile_shape - tile_data.shape
395        if (self.mode != "irregular") and np.any(shape_diff > 0):
396            if self.mode == "constant":
397                tile_data = np.pad(
398                    tile_data,
399                    list((0, diff) for diff in shape_diff),
400                    mode=self.mode,
401                    constant_values=self.constant_value,
402                )
403            elif self.mode == "reflect" or self.mode == "edge" or self.mode == "wrap":
404                tile_data = np.pad(tile_data, list((0, diff) for diff in shape_diff), mode=self.mode)
405
406        return tile_data
407
408    def get_all_tiles(
409        self,
410        data: Union[np.ndarray, Callable[..., np.ndarray]],
411        axis: int = 0,
412        copy_data: bool = True,
413    ) -> np.ndarray:
414        """Returns all tiles joined along a new axis. Does not work for `Tiler.mode = 'irregular'`.
415
416        The `axis` parameter specifies the index of the new axis in the dimensions of the result.
417        For example, if `axis=0` it will be the first dimension and if `axis=-1` it will be the last dimension.
418
419        For more information about `data` and `copy_data` parameters, see `Tiler.get_tile()`.
420
421        Args:
422            data (np.ndarray or callable): Data which will be tiled. A callable can be supplied to load data into memory
423                instead of slicing from an array. The callable should take integers as input, the smallest tile corner
424                coordinates and tile size in each dimension, and output numpy array.
425
426            axis (int): The axis in the result array along which the tiles are stacked.
427
428            copy_data (bool): Specifies whether returned tile is a copy.
429                If `copy_data == False` returns a view.
430                Default is True.
431
432        Returns:
433            np.ndarray: All tiles stacked along a new axis.
434        """
435
436        if self.mode == "irregular":
437            raise ValueError("get_all_tiles does not support irregular mode")
438
439        return np.stack(
440            [self.get_tile(data, x, copy_data=copy_data) for x in range(self.n_tiles)],
441            axis=axis,
442        )
443
444    def get_tile_bbox(
445        self,
446        tile_id: int,
447        with_channel_dim: bool = False,
448        all_corners: bool = False,
449    ) -> Union[Tuple[np.ndarray, np.ndarray], np.ndarray]:
450        """Returns coordinates of the opposite corners of the bounding box (hyperrectangle?) of the tile on padded data.
451
452        Args:
453            tile_id (int): Specifies which tile's bounding coordinates will be returned.
454                Must be between 0 and the total number of tiles.
455
456            with_channel_dim (bool): Specifies whether to return shape with channel dimension or without.
457                Default is False.
458
459            all_corners (bool): If True, returns all vertices of the bounding box.
460                Default is False.
461
462        Returns:
463            (np.ndarray, np.ndarray): Smallest (bottom-left) and largest (top-right) corners of the bounding box.
464
465            np.ndarray: All corners of the bounding box, if `all_corners=True`.
466        """
467
468        if (tile_id < 0) or (tile_id >= self.n_tiles):
469            raise IndexError(
470                f"Out of bounds, there is no tile {tile_id}. There are {len(self) - 1} tiles, starting from index 0."
471            )
472
473        # find min and max vertices
474        bottom_left_corner = self._tile_step * self.get_tile_mosaic_position(tile_id, True)
475        top_right_corner = bottom_left_corner + self.tile_shape
476
477        # remove channel dimension if not required
478        if self.channel_dimension is not None and not with_channel_dim:
479            dim_indices = list(range(self.channel_dimension)) + list(
480                range(self.channel_dimension + 1, len(self._tile_step))
481            )
482            bottom_left_corner = bottom_left_corner[dim_indices]
483            top_right_corner = top_right_corner[dim_indices]
484
485        # by default, return only min/max vertices
486        if not all_corners:
487            return bottom_left_corner, top_right_corner
488
489        # otherwise, return all vertices of the bbox
490        # inspired by https://stackoverflow.com/a/57065356/1668421
491        # but instead create an indexing array from cartesian product of bits
492        # and use it to sample intervals
493        else:
494            n_dim: int = len(bottom_left_corner)  # already channel_dimension adjusted
495            mins = np.minimum(bottom_left_corner, top_right_corner)
496            maxs = np.maximum(bottom_left_corner, top_right_corner)
497            intervals = np.stack([mins, maxs], -1)
498            indexing = np.array(list(itertools.product([0, 1], repeat=n_dim)))
499            corners = np.stack([intervals[x][indexing.T[x]] for x in range(n_dim)], -1)
500            return corners
501
502    def get_tile_mosaic_position(self, tile_id: int, with_channel_dim: bool = False) -> np.ndarray:
503        """Returns tile position in the mosaic.
504
505        Args:
506          tile_id (int): Specifies which tile's mosaic position will be returned. \
507            Must be smaller than the total number of tiles.
508
509          with_channel_dim (bool): Specifies whether to return position with channel dimension or without.
510            Default is False.
511
512        Returns:
513            np.ndarray: Tile mosaic position (tile position relative to other tiles).
514        """
515        if (tile_id < 0) or (tile_id >= self.n_tiles):
516            raise IndexError(
517                f"Out of bounds, there is no tile {tile_id}. There are {len(self) - 1} tiles, starting from index 0."
518            )
519
520        if self.channel_dimension is not None and not with_channel_dim:
521            return self._tile_index[tile_id][~(np.arange(self._n_dim) == self.channel_dimension)]
522        return self._tile_index[tile_id]
523
524    def get_mosaic_shape(self, with_channel_dim: bool = False) -> np.ndarray:
525        """Returns mosaic shape.
526
527        Args:
528            with_channel_dim (bool):
529                Specifies whether to return shape with channel dimension or without. Defaults to False.
530
531        Returns:
532            np.ndarray: Shape of tiles mosaic.
533        """
534        if self.channel_dimension is not None and not with_channel_dim:
535            return self._indexing_shape[~(np.arange(self._n_dim) == self.channel_dimension)]
536        return self._indexing_shape
537
538    def calculate_padding(self) -> Tuple[np.ndarray, List[Tuple[int, int]]]:
539        """Calculate a frame padding for the current Tiler parameters.
540        The padding is overlap//2 or tile_step//2, whichever is bigger.
541        The method returns a tuple (new_shape, padding) where padding is
542        ((before_1, after_1), … (before_N, after_N)), unique pad widths for each axis N.
543
544        In the usual workflow, you'd recalculate tiling settings and then apply padding, prior to tiling.
545        Then when merging, pass padding to `Merger.merge(extra_padding=padding, ...)`:
546        ```python
547        >>> tiler = Tiler(...)
548        >>> merger = Merger(tiler, ...)
549        >>> new_shape, padding = tiler.calculate_padding()
550        >>> tiler.recalculate(data_shape=new_shape)
551        >>> padded_data = np.pad(data, pad_width=padding, mode="reflect")
552        >>> for tile_id, tile in tiler(padded_data):
553        >>>     processed_tile = process(tile)
554        >>>     merger.add(tile_id, processed_tile)
555        >>> final_image = merger.merge(extra_padding=padding)
556        ```
557        Return:
558            np.ndarray: Resulting shape when padding is applied.
559
560            List[Tuple[int, int]]: Calculated padding.
561        """
562
563        # Choosing padding
564        pre_pad = np.maximum(self._tile_step // 2, self._tile_overlap // 2)
565        post_pad = pre_pad + np.mod(self._tile_step, 2)
566
567        new_shape = pre_pad + self.data_shape + post_pad
568        padding = list(zip(pre_pad, post_pad, strict=False))
569
570        return new_shape, padding