Images API reference¶

def has_axis(self, axis: str) -> bool:
    """Return True if the image has the given axis."""
    return self.axes_handler.has_axis(axis)

def set_space_unit(self, unit: SpaceUnits = DefaultSpaceUnit) -> None:
    """Set the unit of the spatial axes; the time unit is untouched.

    Args:
        unit: The space unit to set.
    """
    meta = self._meta_handler.get_meta()
    meta = meta.to_units(space_unit=unit)
    self._meta_handler.update_meta(meta)

def set_time_unit(self, unit: TimeUnits = DefaultTimeUnit) -> None:
    """Set the unit of the time axis; the space unit is untouched.

    Args:
        unit: The time unit to set.
    """
    meta = self._meta_handler.get_meta()
    meta = meta.to_units(time_unit=unit)
    self._meta_handler.update_meta(meta)

@deprecated(replacement="set_space_unit() / set_time_unit()")
def set_axes_units(
    self,
    space_unit: SpaceUnits = DefaultSpaceUnit,
    time_unit: TimeUnits = DefaultTimeUnit,
) -> None:
    """Set BOTH the space and the time units of the image axes.

    Note that both units are set on every call: an *omitted* parameter is
    set to its default, not left unchanged. To change one unit without
    touching the other, use `set_space_unit` / `set_time_unit` — which is
    also why this method is deprecated in their favour.

    Args:
        space_unit: The space unit of the image.
        time_unit: The time unit of the image.
    """
    meta = self._meta_handler.get_meta()
    meta = meta.to_units(space_unit=space_unit, time_unit=time_unit)
    self._meta_handler.update_meta(meta)

def set_axes_names(self, axes_names: Sequence[str]) -> None:
    """Set the axes names of the label.

    Args:
        axes_names (Sequence[str]): The axes names to set.
    """
    meta = self._meta_handler.get_meta()
    meta = meta.rename_axes(axes_names=axes_names)
    self._meta_handler._axes_setup = meta.axes_handler.axes_setup
    self._meta_handler.update_meta(meta)

def set_name(
    self,
    name: str,
) -> None:
    """Set the name of the image in the metadata.

    This does not change the group name or any paths.

    Args:
        name (str): The name of the image.
    """
    meta = self._meta_handler.get_meta()
    meta = meta.rename_image(name=name)
    self._meta_handler.update_meta(meta)

@contextmanager
def track_writes(self) -> Generator[list[tuple[RegionType, ...]]]:
    """Record the region of every `set_*` write made through this handle.

    Only the `set_*` methods are recorded — `set_array`, `set_roi`, and
    their masked variants. Yields a list that accumulates one on-disk
    index tuple per completed write, ready to hand to
    `consolidate(regions=...)` so the pyramid rebuild covers exactly what
    was written:

    ```python
    with image.track_writes() as regions:
        image.set_roi(roi, patch)
        image.set_array(other, y=slice(0, 64))
    image.consolidate(regions=regions)
    ```

    Tracking is scoped to the context and to *this* handle in *this*
    process: writes through setter pipes built directly on the zarr
    array, through another handle to the same image, or in worker
    processes (`ProcessMapper`) are not seen — iterators track their own
    writes through their ROI list instead. Threads sharing this handle
    all record, and a write that raises records nothing. Nested contexts
    each keep their own list.
    """
    regions: list[tuple[RegionType, ...]] = []
    with self._tracker_lock:
        self._write_trackers.append(regions)
    try:
        yield regions
    finally:
        with self._tracker_lock:
            self._write_trackers.remove(regions)

def roi(self, name: str | None = "image") -> Roi:
    """Return the ROI covering the entire image."""
    slices = {}
    for ax_name in ["t", "z", "y", "x"]:
        axis_size = self.dimensions.get(ax_name, default=None)
        if axis_size is None:
            continue
        slices[ax_name] = slice(0, axis_size)
    roi_px = Roi.from_values(name=name, slices=slices, space="pixel")
    return roi_px.to_world(pixel_size=self.pixel_size)

def build_image_roi_table(self, name: str | None = "image") -> RoiTable:
    """Build the ROI table containing the ROI covering the entire image."""
    return RoiTable(rois=[self.roi(name=name)])

def require_dimensions_match(
    self,
    other: "AbstractImage",
    allow_singleton: bool = False,
) -> None:
    """Assert that two images have matching spatial dimensions.

    Args:
        other: The other image to compare to.
        allow_singleton: If True, allow singleton dimensions to be
            compatible with non-singleton dimensions.

    Raises:
        NgioValueError: If the images do not have compatible dimensions.
    """
    self.dimensions.require_dimensions_match(
        other.dimensions, allow_singleton=allow_singleton
    )

def check_if_dimensions_match(
    self,
    other: "AbstractImage",
    allow_singleton: bool = False,
) -> bool:
    """Check if two images have matching spatial dimensions.

    Args:
        other: The other image to compare to.
        allow_singleton: If True, allow singleton dimensions to be
            compatible with non-singleton dimensions.

    Returns:
        bool: True if the images have matching dimensions, False otherwise.
    """
    return self.dimensions.check_if_dimensions_match(
        other.dimensions, allow_singleton=allow_singleton
    )

def require_axes_match(
    self,
    other: "AbstractImage",
) -> None:
    """Assert that two images have compatible axes.

    Args:
        other: The other image to compare to.

    Raises:
        NgioValueError: If the images do not have compatible axes.
    """
    self.dimensions.require_axes_match(other.dimensions)

def check_if_axes_match(
    self,
    other: "AbstractImage",
) -> bool:
    """Check if two images have compatible axes.

    Args:
        other: The other image to compare to.

    Returns:
        bool: True if the images have compatible axes, False otherwise.

    """
    return self.dimensions.check_if_axes_match(other.dimensions)

def require_rescalable(
    self,
    other: "AbstractImage",
) -> None:
    """Assert that two images can be rescaled to each other.

    For this to be true, the images must have the same axes, and
    the pixel sizes must be compatible (i.e. one can be scaled to the other).

    Args:
        other: The other image to compare to.

    Raises:
        NgioValueError: If the images cannot be scaled to each other.
    """
    self.dimensions.require_rescalable(other.dimensions)

def check_if_rescalable(
    self,
    other: "AbstractImage",
) -> bool:
    """Check if two images can be rescaled to each other.

    For this to be true, the images must have the same axes, and
    the pixel sizes must be compatible (i.e. one can be scaled to the other).

    Args:
        other: The other image to compare to.

    Returns:
        bool: True if the images can be rescaled to each other, False otherwise.
    """
    return self.dimensions.check_if_rescalable(other.dimensions)

def get_channel_idx(
    self, channel_label: str | None = None, wavelength_id: str | None = None
) -> int:
    """Get the index of a channel by its label or wavelength ID."""
    return self.channels_meta.get_channel_idx(
        channel_label=channel_label, wavelength_id=wavelength_id
    )

def resolve_channel_selection(
    self, channel_selection: ChannelSlicingInputType = None
) -> dict[str, int | list[int]]:
    """Resolve a channel selection against this image's channel metadata.

    Accepts everything the `get_*` methods accept as `channel_selection` —
    an index, a channel label, a `ChannelSelectionModel`, or a sequence of
    those — and returns the slicing entry it resolves to (`{"c": index}`
    or `{"c": [indices]}`; `{}` for `None`), ready to use as slicing
    kwargs. Resolution touches only metadata, so this is also the way to
    validate a selection before loading any data.

    Raises:
        NgioValueError: If a referenced channel does not exist.
    """
    return _parse_channel_selection(self, channel_selection)

def get_as_numpy(
    self,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    **slicing_kwargs: slice | int | Sequence[int] | None,
) -> np.ndarray:
    """Get the image as a numpy array.

    Args:
        channel_selection: Select a specific channel by label.
            If None, all channels are returned.
            Alternatively, you can slice arbitrary channels
            using the slicing kwargs (c=[0, 2]).
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        **slicing_kwargs: The slices to get the array.

    Returns:
        The array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_as_numpy(
        axes_order=axes_order, transforms=transforms, **_slicing_kwargs
    )

def get_roi_as_numpy(
    self,
    roi: Roi,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    **slicing_kwargs: SlicingInputType,
) -> np.ndarray:
    """Get the image as a numpy array for a region of interest.

    Args:
        roi: The region of interest to get the array.
        channel_selection: Select a what subset of channels to return.
            If None, all channels are returned.
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        **slicing_kwargs: Per-axis selections in absolute
            coordinates; an explicit selection on an axis the `roi`
            already pins replaces the roi-derived one (and drops the
            pipe's `roi`).

    Returns:
        The array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_roi_as_numpy(
        roi=roi, axes_order=axes_order, transforms=transforms, **_slicing_kwargs
    )

def get_as_dask(
    self,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    **slicing_kwargs: SlicingInputType,
) -> da.Array:
    """Get the image as a dask array.

    Args:
        channel_selection: Select a what subset of channels to return.
            If None, all channels are returned.
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        **slicing_kwargs: The slices to get the array.

    Returns:
        The dask array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_as_dask(
        axes_order=axes_order, transforms=transforms, **_slicing_kwargs
    )

def get_roi_as_dask(
    self,
    roi: Roi,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    **slicing_kwargs: SlicingInputType,
) -> da.Array:
    """Get the image as a dask array for a region of interest.

    Args:
        roi: The region of interest to get the array.
        channel_selection: Select a what subset of channels to return.
            If None, all channels are returned.
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        **slicing_kwargs: Per-axis selections in absolute
            coordinates; an explicit selection on an axis the `roi`
            already pins replaces the roi-derived one (and drops the
            pipe's `roi`).

    Returns:
        The dask array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_roi_as_dask(
        roi=roi, axes_order=axes_order, transforms=transforms, **_slicing_kwargs
    )

def get_array(
    self,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    mode: Literal["numpy", "dask"] = "numpy",
    **slicing_kwargs: SlicingInputType,
) -> np.ndarray | da.Array:
    """Get the image as a zarr array.

    Args:
        channel_selection: Select a what subset of channels to return.
            If None, all channels are returned.
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        mode: The object type to return.
            Can be "dask", "numpy".
        **slicing_kwargs: The slices to get the array.

    Returns:
        The zarr array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_array(
        axes_order=axes_order, mode=mode, transforms=transforms, **_slicing_kwargs
    )

def get_roi(
    self,
    roi: Roi,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    mode: Literal["numpy", "dask"] = "numpy",
    **slicing_kwargs: SlicingInputType,
) -> np.ndarray | da.Array:
    """Get the image as a zarr array for a region of interest.

    Args:
        roi: The region of interest to get the array.
        channel_selection: Select a what subset of channels to return.
            If None, all channels are returned.
        axes_order: The order of the axes to return the array.
        transforms: The transforms to apply to the array.
        mode: The object type to return.
            Can be "dask", "numpy".
        **slicing_kwargs: Per-axis selections in absolute
            coordinates; an explicit selection on an axis the `roi`
            already pins replaces the roi-derived one (and drops the
            pipe's `roi`).

    Returns:
        The zarr array of the region of interest.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    return self._get_roi(
        roi=roi,
        axes_order=axes_order,
        mode=mode,
        transforms=transforms,
        **_slicing_kwargs,
    )

def set_array(
    self,
    patch: np.ndarray | da.Array,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    merge: MergeInput | None = None,
    **slicing_kwargs: SlicingInputType,
) -> None:
    """Set the image array.

    Dask patches are serial-only: concurrent dask writes from several
    threads can silently lose updates (numpy patches are unaffected).

    Args:
        patch: The array to set.
        channel_selection: Select a what subset of channels to return.
            If None, all channels are set.
        axes_order: The order of the axes to set the array.
        transforms: The transforms to apply to the array.
        merge: How to combine the patch with what is already there —
            a rule name, a callable, or a policy. `None` overwrites.
        **slicing_kwargs: The slices to set the array.
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    self._set_array(
        patch=patch,
        axes_order=axes_order,
        transforms=transforms,
        merge=merge,
        **_slicing_kwargs,
    )

def set_roi(
    self,
    roi: Roi,
    patch: np.ndarray | da.Array,
    channel_selection: ChannelSlicingInputType = None,
    axes_order: Sequence[str] | None = None,
    transforms: Sequence[TransformProtocol] | None = None,
    merge: MergeInput | None = None,
    **slicing_kwargs: SlicingInputType,
) -> None:
    """Set the image array for a region of interest.

    Dask patches are serial-only: concurrent dask writes from several
    threads can silently lose updates (numpy patches are unaffected).

    Args:
        roi: The region of interest to set the array.
        patch: The array to set.
        channel_selection: Select a what subset of channels to return.
        axes_order: The order of the axes to set the array.
        transforms: The transforms to apply to the array.
        merge: How to combine the patch with what is already there —
            a rule name, a callable, or a policy. `None` overwrites.
        **slicing_kwargs: Per-axis selections in absolute
            coordinates; an explicit selection on an axis the `roi`
            already pins replaces the roi-derived one (and drops the
            pipe's `roi`).
    """
    _slicing_kwargs = add_channel_selection_to_slicing_dict(
        image=self, channel_selection=channel_selection, slicing_dict=slicing_kwargs
    )
    self._set_roi(
        roi=roi,
        patch=patch,
        axes_order=axes_order,
        transforms=transforms,
        merge=merge,
        **_slicing_kwargs,
    )

def consolidate(
    self,
    order: InterpolationOrder = "linear",
    mode: ConsolidationMode | None = None,
    regions: ConsolidationRegions | None = None,
) -> None:
    """Consolidate the image on disk.

    Args:
        order: The interpolation order.
        mode: How to build each level, see `ConsolidationMode`.
        regions: Where this level changed — typically the `Roi`s that were
            written (`set_roi`'s own argument fits directly); raw on-disk
            index tuples, as a setter pipe's
            `slicing_ops.normalized_slicing_tuple` produces, also work.
            Only the pyramid regions derived from them are rebuilt,
            identically to a full rebuild; outside the exact envelope
            (integral downsamples, `order` not `"cubic"`, coverage below
            `ConsolidationConfig.partial_max_coverage`) the whole pyramid
            is rebuilt instead, silently. `None` rebuilds everything.
    """
    self._consolidate(order=order, mode=mode, regions=regions)