bioimage_py.wrapper

27class WrapperSource(Source):
28    """Base class for wrappers that delegate metadata to the wrapped source.
29
30    Subclasses implement :meth:`__getitem__` (the transform) and :meth:`_params` (the
31    keyword arguments needed to rebuild the wrapper). The output dtype may differ from the
32    wrapped source, so subclasses can override :attr:`dtype`.
33    """
34
35    def __init__(self, source: SourceLike) -> None:
36        self._source = as_source(source)
37
38    @property
39    def source(self) -> Source:
40        """The wrapped source."""
41        return self._source
42
43    @property
44    def _wrapped_sources(self) -> Tuple[Source, ...]:
45        """The wrapped source(s). Single-source wrappers wrap exactly one; override for many."""
46        return (self._source,)
47
48    def _setitem(self, roi: Tuple[slice, ...], value: np.ndarray) -> None:
49        raise TypeError("Wrapper sources are read-only.")
50
51    @property
52    def writable(self) -> bool:
53        """Wrapper sources are read-only."""
54        return False
55
56    @property
57    def shape(self) -> Tuple[int, ...]:
58        return self._source.shape
59
60    @property
61    def dtype(self) -> np.dtype:
62        return self._source.dtype
63
64    @property
65    def chunks(self) -> Optional[Tuple[int, ...]]:
66        return self._source.chunks
67
68    @property
69    def shards(self) -> Optional[Tuple[int, ...]]:
70        return self._source.shards
71
72    def _params(self) -> dict:
73        """Return the keyword arguments needed to reconstruct this wrapper."""
74        return {}
75
76    @classmethod
77    def _from_wrapped(cls, sources: Sequence[Source], params: dict) -> "WrapperSource":
78        """Reconstruct the wrapper from its wrapped source(s) and params.
79
80        The default rebuilds a single-source wrapper as ``cls(source, **params)``. Multi-source
81        wrappers override this to consume all reconstructed sources.
82        """
83        return cls(sources[0], **params)
84
85    def to_spec(self) -> SourceSpec:
86        """Return a spec recording the wrapper class, its params, and the wrapped spec(s)."""
87        params = dict(self._params())
88        params["cls"] = type(self).__name__
89        wrapped_specs = [s.to_spec() for s in self._wrapped_sources]
90        # Single-source wrappers store a single spec; multi-source wrappers store a list.
91        wrapped = wrapped_specs[0] if len(wrapped_specs) == 1 else wrapped_specs
92        return SourceSpec(kind="wrapper", params=params, wrapped=wrapped)

108@register_wrapper
109class SimpleTransformationSource(WrapperSource):
110    """Apply a value-only transformation to the wrapped source on read.
111
112    The transformation depends only on the data values, not on coordinates, so its signature is
113    ``transformation(block)``. The callable is captured in the spec (it round-trips via cloudpickle),
114    so prefer picklable callables; concrete wrappers (e.g. :class:`ThresholdSource`) instead store
115    simple parameters and rebuild the callable in their constructor.
116
117    Args:
118        source: The wrapped source-like object.
119        transformation: The value transformation to apply to each read block.
120        with_channels: Whether the wrapped source has a leading channel axis. If set, that axis is
121            hidden from this wrapper's shape and passed through to the transformation on read.
122        dtype: The output dtype. Defaults to the wrapped source's dtype.
123    """
124
125    def __init__(self, source: SourceLike, transformation: Callable, *,
126                 with_channels: bool = False, dtype: Optional[np.dtype] = None) -> None:
127        super().__init__(source)
128        if not callable(transformation):
129            raise ValueError("Expect the transformation to be callable.")
130        self._transformation = transformation
131        self._with_channels = bool(with_channels)
132        self._dtype = None if dtype is None else np.dtype(dtype)
133
134    @property
135    def shape(self) -> Tuple[int, ...]:
136        return self._source.shape[1:] if self._with_channels else self._source.shape
137
138    @property
139    def ndim(self) -> int:
140        return self._source.ndim - 1 if self._with_channels else self._source.ndim
141
142    @property
143    def dtype(self) -> np.dtype:
144        return self._dtype if self._dtype is not None else self._source.dtype
145
146    @property
147    def chunks(self) -> Optional[Tuple[int, ...]]:
148        src_chunks = self._source.chunks
149        if src_chunks is None:
150            return None
151        return src_chunks[1:] if self._with_channels else src_chunks
152
153    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
154        """Return the transformed data at ``roi``."""
155        index = (slice(None),) + tuple(roi) if self._with_channels else roi
156        return self._transformation(self._source[index])
157
158    def _params(self) -> dict:
159        """Return the keyword arguments needed to reconstruct this wrapper."""
160        return {"transformation": self._transformation, "with_channels": self._with_channels,
161                "dtype": self._dtype}

164@register_wrapper
165class SimpleTransformationWithHaloSource(SimpleTransformationSource):
166    """Apply a value-only transformation that needs a halo, reading through it on each block.
167
168    Like :class:`SimpleTransformationSource`, but the read region is extended by ``halo`` on each
169    spatial axis (clamped to the data bounds) before the transformation, and the result is cropped
170    back to the requested region. Use this for transformations whose output near a block boundary
171    depends on neighbouring values (e.g. a local filter).
172
173    Args:
174        source: The wrapped source-like object.
175        transformation: The value transformation to apply to each (haloed) read block.
176        halo: The per-axis halo, with one entry per spatial axis.
177        with_channels: Whether the wrapped source has a leading channel axis (see
178            :class:`SimpleTransformationSource`).
179        dtype: The output dtype. Defaults to the wrapped source's dtype.
180    """
181
182    def __init__(self, source: SourceLike, transformation: Callable, halo: Sequence[int], *,
183                 with_channels: bool = False, dtype: Optional[np.dtype] = None) -> None:
184        super().__init__(source, transformation, with_channels=with_channels, dtype=dtype)
185        halo = tuple(int(h) for h in halo)
186        if len(halo) != self.ndim:
187            raise ValueError(f"Expect halo of length {self.ndim}, got {len(halo)}.")
188        self._halo = halo
189
190    def _extend_halo(self, roi: Tuple[slice, ...]) -> Tuple[Tuple[slice, ...], Tuple[slice, ...]]:
191        """Extend ``roi`` by the halo (clamped to bounds) and return it with the local crop."""
192        shape = self.shape
193        extended_index, local_crop = [], []
194        for sl, ha, sh in zip(roi, self._halo, shape):
195            idx_start = int(sl.start) if sl.start is not None else 0
196            idx_stop = int(sl.stop) if sl.stop is not None else sh
197            start = max(idx_start - ha, 0)
198            stop = min(idx_stop + ha, sh)
199            extended_index.append(slice(start, stop))
200            crop_len = stop - start
201            left_halo = idx_start - start
202            right_halo = stop - idx_stop
203            local_crop.append(slice(left_halo, crop_len - right_halo))
204        return tuple(extended_index), tuple(local_crop)
205
206    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
207        """Return the transformed data at ``roi``, reading through the halo."""
208        index, local_crop = self._extend_halo(roi)
209        if self._with_channels:
210            index = (slice(None),) + index
211            local_crop = (slice(None),) + local_crop
212        out = self._transformation(self._source[index])
213        return out[local_crop]
214
215    def _params(self) -> dict:
216        """Return the keyword arguments needed to reconstruct this wrapper."""
217        params = super()._params()
218        params["halo"] = self._halo
219        return params

222@register_wrapper
223class TransformationSource(WrapperSource):
224    """Apply a coordinate-aware transformation to the wrapped source on read.
225
226    The transformation may depend on both the data values and the read coordinates, so its
227    signature is ``transformation(block, roi)`` where ``roi`` is the tuple of slices being read.
228
229    Args:
230        source: The wrapped source-like object.
231        transformation: The transformation to apply, called as ``transformation(block, roi)``.
232        dtype: The output dtype. Defaults to the wrapped source's dtype.
233    """
234
235    def __init__(self, source: SourceLike, transformation: Callable, *,
236                 dtype: Optional[np.dtype] = None) -> None:
237        super().__init__(source)
238        if not callable(transformation):
239            raise ValueError("Expect the transformation to be callable.")
240        self._transformation = transformation
241        self._dtype = None if dtype is None else np.dtype(dtype)
242
243    @property
244    def dtype(self) -> np.dtype:
245        return self._dtype if self._dtype is not None else self._source.dtype
246
247    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
248        """Return the transformed data at ``roi`` (a full tuple of concrete slices)."""
249        return self._transformation(self._source[roi], roi)
250
251    def _params(self) -> dict:
252        """Return the keyword arguments needed to reconstruct this wrapper."""
253        return {"transformation": self._transformation, "dtype": self._dtype}

256@register_wrapper
257class MultiTransformationSource(WrapperSource):
258    """Apply a transformation jointly to blocks read from several equally-shaped sources.
259
260    The transformation depends only on the data values, so its signature is
261    ``transformation(*blocks)`` (or ``transformation(blocks)`` when ``apply_to_list`` is set). All
262    wrapped sources must have the same shape.
263
264    Args:
265        transformation: The transformation to apply to the per-source blocks.
266        sources: The wrapped source-like objects (at least one).
267        apply_to_list: Whether the blocks are passed as a single list argument instead of being
268            splatted as individual arguments.
269        dtype: The output dtype. Defaults to the first source's dtype.
270    """
271
272    def __init__(self, transformation: Callable, *sources: SourceLike,
273                 apply_to_list: bool = False, dtype: Optional[np.dtype] = None) -> None:
274        if not callable(transformation):
275            raise ValueError("Expect the transformation to be callable.")
276        if len(sources) == 0:
277            raise ValueError("Expect at least one source.")
278        self._sources: List[Source] = [as_source(s) for s in sources]
279        if any(s.shape != self._sources[0].shape for s in self._sources[1:]):
280            raise ValueError("All sources must have the same shape.")
281        # The first source backs the inherited metadata (shape / chunks / shards).
282        super().__init__(self._sources[0])
283        self._transformation = transformation
284        self._apply_to_list = bool(apply_to_list)
285        self._dtype = None if dtype is None else np.dtype(dtype)
286
287    @property
288    def _wrapped_sources(self) -> Tuple[Source, ...]:
289        return tuple(self._sources)
290
291    @property
292    def dtype(self) -> np.dtype:
293        return self._dtype if self._dtype is not None else self._sources[0].dtype
294
295    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
296        """Return the joint transformation of all sources at ``roi``."""
297        inputs = [s[roi] for s in self._sources]
298        return self._transformation(inputs) if self._apply_to_list else self._transformation(*inputs)
299
300    def _params(self) -> dict:
301        """Return the keyword arguments needed to reconstruct this wrapper."""
302        return {"transformation": self._transformation, "apply_to_list": self._apply_to_list,
303                "dtype": self._dtype}
304
305    @classmethod
306    def _from_wrapped(cls, sources: Sequence[Source], params: dict) -> "MultiTransformationSource":
307        """Rebuild from all wrapped sources (transformation comes first, positionally)."""
308        params = dict(params)
309        transformation = params.pop("transformation")
310        return cls(transformation, *sources, **params)

13@register_wrapper
14class ThresholdSource(SimpleTransformationSource):
15    """Threshold the wrapped source on read.
16
17    ``operator(source[roi], threshold)`` is returned as a boolean array (``operator`` defaults to
18    ``numpy.greater``, i.e. ``source[roi] > threshold``).
19
20    Args:
21        source: The wrapped source-like object.
22        threshold: The threshold value.
23        operator: The comparison operator applied as ``operator(block, threshold)``.
24    """
25
26    def __init__(self, source: SourceLike, threshold: float, operator: callable = np.greater) -> None:
27        self._threshold = float(threshold)
28        self._operator = operator
29        super().__init__(source, lambda block: operator(block, self._threshold), dtype=np.dtype(bool))
30
31    @property
32    def threshold(self) -> float:
33        """The threshold value."""
34        return self._threshold
35
36    def _params(self) -> dict:
37        """Return the keyword arguments needed to reconstruct this wrapper."""
38        return {"threshold": self._threshold, "operator": self._operator}

41@register_wrapper
42class NormalizeSource(SimpleTransformationSource):
43    """Normalize the wrapped source to ``[0, 1]`` on read.
44
45    Each read block is independently normalized by its own min and max. Note that this means a
46    block-wise read is *not* equivalent to a single whole-array read (the normalization is
47    block-local), matching elf's ``NormalizeWrapper``.
48
49    Args:
50        source: The wrapped source-like object.
51        dtype: The output (floating point) dtype.
52        with_channels: Whether the wrapped source has a leading channel axis.
53    """
54
55    eps = 1.0e-6
56
57    def __init__(self, source: SourceLike, dtype: str = "float32", with_channels: bool = False) -> None:
58        self._norm_dtype = np.dtype(dtype)
59        super().__init__(source, self._normalize, dtype=self._norm_dtype, with_channels=with_channels)
60
61    def _normalize(self, block: np.ndarray) -> np.ndarray:
62        """Normalize a block to ``[0, 1]`` using its own min and max."""
63        block = block.astype(self._norm_dtype)
64        block -= block.min()
65        block /= (block.max() + self.eps)
66        return block
67
68    def _params(self) -> dict:
69        """Return the keyword arguments needed to reconstruct this wrapper."""
70        return {"dtype": str(self._norm_dtype), "with_channels": self._with_channels}

108@register_wrapper
109class RoiSource(WrapperSource):
110    """Restrict the wrapped source to a region of interest.
111
112    This is a read-and-write view: reads and writes are offset into the parent source, so writing
113    through a :class:`RoiSource` updates the parent (when the parent is writable). For block-wise
114    *distributed* output, the ROI offset must be chunk-aligned in the parent, otherwise concurrent
115    writes to a shared parent chunk can corrupt it (the runner aligns blocks to this wrapper's
116    shape, not the parent's). For an unaligned or read-only ROI, use it purely as an input view.
117
118    Args:
119        source: The wrapped source-like object.
120        roi: The region of interest as a tuple of ints / slices over the source. Integer entries
121            select a single index along that axis; missing trailing axes default to the full extent.
122        squeeze: Whether to drop axes selected by an integer entry from the wrapper's shape and
123            output (and require matching input on write). Defaults to False.
124    """
125
126    def __init__(self, source: SourceLike, roi: Sequence[Union[int, slice]], squeeze: bool = False) -> None:
127        super().__init__(source)
128        self._orig_roi = tuple(roi)
129        self._squeeze = bool(squeeze)
130        self._roi, roi_squeeze = _normalize_construction_roi(self._orig_roi, self._source.shape)
131        # Axes introduced as singletons by an integer roi entry; only dropped when squeeze is set.
132        self._squeeze_axes = roi_squeeze if self._squeeze else ()
133        self._kept_axes = tuple(ax for ax in range(len(self._roi)) if ax not in self._squeeze_axes)
134
135    @property
136    def shape(self) -> Tuple[int, ...]:
137        return tuple(self._roi[ax].stop - self._roi[ax].start for ax in self._kept_axes)
138
139    @property
140    def ndim(self) -> int:
141        return len(self._kept_axes)
142
143    @property
144    def chunks(self) -> Optional[Tuple[int, ...]]:
145        src_chunks = self._source.chunks
146        if src_chunks is None:
147            return None
148        return tuple(min(src_chunks[ax], self._roi[ax].stop - self._roi[ax].start) for ax in self._kept_axes)
149
150    @property
151    def shards(self) -> Optional[Tuple[int, ...]]:
152        # A sub-region does not in general share the parent's shard grid.
153        return None
154
155    @property
156    def writable(self) -> bool:
157        """A roi view is writable when its parent is."""
158        return self._source.writable
159
160    def _map_roi_to_source(self, roi: Tuple[slice, ...]) -> Tuple[slice, ...]:
161        """Map a roi over the kept (reduced) axes into a full index of the parent source."""
162        full_index = list(self._roi)  # Default to the construction roi (singleton on squeezed axes).
163        for sl, ax in zip(roi, self._kept_axes):
164            offset = self._roi[ax].start
165            start = (int(sl.start) if sl.start is not None else 0) + offset
166            stop = (int(sl.stop) if sl.stop is not None else (self._roi[ax].stop - offset)) + offset
167            full_index[ax] = slice(start, stop)
168        return tuple(full_index)
169
170    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
171        """Return the parent data for the (reduced) region ``roi``."""
172        out = self._source[self._map_roi_to_source(roi)]
173        if self._squeeze_axes:
174            out = np.squeeze(out, axis=self._squeeze_axes)
175        return out
176
177    def _setitem(self, roi: Tuple[slice, ...], value: np.ndarray) -> None:
178        """Write ``value`` into the corresponding region of the parent source."""
179        value = np.asarray(value)
180        if self._squeeze_axes:
181            value = np.expand_dims(value, axis=self._squeeze_axes)
182        self._source[self._map_roi_to_source(roi)] = value
183
184    def _params(self) -> dict:
185        """Return the keyword arguments needed to reconstruct this wrapper."""
186        return {"roi": self._orig_roi, "squeeze": self._squeeze}

189@register_wrapper
190class PadSource(WrapperSource):
191    """Right-pad the wrapped source on read.
192
193    The wrapper's shape is the source shape grown by ``pad_width`` per axis. Reads that extend past
194    the source are filled by :func:`numpy.pad`. Only right-padding is supported (the source occupies
195    the lower corner of the padded space). This is a read-only view.
196
197    Args:
198        source: The wrapped source-like object.
199        pad_width: The number of elements to append along each axis.
200        mode: The padding mode passed to :func:`numpy.pad`.
201    """
202
203    def __init__(self, source: SourceLike, pad_width: Sequence[int], mode: str = "constant") -> None:
204        super().__init__(source)
205        if len(pad_width) != self._source.ndim:
206            raise ValueError(f"Expect pad_width of length {self._source.ndim}, got {len(pad_width)}.")
207        self._pad_width = tuple(int(p) for p in pad_width)
208        self._src_shape = self._source.shape
209        self._mode = mode
210
211    @property
212    def shape(self) -> Tuple[int, ...]:
213        return tuple(sh + pw for sh, pw in zip(self._src_shape, self._pad_width))
214
215    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
216        """Return the (possibly right-padded) data at ``roi``."""
217        local_pad, local_index = [], []
218        for sl, sh, psh in zip(roi, self._src_shape, self.shape):
219            start = int(sl.start) if sl.start is not None else 0
220            stop = int(sl.stop) if sl.stop is not None else psh
221            overhang_start = max(0, start - sh)
222            overhang_stop = max(0, stop - sh)
223            if overhang_start > 0:
224                raise NotImplementedError("PadSource only supports right-padding.")
225            elif overhang_stop > 0:
226                local_pad.append(overhang_stop)
227                local_index.append(slice(start, sh))
228            else:
229                local_pad.append(0)
230                local_index.append(slice(start, stop))
231
232        out = self._source[tuple(local_index)]
233        if any(lpad > 0 for lpad in local_pad):
234            pad_width = tuple((0, lpad) for lpad in local_pad)
235            out = np.pad(out, pad_width, mode=self._mode)
236        return out
237
238    def _params(self) -> dict:
239        """Return the keyword arguments needed to reconstruct this wrapper."""
240        return {"pad_width": self._pad_width, "mode": self._mode}

243@register_wrapper
244class ExpandDimsSource(WrapperSource):
245    """Insert one or more singleton axes into the wrapped source on read.
246
247    The inverse of a squeezing :class:`RoiSource`: the wrapper presents the wrapped source with
248    extra length-1 axes inserted at ``axis``, following :func:`numpy.expand_dims` semantics (``axis``
249    refers to positions in the *expanded* result). Use this to promote e.g. a 2d image to a 3d
250    volume ``(1, y, x)`` without materializing a copy. This is a read-only view.
251
252    Args:
253        source: The wrapped source-like object.
254        axis: Position(s) in the expanded result at which to insert a new singleton axis. May be a
255            single int or a sequence of ints; negative values count from the end.
256    """
257
258    def __init__(self, source: SourceLike, axis: Union[int, Sequence[int]] = 0) -> None:
259        super().__init__(source)
260        self._axis = axis
261        axes = (axis,) if isinstance(axis, (int, np.integer)) else tuple(int(a) for a in axis)
262        out_ndim = self._source.ndim + len(axes)
263        normalized = []
264        for a in axes:
265            a = int(a)
266            if a < 0:
267                a += out_ndim
268            if not 0 <= a < out_ndim:
269                raise ValueError(f"axis {axis} is out of bounds for expanded ndim {out_ndim}.")
270            normalized.append(a)
271        if len(set(normalized)) != len(normalized):
272            raise ValueError(f"repeated axis in {axis}.")
273        self._new_axes = tuple(sorted(normalized))
274        # Axes of the expanded shape that map back to the wrapped source (in order).
275        self._kept_axes = tuple(ax for ax in range(out_ndim) if ax not in self._new_axes)
276
277    @property
278    def ndim(self) -> int:
279        return self._source.ndim + len(self._new_axes)
280
281    @property
282    def shape(self) -> Tuple[int, ...]:
283        out = [1] * self.ndim
284        for ax, sh in zip(self._kept_axes, self._source.shape):
285            out[ax] = sh
286        return tuple(out)
287
288    @property
289    def chunks(self) -> Optional[Tuple[int, ...]]:
290        src_chunks = self._source.chunks
291        if src_chunks is None:
292            return None
293        out = [1] * self.ndim
294        for ax, c in zip(self._kept_axes, src_chunks):
295            out[ax] = c
296        return tuple(out)
297
298    @property
299    def shards(self) -> Optional[Tuple[int, ...]]:
300        # Inserting singleton axes does not in general preserve the parent's shard grid.
301        return None
302
303    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
304        """Return the wrapped data at ``roi`` with the singleton axes inserted.
305
306        ``roi`` is a full tuple of slices over the expanded shape (the inserted axes are
307        ``slice(0, 1)``); read the wrapped source over the kept axes, then re-insert the axes.
308        """
309        src_index = tuple(roi[ax] for ax in self._kept_axes)
310        return np.expand_dims(self._source[src_index], self._new_axes)
311
312    def _params(self) -> dict:
313        """Return the keyword arguments needed to reconstruct this wrapper."""
314        return {"axis": self._axis}

 24@register_wrapper
 25class AffineSource(WrapperSource):
 26    """Apply an affine transformation to the wrapped source on read.
 27
 28    The transformation is given either by ``affine_matrix`` or by individual parameters for
 29    ``scale`` / ``rotation`` / ``shear`` / ``translation`` (exactly one of the two ways).
 30
 31    Args:
 32        source: The wrapped source-like object (2D or 3D).
 33        shape: The output shape. Defaults to the wrapped source's shape.
 34        affine_matrix: The matrix defining the affine transformation, mapping output to input
 35            coordinates, of shape ``(ndim + 1, ndim + 1)``.
 36        scale: The scale factors (used when ``affine_matrix`` is not given).
 37        rotation: The rotation angles in degrees (used when ``affine_matrix`` is not given).
 38        shear: The shear angles in degrees (not implemented correctly yet; leave unset).
 39        translation: The translation vector (used when ``affine_matrix`` is not given).
 40        order: The interpolation order, supports orders 0 to 5. Use ``0`` (nearest) for label data.
 41        fill_value: The value used for output coordinates that map outside the input.
 42        anti_aliasing: Whether to Gaussian pre-smooth the input before sampling to avoid aliasing
 43            when downsampling. Recommended for intensity data; leave ``False`` for labels.
 44    """
 45
 46    def __init__(
 47        self,
 48        source: SourceLike,
 49        shape: Optional[Tuple[int, ...]] = None,
 50        *,
 51        affine_matrix: Optional[np.ndarray] = None,
 52        scale: Optional[List[float]] = None,
 53        rotation: Optional[List[float]] = None,
 54        shear: Optional[List[float]] = None,
 55        translation: Optional[List[float]] = None,
 56        order: int = 0,
 57        fill_value: Number = 0,
 58        anti_aliasing: bool = False,
 59    ) -> None:
 60        super().__init__(source)
 61        ndim = self._source.ndim
 62        if ndim not in (2, 3):
 63            raise ValueError(f"AffineSource supports 2d or 3d data, got {ndim}d.")
 64        if not 0 <= int(order) <= 5:
 65            raise ValueError(f"order must be in [0, 5], got {order}.")
 66
 67        have_matrix = affine_matrix is not None
 68        have_parameter = any(p is not None for p in (scale, rotation, shear, translation))
 69        if have_matrix == have_parameter:
 70            raise ValueError("Pass exactly one of affine_matrix or the scale/rotation/shear/translation parameters.")
 71
 72        if have_matrix:
 73            matrix = np.asarray(affine_matrix, dtype="float64")
 74        else:
 75            matrix = compute_affine_matrix(scale, rotation, shear, translation)
 76        if matrix.shape != (ndim + 1, ndim + 1):
 77            raise ValueError(f"Invalid affine matrix shape {matrix.shape}, expected {(ndim + 1, ndim + 1)}.")
 78
 79        self._matrix = matrix
 80        self._shape = self._source.shape if shape is None else tuple(int(s) for s in shape)
 81        self._order = int(order)
 82        self._fill_value = fill_value
 83        self._anti_aliasing = bool(anti_aliasing)
 84
 85    @property
 86    def shape(self) -> Tuple[int, ...]:
 87        """The output (transformed) shape."""
 88        return self._shape
 89
 90    @property
 91    def matrix(self) -> np.ndarray:
 92        """The affine matrix mapping output to input coordinates."""
 93        return self._matrix
 94
 95    @property
 96    def shards(self) -> Optional[Tuple[int, ...]]:
 97        # An affine of the input does not map onto the input's shard grid.
 98        return None
 99
100    def _getitem(self, roi: Tuple[slice, ...]) -> np.ndarray:
101        """Return the affine-transformed data for the output region ``roi``."""
102        sigma = None
103        if self._anti_aliasing:
104            aa = np.asarray(
105                bic.transformation.compute_anti_aliasing_sigma(self._matrix, self._source.ndim),
106                dtype="float64",
107            )
108            if np.any(aa > 0):
109                sigma = aa
110        return transform_subvolume_affine(
111            self._source, self._matrix, roi, order=self._order,
112            fill_value=self._fill_value, sigma=sigma,
113        )
114
115    def _params(self) -> dict:
116        """Return the keyword arguments needed to reconstruct this wrapper."""
117        return {
118            "shape": self._shape,
119            "affine_matrix": self._matrix.tolist(),
120            "order": self._order,
121            "fill_value": self._fill_value,
122            "anti_aliasing": self._anti_aliasing,
123        }