Skip to content

zarrnii.plugins

Plugin API for segmentation and scaled-processing workflows, including hook markers, hook specifications, plugin-manager helpers, and bundled plugins.

Package exports

ZarrNii plugins package.

This package provides extensible plugin architectures for various image processing tasks such as segmentation, filtering, and analysis using the pluggy framework.

Plugin authors should depend on zarrnii and import hookimpl from :mod:zarrnii.plugins.

Classes

zarrnii.plugins.GaussianBiasFieldCorrection(sigma=5.0, mode='reflect')

Bias field correction plugin using multi-resolution processing.

This plugin estimates a smooth bias field at low resolution using simple smoothing (as a placeholder for more sophisticated methods like N4) and applies the correction to full resolution data by division.

Parameters:

  • sigma (float, default: 5.0 ) –

    Standard deviation for Gaussian smoothing (default: 5.0)

  • mode (str, default: 'reflect' ) –

    Boundary condition for smoothing (default: 'reflect')

Initialize bias field correction plugin.

Parameters:

  • sigma (float, default: 5.0 ) –

    Standard deviation for Gaussian smoothing

  • mode (str, default: 'reflect' ) –

    Boundary condition for smoothing

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
29
30
31
32
33
34
35
36
37
38
def __init__(self, sigma: float = 5.0, mode: str = "reflect"):
    """
    Initialize bias field correction plugin.

    Args:
        sigma: Standard deviation for Gaussian smoothing
        mode: Boundary condition for smoothing
    """
    self.sigma = sigma
    self.mode = mode

Functions

zarrnii.plugins.GaussianBiasFieldCorrection.lowres_func(lowres_array)

Estimate bias field from low-resolution data.

This is a simplified bias field estimation using Gaussian smoothing. In practice, this could be replaced with more sophisticated methods like N4 bias field correction.

Parameters:

  • lowres_array (ndarray) –

    Downsampled input image

Returns:

  • ndarray –

    Estimated bias field at low resolution

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
 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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Estimate bias field from low-resolution data.

    This is a simplified bias field estimation using Gaussian smoothing.
    In practice, this could be replaced with more sophisticated methods
    like N4 bias field correction.

    Args:
        lowres_array: Downsampled input image

    Returns:
        Estimated bias field at low resolution
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    # Handle different array dimensions
    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        leading_shape = work_array.shape[:-3]
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Ensure we're working with float data
    if work_array.dtype.kind in ["i", "u"]:  # integer types
        work_array = work_array.astype(np.float32)

    # Simple bias field estimation: smooth the image to get low-frequency components
    if work_array.ndim == 2:
        smoothed = ndimage.gaussian_filter(
            work_array, sigma=self.sigma, mode=self.mode
        )
    else:
        # Apply smoothing to each volume if we have multiple
        if work_array.ndim == 4:  # batched 3D volumes
            smoothed = np.zeros_like(work_array)
            for i in range(work_array.shape[0]):
                smoothed[i] = ndimage.gaussian_filter(
                    work_array[i], sigma=self.sigma, mode=self.mode
                )
        else:  # single 3D volume
            smoothed = ndimage.gaussian_filter(
                work_array, sigma=self.sigma, mode=self.mode
            )

    # Reshape back to original shape if needed
    if original_shape != smoothed.shape:
        smoothed = smoothed.reshape(original_shape)

    # Avoid division by zero by adding small epsilon
    smoothed = np.maximum(smoothed, np.finfo(smoothed.dtype).eps)

    return smoothed
zarrnii.plugins.GaussianBiasFieldCorrection.highres_func(fullres_array, upsampled_output)

Apply bias field correction to full-resolution data.

This function takes the upsampled bias field (same size as fullres_array) and applies it to the full-resolution data by division.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution array (dask or NumPy)

  • upsampled_output –

    Upsampled bias field (same shape as fullres_array; dask or NumPy)

Returns:

  • –

    Bias-corrected full-resolution array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply bias field correction to full-resolution data.

    This function takes the upsampled bias field (same size as fullres_array)
    and applies it to the full-resolution data by division.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution array (dask or NumPy)
        upsampled_output: Upsampled bias field (same shape as fullres_array;
            dask or NumPy)

    Returns:
        Bias-corrected full-resolution array (same type as inputs)
    """
    # Avoid division by zero by clamping the bias field to a small epsilon.
    # np.maximum works for both dask arrays (via __array_ufunc__) and NumPy.
    epsilon = np.finfo(np.float32).eps
    corrected_array = fullres_array / np.maximum(upsampled_output, epsilon)

    return corrected_array
zarrnii.plugins.GaussianBiasFieldCorrection.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
130
131
132
133
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "Gaussian Bias Field Correction"
zarrnii.plugins.GaussianBiasFieldCorrection.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
135
136
137
138
139
140
141
142
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution bias field correction. Estimates smooth bias field "
        "at low resolution using Gaussian smoothing and applies correction "
        "to full resolution data by division."
    )

zarrnii.plugins.N4BiasFieldCorrection(spline_param=[2, 2, 2], convergence={'iters': [50, 50, 50, 50], 'tol': 1e-07}, shrink_factor=1)

N4 bias field correction plugin using multi-resolution processing.

This plugin estimates a smooth bias field at low resolution using the N4 bias field correction algorithm from ANTsPy and applies the correction to full resolution data by division.

Parameters:

  • spline_param (tuple[int, int, int], default: [2, 2, 2] ) –

    Spacing between knots for spline fitting (default: 200)

  • convergence (Optional[Dict[str, Any]], default: {'iters': [50, 50, 50, 50], 'tol': 1e-07} ) –

    Convergence criteria [iters, tol] (default: [50, 0.001])

  • shrink_factor (int, default: 1 ) –

    Shrink factor for processing (default: 1)

Initialize N4 bias field correction plugin.

Parameters:

  • spline_param (tuple[int, int, int], default: [2, 2, 2] ) –

    Spacing between knots for spline fitting

  • convergence (Optional[Dict[str, Any]], default: {'iters': [50, 50, 50, 50], 'tol': 1e-07} ) –

    Convergence criteria dict with 'iters' (list), 'tol'

  • shrink_factor (int, default: 1 ) –

    Shrink factor for processing

Raises:

  • ImportError –

    If antspyx is not installed

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
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
def __init__(
    self,
    spline_param: tuple[int, int, int] = [2, 2, 2],
    convergence: Optional[Dict[str, Any]] = {
        "iters": [50, 50, 50, 50],
        "tol": 1e-07,
    },
    shrink_factor: int = 1,
):
    """
    Initialize N4 bias field correction plugin.

    Args:
        spline_param: Spacing between knots for spline fitting
        convergence: Convergence criteria dict with 'iters' (list), 'tol'
        shrink_factor: Shrink factor for processing

    Raises:
        ImportError: If antspyx is not installed
    """
    if not HAS_ANTSPYX:
        raise ImportError(
            "antspyx is required for N4BiasFieldCorrection. "
            "Install it with: pip install 'zarrnii[n4]' "
            "or pip install antspyx"
        )

    self.spline_param = spline_param
    self.convergence = convergence
    self.shrink_factor = shrink_factor

Functions

zarrnii.plugins.N4BiasFieldCorrection.lowres_func(lowres_array)

Estimate bias field from low-resolution data using N4 algorithm.

This function uses ANTsPy's N4 bias field correction to estimate the bias field at low resolution.

Parameters:

  • lowres_array (ndarray) –

    Downsampled input image

Returns:

  • ndarray –

    Estimated bias field at low resolution

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.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
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Estimate bias field from low-resolution data using N4 algorithm.

    This function uses ANTsPy's N4 bias field correction to estimate
    the bias field at low resolution.

    Args:
        lowres_array: Downsampled input image

    Returns:
        Estimated bias field at low resolution
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    # Handle different array dimensions
    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Ensure we're working with float data
    if work_array.dtype.kind in ["i", "u"]:  # integer types
        work_array = work_array.astype(np.float32)

    # Apply N4 bias field correction
    if work_array.ndim == 2:
        # For 2D data
        ants_img = ants.from_numpy(work_array)
        self.spline_param = [2, 2]
        # Use return_bias_field=True to get the bias field
        bias_result = ants.n4_bias_field_correction(
            ants_img,
            return_bias_field=True,
            spline_param=self.spline_param,
            convergence=self.convergence,
            shrink_factor=self.shrink_factor,
        )
        # Extract the bias field
        bias_field = bias_result.numpy()
    else:
        # Apply N4 to each volume if we have multiple or single 3D volume
        if work_array.ndim == 4:  # batched 3D volumes
            bias_field = np.zeros_like(work_array)
            for i in range(work_array.shape[0]):
                ants_img = ants.from_numpy(work_array[i])
                bias_result = ants.n4_bias_field_correction(
                    ants_img,
                    return_bias_field=True,
                    spline_param=self.spline_param,
                    convergence=self.convergence,
                    shrink_factor=self.shrink_factor,
                )
                bias_field[i] = bias_result.numpy()
        else:  # single 3D volume
            ants_img = ants.from_numpy(work_array)
            bias_result = ants.n4_bias_field_correction(
                ants_img,
                return_bias_field=True,
                spline_param=self.spline_param,
                convergence=self.convergence,
                shrink_factor=self.shrink_factor,
            )
            bias_field = bias_result.numpy()

    # Reshape back to original shape if needed
    if original_shape != bias_field.shape:
        bias_field = bias_field.reshape(original_shape)

    # Avoid division by zero by adding small epsilon
    bias_field = np.maximum(bias_field, np.finfo(bias_field.dtype).eps)

    return bias_field
zarrnii.plugins.N4BiasFieldCorrection.highres_func(fullres_array, upsampled_output)

Apply bias field correction to full-resolution data.

This function takes the upsampled bias field (same size as fullres_array) and applies it to the full-resolution data by division.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution array (dask or NumPy)

  • upsampled_output –

    Upsampled bias field (same shape as fullres; dask or NumPy)

Returns:

  • –

    Bias-corrected full-resolution array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply bias field correction to full-resolution data.

    This function takes the upsampled bias field (same size as
    fullres_array) and applies it to the full-resolution data by division.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution array (dask or NumPy)
        upsampled_output: Upsampled bias field (same shape as fullres;
            dask or NumPy)

    Returns:
        Bias-corrected full-resolution array (same type as inputs)
    """
    # Avoid division by zero by clamping the bias field to a small epsilon.
    # np.maximum works for both dask arrays (via __array_ufunc__) and NumPy.
    epsilon = np.finfo(np.float32).eps
    corrected_array = fullres_array / np.maximum(upsampled_output, epsilon)

    return corrected_array
zarrnii.plugins.N4BiasFieldCorrection.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
180
181
182
183
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "N4 Bias Field Correction"
zarrnii.plugins.N4BiasFieldCorrection.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
185
186
187
188
189
190
191
192
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution N4 bias field correction. Estimates smooth bias "
        "field at low resolution using ANTsPy N4 algorithm and applies "
        "correction to full resolution data by division."
    )

zarrnii.plugins.N4BiasFieldApply(log_space=False, log_offset=1e-06)

Bias field correction plugin for pre-computed N4 bias fields.

This plugin is a companion to the full :class:~zarrnii.plugins.N4BiasFieldCorrection plugin. Instead of running N4 itself it expects the user to supply the already-corrected downsampled image through the lowres_znimg parameter of :meth:~zarrnii.core.ZarrNii.apply_scaled_processing.

The lowres_func is a passthrough — it validates, float-casts, and zero-clamps the input, then optionally log-transforms it before upsampling. The highres_func divides the full-resolution image by the (optionally exponentiated) upsampled bias field to produce a corrected image.

Because no ANTsPy call is made this plugin has no antspyx dependency and works with the base zarrnii install.

Parameters:

  • log_space (bool, default: False ) –

    When True the bias field is log-transformed before upsampling and exponentiated before the correction is applied. This produces smoother interpolation because the bias field is approximately log-linear. Default: False.

  • log_offset (float, default: 1e-06 ) –

    Small positive constant added to the bias field before taking the logarithm to guard against log(0). Only used when log_space=True. Default: 1e-6.

Example (linear space, default)::

corrected = fullres.apply_scaled_processing(
    N4BiasFieldApply(),
    method="map_blocks",
    lowres_znimg=precomputed_bias_field_znimg,
)

Example (log space)::

corrected = fullres.apply_scaled_processing(
    N4BiasFieldApply(log_space=True),
    method="map_blocks",
    lowres_znimg=precomputed_bias_field_znimg,
)

Initialize the N4BiasFieldApply plugin.

Parameters:

  • log_space (bool, default: False ) –

    If True, log-transform the bias field before upsampling and exponentiate after upsampling before dividing.

  • log_offset (float, default: 1e-06 ) –

    Small positive constant added before log for numerical stability. Only used when log_space=True.

Raises:

  • ValueError –

    If log_offset is not strictly positive.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield_apply.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def __init__(self, log_space: bool = False, log_offset: float = 1e-6):
    """
    Initialize the N4BiasFieldApply plugin.

    Args:
        log_space: If ``True``, log-transform the bias field before
            upsampling and exponentiate after upsampling before dividing.
        log_offset: Small positive constant added before ``log`` for
            numerical stability.  Only used when ``log_space=True``.

    Raises:
        ValueError: If ``log_offset`` is not strictly positive.
    """
    if log_offset <= 0:
        raise ValueError(
            f"log_offset must be strictly positive, got {log_offset!r}"
        )
    self.log_space = log_space
    self.log_offset = log_offset

Functions

zarrnii.plugins.N4BiasFieldApply.lowres_func(lowres_array)

Return the pre-computed bias field, optionally log-transformed.

The downsampled input is expected to already contain the N4 bias field values (i.e. the output of a prior N4 run). When log_space=True the field is transformed as log(field + log_offset) before being returned so that the subsequent upsampling step interpolates in log space.

Parameters:

  • lowres_array (ndarray) –

    Pre-computed bias field at low resolution.

Returns:

  • ndarray –

    The bias field, cast to float32, zero-clamped, and optionally

  • ndarray –

    log-transformed.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield_apply.py 
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
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Return the pre-computed bias field, optionally log-transformed.

    The downsampled input is expected to already contain the N4 bias
    field values (i.e. the output of a prior N4 run).  When
    ``log_space=True`` the field is transformed as
    ``log(field + log_offset)`` before being returned so that the
    subsequent upsampling step interpolates in log space.

    Args:
        lowres_array: Pre-computed bias field at low resolution.

    Returns:
        The bias field, cast to float32, zero-clamped, and optionally
        log-transformed.
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")
    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Ensure float so that subsequent division in highres_func is safe.
    if lowres_array.dtype.kind in ("i", "u"):
        lowres_array = lowres_array.astype(np.float32)

    if self.log_space:
        # Clamp to zero first, then add log_offset to guarantee the
        # argument to log is strictly positive even if the bias field
        # contains negative values (e.g. from integer underflow).
        return np.log(np.maximum(lowres_array, 0.0) + self.log_offset)
    else:
        # Clamp to a small positive value to guard against division by zero
        # when the field is later upsampled and applied.
        return np.maximum(lowres_array, np.finfo(np.float32).eps)
zarrnii.plugins.N4BiasFieldApply.highres_func(fullres_array, upsampled_output, upsampled_lowres_mask=None)

Apply the upsampled bias field to full-resolution data.

Divides the full-resolution image by the (optionally exponentiated) upsampled bias field to produce a bias-corrected image.

When log_space=True the upsampled field is first exponentiated (exp(upsampled_log_bias)) before dividing, recovering the linear bias field that was log-transformed in :meth:lowres_func.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array (ndarray | Array) –

    Full-resolution array (dask or NumPy).

  • upsampled_output (ndarray | Array) –

    Upsampled bias field with the same shape as fullres_array (dask or NumPy). Contains log-transformed values when log_space=True.

  • upsampled_lowres_mask (ndarray | Array | None, default: None ) –

    Optional upsampled mask (0..1) used to constrain where correction is applied. When provided, correction is blended as corrected * mask + original * (1 - mask).

Returns:

  • ndarray | Array –

    Bias-corrected full-resolution array (same type as inputs).

Source code in zarrnii/plugins/scaled_processing/n4_biasfield_apply.py 
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
@hookimpl
def highres_func(
    self,
    fullres_array: np.ndarray | da.Array,
    upsampled_output: np.ndarray | da.Array,
    upsampled_lowres_mask: np.ndarray | da.Array | None = None,
) -> np.ndarray | da.Array:
    """
    Apply the upsampled bias field to full-resolution data.

    Divides the full-resolution image by the (optionally exponentiated)
    upsampled bias field to produce a bias-corrected image.

    When ``log_space=True`` the upsampled field is first exponentiated
    (``exp(upsampled_log_bias)``) before dividing, recovering the linear
    bias field that was log-transformed in :meth:`lowres_func`.

    Works with both dask arrays (``"default"`` method) and plain NumPy
    arrays (``"map_blocks"`` method) because only NumPy-compatible
    operations are used.

    Args:
        fullres_array: Full-resolution array (dask or NumPy).
        upsampled_output: Upsampled bias field with the same shape as
            ``fullres_array`` (dask or NumPy).  Contains log-transformed
            values when ``log_space=True``.
        upsampled_lowres_mask: Optional upsampled mask (0..1) used to
            constrain where correction is applied. When provided,
            correction is blended as
            ``corrected * mask + original * (1 - mask)``.

    Returns:
        Bias-corrected full-resolution array (same type as inputs).
    """
    epsilon = np.finfo(np.float32).eps
    if self.log_space:
        bias = np.exp(upsampled_output)
    else:
        bias = upsampled_output
    corrected = fullres_array / np.maximum(bias, epsilon)

    if upsampled_lowres_mask is None:
        return corrected
    else:
        return np.where(upsampled_lowres_mask > 0.5, corrected, 0)
zarrnii.plugins.N4BiasFieldApply.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield_apply.py 
195
196
197
198
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "N4 Bias Field Apply"
zarrnii.plugins.N4BiasFieldApply.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield_apply.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    log_note = (
        " Bias field is interpolated in log space for smoother upsampling."
        if self.log_space
        else ""
    )
    return (
        "Apply a pre-computed N4 bias field to full-resolution data. "
        "The bias field must be supplied as a pre-computed downsampled "
        "image via the ``lowres_znimg`` argument. Optionally provide "
        "``lowres_mask`` to constrain correction to masked regions; no "
        "ANTsPy dependency "
        f"is required.{log_note}"
    )

zarrnii.plugins.SegmentationCleaner(mask_threshold=50, max_extent=0.15, exclusion_threshold=50)

Segmentation cleaning plugin using multi-resolution processing.

This plugin removes artifactual objects from segmentations by: 1. At low resolution: performing connected components analysis on a thresholded mask and identifying objects with extent below a threshold 2. At high resolution: applying an exclusion mask to remove these objects from the full-resolution segmentation

The extent of a region is defined as the ratio of pixels in the region to pixels in the total bounding box. Objects with low extent (e.g., < 0.15) are typically large artifactual objects with sparse coverage.

Parameters:

  • mask_threshold (float, default: 50 ) –

    Threshold for creating initial binary mask (default: 50)

  • max_extent (float, default: 0.15 ) –

    Maximum extent threshold for exclusion (default: 0.15)

  • exclusion_threshold (float, default: 50 ) –

    Threshold for upsampled exclusion mask (default: 50)

Initialize segmentation cleaner plugin.

Parameters:

  • mask_threshold (float, default: 50 ) –

    Threshold for creating initial binary mask from input

  • max_extent (float, default: 0.15 ) –

    Maximum extent to include in exclusion mask (objects with extent < max_extent are considered artifactual)

  • exclusion_threshold (float, default: 50 ) –

    Threshold for applying upsampled exclusion mask

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    mask_threshold: float = 50,
    max_extent: float = 0.15,
    exclusion_threshold: float = 50,
):
    """
    Initialize segmentation cleaner plugin.

    Args:
        mask_threshold: Threshold for creating initial binary mask from input
        max_extent: Maximum extent to include in exclusion mask (objects with
                   extent < max_extent are considered artifactual)
        exclusion_threshold: Threshold for applying upsampled exclusion mask
    """
    self.mask_threshold = mask_threshold
    self.max_extent = max_extent
    self.exclusion_threshold = exclusion_threshold

Functions

zarrnii.plugins.SegmentationCleaner.lowres_func(lowres_array)

Create exclusion mask from low-resolution segmentation data.

This function performs connected components analysis on thresholded low-resolution data and creates an exclusion mask containing objects with extent below the threshold.

Parameters:

  • lowres_array (ndarray) –

    Downsampled segmentation image

Returns:

  • ndarray –

    Exclusion mask at low resolution (uint8, values 0 or 100)

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
 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
 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
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Create exclusion mask from low-resolution segmentation data.

    This function performs connected components analysis on thresholded
    low-resolution data and creates an exclusion mask containing objects
    with extent below the threshold.

    Args:
        lowres_array: Downsampled segmentation image

    Returns:
        Exclusion mask at low resolution (uint8, values 0 or 100)
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Process connected components and create exclusion mask
    if work_array.ndim == 2:
        # 2D case
        mask_img = work_array > self.mask_threshold
        conncomp, nlabels = label(mask_img, return_num=True)
        props = regionprops(conncomp)
        keep_labels = {r.label for r in props if r.extent < self.max_extent}
        exclude_mask = np.isin(conncomp, list(keep_labels))
        exclude_mask = exclude_mask.astype("uint8") * 100
    else:
        # 3D or batched 3D case
        if work_array.ndim == 4:
            # Batched 3D volumes - process each separately
            exclude_mask = np.zeros_like(work_array, dtype="uint8")
            for i in range(work_array.shape[0]):
                mask_img = work_array[i] > self.mask_threshold
                conncomp, nlabels = label(mask_img, return_num=True)
                props = regionprops(conncomp)
                keep_labels = {r.label for r in props if r.extent < self.max_extent}
                batch_exclude = np.isin(conncomp, list(keep_labels))
                exclude_mask[i] = batch_exclude.astype("uint8") * 100
        else:
            # Single 3D volume
            mask_img = work_array > self.mask_threshold
            conncomp, nlabels = label(mask_img, return_num=True)
            props = regionprops(conncomp)
            keep_labels = {r.label for r in props if r.extent < self.max_extent}
            exclude_mask = np.isin(conncomp, list(keep_labels))
            exclude_mask = exclude_mask.astype("uint8") * 100

    # Reshape back to original shape if needed
    if original_shape != exclude_mask.shape:
        exclude_mask = exclude_mask.reshape(original_shape)

    return exclude_mask
zarrnii.plugins.SegmentationCleaner.highres_func(fullres_array, upsampled_output)

Apply exclusion mask to full-resolution segmentation data.

This function takes the upsampled exclusion mask and applies it to the full-resolution segmentation by zeroing out the excluded regions.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution segmentation array (dask or NumPy)

  • upsampled_output –

    Upsampled exclusion mask (same shape as fullres; dask or NumPy)

Returns:

  • –

    Cleaned full-resolution segmentation array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply exclusion mask to full-resolution segmentation data.

    This function takes the upsampled exclusion mask and applies it to
    the full-resolution segmentation by zeroing out the excluded regions.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution segmentation array (dask or NumPy)
        upsampled_output: Upsampled exclusion mask (same shape as fullres;
            dask or NumPy)

    Returns:
        Cleaned full-resolution segmentation array (same type as inputs)
    """
    # Threshold the upsampled exclusion mask.
    # Values >= exclusion_threshold (e.g., 50) indicate regions to exclude.
    exclusion_mask = upsampled_output >= self.exclusion_threshold

    # Apply mask: set excluded regions to zero.
    # np.where works for both dask arrays (via __array_ufunc__) and NumPy.
    cleaned_array = np.where(exclusion_mask, 0, fullres_array)

    return cleaned_array
zarrnii.plugins.SegmentationCleaner.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
153
154
155
156
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "Segmentation Cleaner"
zarrnii.plugins.SegmentationCleaner.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
158
159
160
161
162
163
164
165
166
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution segmentation cleaning. Identifies and removes "
        "artifactual objects by performing connected components analysis at "
        "low resolution, filtering by extent, and applying exclusion mask to "
        "full resolution data."
    )

zarrnii.plugins.LocalOtsuSegmentation(nbins=256)

Local Otsu thresholding segmentation plugin.

This plugin uses Otsu's method to automatically determine an optimal threshold for binary image segmentation. The method assumes a bimodal histogram and finds the threshold that minimizes intra-class variance. The threshold is computed locally for each processing block, making it suitable for images with varying illumination or contrast.

Parameters:

  • nbins (int, default: 256 ) –

    Number of bins for histogram computation (default: 256)

Initialize local Otsu segmentation plugin.

Parameters:

  • nbins (int, default: 256 ) –

    Number of bins for histogram computation

Source code in zarrnii/plugins/segmentation/local_otsu.py 
32
33
34
35
36
37
38
39
def __init__(self, nbins: int = 256):
    """
    Initialize local Otsu segmentation plugin.

    Args:
        nbins: Number of bins for histogram computation
    """
    self.nbins = nbins

Functions

zarrnii.plugins.LocalOtsuSegmentation.segment(image, metadata=None)

Segment image using local Otsu thresholding.

Parameters:

  • image (ndarray) –

    Input image as numpy array

  • metadata (Optional[Dict[str, Any]], default: None ) –

    Optional metadata (unused in Otsu method)

Returns:

  • ndarray –

    Binary segmentation mask as numpy array with same shape as input.

  • ndarray –

    Values are 0 (background) and 1 (foreground).

Raises:

  • ValueError –

    If input image is empty or has invalid dimensions

Source code in zarrnii/plugins/segmentation/local_otsu.py 
 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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@hookimpl
def segment(
    self, image: np.ndarray, metadata: Optional[Dict[str, Any]] = None
) -> np.ndarray:
    """
    Segment image using local Otsu thresholding.

    Args:
        image: Input image as numpy array
        metadata: Optional metadata (unused in Otsu method)

    Returns:
        Binary segmentation mask as numpy array with same shape as input.
        Values are 0 (background) and 1 (foreground).

    Raises:
        ValueError: If input image is empty or has invalid dimensions
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    if image.ndim < 2:
        raise ValueError("Input image must be at least 2D")

    # Store original shape for output
    original_shape = image.shape

    # Ensure image is in a suitable format for Otsu thresholding
    if image.dtype == bool:
        # Already binary, return as-is
        return image.astype(np.uint8)

    # Handle multi-dimensional images
    work_image = image

    # For images with more than 3 dimensions, flatten extra dimensions
    if work_image.ndim > 3:
        # Reshape to 3D by flattening leading dimensions
        leading_dims = work_image.shape[:-2]  # All dimensions except last 2
        work_image = work_image.reshape(-1, *work_image.shape[-2:])

    if work_image.ndim == 3:
        # For 3D images, process the first slice/channel
        if work_image.shape[0] <= 4:  # Likely channels-first format
            work_image = work_image[0]
        else:
            # If first dimension is large, likely depth dimension, take middle slice
            mid_slice = work_image.shape[0] // 2
            work_image = work_image[mid_slice]

    # Now work_image should be 2D
    try:
        threshold = threshold_otsu(work_image, nbins=self.nbins)
    except ValueError as e:
        # Handle edge cases where Otsu fails (e.g., constant image)
        if "all values are identical" in str(e).lower() or np.all(
            work_image == work_image.flat[0]
        ):
            # Return all zeros for constant images with original shape
            return np.zeros(original_shape, dtype=np.uint8)
        else:
            raise e

    # Apply threshold to original image
    binary_mask = image > threshold

    return binary_mask.astype(np.uint8)
zarrnii.plugins.LocalOtsuSegmentation.segmentation_plugin_name()

Return the name of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/local_otsu.py 
109
110
111
112
@hookimpl
def segmentation_plugin_name(self) -> str:
    """Return the name of the segmentation algorithm."""
    return "Local Otsu Thresholding"
zarrnii.plugins.LocalOtsuSegmentation.segmentation_plugin_description()

Return a description of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/local_otsu.py 
114
115
116
117
118
119
120
121
122
@hookimpl
def segmentation_plugin_description(self) -> str:
    """Return a description of the segmentation algorithm."""
    return (
        "Local Otsu's automatic threshold selection method for binary segmentation. "
        "Finds the threshold that minimizes intra-class variance assuming a "
        "bimodal intensity distribution. Threshold is computed locally for each "
        "processing block, suitable for images with varying illumination."
    )
zarrnii.plugins.LocalOtsuSegmentation.get_threshold(image)

Get the Otsu threshold value without applying segmentation.

Parameters:

  • image (ndarray) –

    Input image as numpy array

Returns:

  • float –

    Computed Otsu threshold value

Source code in zarrnii/plugins/segmentation/local_otsu.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def get_threshold(self, image: np.ndarray) -> float:
    """
    Get the Otsu threshold value without applying segmentation.

    Args:
        image: Input image as numpy array

    Returns:
        Computed Otsu threshold value
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    # Handle multi-channel images by taking the first channel if needed
    if image.ndim > 3:
        while image.ndim > 3:
            image = image[0]

    if image.ndim == 3 and image.shape[0] <= 4:
        image = image[0]

    return threshold_otsu(image, nbins=self.nbins)

zarrnii.plugins.ThresholdSegmentation(thresholds, inclusive=True)

Threshold-based segmentation plugin.

This plugin applies threshold-based segmentation using either a single threshold value or multiple threshold values to create labeled regions. It can use manually specified thresholds or thresholds computed from analysis functions like Otsu multi-thresholding.

Parameters:

  • thresholds (Union[float, List[float]]) –

    Single threshold value or list of threshold values. For single threshold, creates binary segmentation (0/1). For multiple thresholds, creates multi-class segmentation (0/1/2/...).

  • inclusive (bool, default: True ) –

    Whether thresholds are inclusive (default: True). If True, pixels >= threshold are labeled as foreground. If False, pixels > threshold are labeled as foreground.

Initialize threshold segmentation plugin.

Parameters:

  • thresholds (Union[float, List[float]]) –

    Single threshold or list of thresholds

  • inclusive (bool, default: True ) –

    Whether thresholds are inclusive (>= vs >)

Source code in zarrnii/plugins/segmentation/threshold.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, thresholds: Union[float, List[float]], inclusive: bool = True):
    """
    Initialize threshold segmentation plugin.

    Args:
        thresholds: Single threshold or list of thresholds
        inclusive: Whether thresholds are inclusive (>= vs >)
    """
    # Normalize thresholds to always be a list
    if isinstance(thresholds, (int, float)):
        self.thresholds = [float(thresholds)]
    else:
        self.thresholds = [float(t) for t in thresholds]

    # Sort thresholds to ensure proper ordering
    self.thresholds = sorted(self.thresholds)
    self.inclusive = inclusive

Functions

zarrnii.plugins.ThresholdSegmentation.segment(image, metadata=None)

Segment image using threshold values.

Parameters:

  • image (ndarray) –

    Input image as numpy array

  • metadata (Optional[Dict[str, Any]], default: None ) –

    Optional metadata (unused in threshold method)

Returns:

  • ndarray –

    Labeled segmentation mask as numpy array with same shape as input.

  • ndarray –

    Values are 0 (background), 1, 2, ... up to len(thresholds) classes.

Raises:

  • ValueError –

    If input image is empty

Source code in zarrnii/plugins/segmentation/threshold.py 
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
@hookimpl
def segment(
    self, image: np.ndarray, metadata: Optional[Dict[str, Any]] = None
) -> np.ndarray:
    """
    Segment image using threshold values.

    Args:
        image: Input image as numpy array
        metadata: Optional metadata (unused in threshold method)

    Returns:
        Labeled segmentation mask as numpy array with same shape as input.
        Values are 0 (background), 1, 2, ... up to len(thresholds) classes.

    Raises:
        ValueError: If input image is empty
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    # Initialize result with zeros (background class)
    result = np.zeros(image.shape, dtype=np.uint8)

    # Apply thresholds to create labeled regions
    for i, threshold in enumerate(self.thresholds):
        if self.inclusive:
            mask = image >= threshold
        else:
            mask = image > threshold

        # Assign class label (i+1) to pixels above threshold
        result[mask] = i + 1

    return result
zarrnii.plugins.ThresholdSegmentation.segmentation_plugin_name()

Return the name of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/threshold.py 
89
90
91
92
93
94
95
@hookimpl
def segmentation_plugin_name(self) -> str:
    """Return the name of the segmentation algorithm."""
    if len(self.thresholds) == 1:
        return "Binary Threshold"
    else:
        return "Multi-level Threshold"
zarrnii.plugins.ThresholdSegmentation.segmentation_plugin_description()

Return a description of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/threshold.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@hookimpl
def segmentation_plugin_description(self) -> str:
    """Return a description of the segmentation algorithm."""
    if len(self.thresholds) == 1:
        op = ">=" if self.inclusive else ">"
        return (
            f"Binary threshold segmentation using threshold = {self.thresholds[0]}. "
            f"Pixels {op} threshold are labeled as foreground (1), others as background (0)."
        )
    else:
        op = ">=" if self.inclusive else ">"
        return (
            f"Multi-level threshold segmentation using {len(self.thresholds)} thresholds: "
            f"{self.thresholds}. Creates {len(self.thresholds) + 1} labeled regions based on "
            f"which thresholds each pixel exceeds (using {op} comparison)."
        )
zarrnii.plugins.ThresholdSegmentation.get_thresholds()

Get the threshold values used by this plugin.

Returns:

  • List[float] –

    List of threshold values

Source code in zarrnii/plugins/segmentation/threshold.py 
118
119
120
121
122
123
124
125
def get_thresholds(self) -> List[float]:
    """
    Get the threshold values used by this plugin.

    Returns:
        List of threshold values
    """
    return self.thresholds.copy()

Functions

zarrnii.plugins.get_global_plugin_manager()

Get the global plugin manager instance.

Returns:

  • PluginManager –

    Global PluginManager instance

Source code in zarrnii/plugins/plugin_manager.py 
34
35
36
37
38
39
40
41
42
43
44
def get_global_plugin_manager() -> pluggy.PluginManager:
    """
    Get the global plugin manager instance.

    Returns:
        Global PluginManager instance
    """
    global _plugin_manager
    if _plugin_manager is None:
        _plugin_manager = get_plugin_manager()
    return _plugin_manager

zarrnii.plugins.get_plugin_manager()

Get or create the ZarrNii plugin manager.

Returns:

  • PluginManager –

    PluginManager instance configured for ZarrNii plugins

Source code in zarrnii/plugins/plugin_manager.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_plugin_manager() -> pluggy.PluginManager:
    """
    Get or create the ZarrNii plugin manager.

    Returns:
        PluginManager instance configured for ZarrNii plugins
    """
    pm = pluggy.PluginManager("zarrnii")
    pm.add_hookspecs(ZarrNiiSpec)
    # Discover and register external plugins declared via setuptools entry points
    # under the "zarrnii" group (matching the pluggy project name).
    pm.load_setuptools_entrypoints("zarrnii")
    return pm

Hook specifications

Plugin hook specifications for ZarrNii plugins.

This module defines the hook specifications that plugins must implement.

Classes

zarrnii.plugins.hookspecs.ZarrNiiSpec

Hook specifications for ZarrNii plugins.

Plugin authors should implement any subset of these hooks as plain methods decorated with @hookimpl from :mod:zarrnii.plugins.

Example::

from zarrnii.plugins import hookimpl

class MyPlugin:
    @hookimpl
    def segment(self, image, metadata=None):
        ...

Functions

zarrnii.plugins.hookspecs.ZarrNiiSpec.segment(image, metadata=None)

Segment an image and return a binary or labeled mask.

Parameters:

  • image –

    Input image as numpy array.

  • metadata –

    Optional metadata dictionary containing image information.

Returns:

  • –

    Segmented image as numpy array.

Source code in zarrnii/plugins/hookspecs.py 
28
29
30
31
32
33
34
35
36
37
38
@hookspec
def segment(self, image, metadata=None):
    """Segment an image and return a binary or labeled mask.

    Args:
        image: Input image as numpy array.
        metadata: Optional metadata dictionary containing image information.

    Returns:
        Segmented image as numpy array.
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.segmentation_plugin_name()

Return the name of the segmentation algorithm.

Returns:

  • str –

    String name of the algorithm.

Source code in zarrnii/plugins/hookspecs.py 
40
41
42
43
44
45
46
@hookspec
def segmentation_plugin_name(self) -> str:
    """Return the name of the segmentation algorithm.

    Returns:
        String name of the algorithm.
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.segmentation_plugin_description()

Return a description of the segmentation algorithm.

Returns:

  • str –

    String description of the algorithm.

Source code in zarrnii/plugins/hookspecs.py 
48
49
50
51
52
53
54
@hookspec
def segmentation_plugin_description(self) -> str:
    """Return a description of the segmentation algorithm.

    Returns:
        String description of the algorithm.
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.scaled_processing_plugin_name()

Return the name of the scaled processing algorithm.

Returns:

  • str –

    String name of the algorithm.

Source code in zarrnii/plugins/hookspecs.py 
56
57
58
59
60
61
62
@hookspec
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the scaled processing algorithm.

    Returns:
        String name of the algorithm.
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.scaled_processing_plugin_description()

Return a description of the scaled processing algorithm.

Returns:

  • str –

    String description of the algorithm.

Source code in zarrnii/plugins/hookspecs.py 
64
65
66
67
68
69
70
@hookspec
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the scaled processing algorithm.

    Returns:
        String description of the algorithm.
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.lowres_func(lowres_array)

Process low-resolution data and return the result.

This function operates on a downsampled numpy array and computes the algorithm output that will be upsampled and applied to the full-resolution data.

Parameters:

  • lowres_array –

    Downsampled input image as numpy array.

Returns:

  • –

    Low-resolution output array (e.g., bias field, correction map).

Source code in zarrnii/plugins/hookspecs.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@hookspec
def lowres_func(self, lowres_array):
    """Process low-resolution data and return the result.

    This function operates on a downsampled numpy array and computes
    the algorithm output that will be upsampled and applied to the
    full-resolution data.

    Args:
        lowres_array: Downsampled input image as numpy array.

    Returns:
        Low-resolution output array (e.g., bias field, correction map).
    """
zarrnii.plugins.hookspecs.ZarrNiiSpec.highres_func(fullres_array, upsampled_output)

Apply upsampled output to full-resolution data blockwise.

Parameters:

  • fullres_array –

    Full-resolution dask array.

  • upsampled_output –

    Upsampled output (same shape as fullres_array).

Returns:

  • –

    Processed full-resolution dask array.

Source code in zarrnii/plugins/hookspecs.py 
87
88
89
90
91
92
93
94
95
96
97
@hookspec
def highres_func(self, fullres_array, upsampled_output):
    """Apply upsampled output to full-resolution data blockwise.

    Args:
        fullres_array: Full-resolution dask array.
        upsampled_output: Upsampled output (same shape as fullres_array).

    Returns:
        Processed full-resolution dask array.
    """

Plugin manager

Plugin manager for ZarrNii plugins.

This module provides the plugin manager that discovers and manages plugins using the pluggy framework.

Classes

Functions

zarrnii.plugins.plugin_manager.get_plugin_manager()

Get or create the ZarrNii plugin manager.

Returns:

  • PluginManager –

    PluginManager instance configured for ZarrNii plugins

Source code in zarrnii/plugins/plugin_manager.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
def get_plugin_manager() -> pluggy.PluginManager:
    """
    Get or create the ZarrNii plugin manager.

    Returns:
        PluginManager instance configured for ZarrNii plugins
    """
    pm = pluggy.PluginManager("zarrnii")
    pm.add_hookspecs(ZarrNiiSpec)
    # Discover and register external plugins declared via setuptools entry points
    # under the "zarrnii" group (matching the pluggy project name).
    pm.load_setuptools_entrypoints("zarrnii")
    return pm

zarrnii.plugins.plugin_manager.get_global_plugin_manager()

Get the global plugin manager instance.

Returns:

  • PluginManager –

    Global PluginManager instance

Source code in zarrnii/plugins/plugin_manager.py 
34
35
36
37
38
39
40
41
42
43
44
def get_global_plugin_manager() -> pluggy.PluginManager:
    """
    Get the global plugin manager instance.

    Returns:
        Global PluginManager instance
    """
    global _plugin_manager
    if _plugin_manager is None:
        _plugin_manager = get_plugin_manager()
    return _plugin_manager

Built-in segmentation plugins

Local Otsu thresholding segmentation plugin.

This module implements Otsu's automatic threshold selection method for binary image segmentation, applied locally to each processing block.

Classes

zarrnii.plugins.segmentation.local_otsu.LocalOtsuSegmentation(nbins=256)

Local Otsu thresholding segmentation plugin.

This plugin uses Otsu's method to automatically determine an optimal threshold for binary image segmentation. The method assumes a bimodal histogram and finds the threshold that minimizes intra-class variance. The threshold is computed locally for each processing block, making it suitable for images with varying illumination or contrast.

Parameters:

  • nbins (int, default: 256 ) –

    Number of bins for histogram computation (default: 256)

Initialize local Otsu segmentation plugin.

Parameters:

  • nbins (int, default: 256 ) –

    Number of bins for histogram computation

Source code in zarrnii/plugins/segmentation/local_otsu.py 
32
33
34
35
36
37
38
39
def __init__(self, nbins: int = 256):
    """
    Initialize local Otsu segmentation plugin.

    Args:
        nbins: Number of bins for histogram computation
    """
    self.nbins = nbins

Functions

zarrnii.plugins.segmentation.local_otsu.LocalOtsuSegmentation.segment(image, metadata=None)

Segment image using local Otsu thresholding.

Parameters:

  • image (ndarray) –

    Input image as numpy array

  • metadata (Optional[Dict[str, Any]], default: None ) –

    Optional metadata (unused in Otsu method)

Returns:

  • ndarray –

    Binary segmentation mask as numpy array with same shape as input.

  • ndarray –

    Values are 0 (background) and 1 (foreground).

Raises:

  • ValueError –

    If input image is empty or has invalid dimensions

Source code in zarrnii/plugins/segmentation/local_otsu.py 
 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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@hookimpl
def segment(
    self, image: np.ndarray, metadata: Optional[Dict[str, Any]] = None
) -> np.ndarray:
    """
    Segment image using local Otsu thresholding.

    Args:
        image: Input image as numpy array
        metadata: Optional metadata (unused in Otsu method)

    Returns:
        Binary segmentation mask as numpy array with same shape as input.
        Values are 0 (background) and 1 (foreground).

    Raises:
        ValueError: If input image is empty or has invalid dimensions
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    if image.ndim < 2:
        raise ValueError("Input image must be at least 2D")

    # Store original shape for output
    original_shape = image.shape

    # Ensure image is in a suitable format for Otsu thresholding
    if image.dtype == bool:
        # Already binary, return as-is
        return image.astype(np.uint8)

    # Handle multi-dimensional images
    work_image = image

    # For images with more than 3 dimensions, flatten extra dimensions
    if work_image.ndim > 3:
        # Reshape to 3D by flattening leading dimensions
        leading_dims = work_image.shape[:-2]  # All dimensions except last 2
        work_image = work_image.reshape(-1, *work_image.shape[-2:])

    if work_image.ndim == 3:
        # For 3D images, process the first slice/channel
        if work_image.shape[0] <= 4:  # Likely channels-first format
            work_image = work_image[0]
        else:
            # If first dimension is large, likely depth dimension, take middle slice
            mid_slice = work_image.shape[0] // 2
            work_image = work_image[mid_slice]

    # Now work_image should be 2D
    try:
        threshold = threshold_otsu(work_image, nbins=self.nbins)
    except ValueError as e:
        # Handle edge cases where Otsu fails (e.g., constant image)
        if "all values are identical" in str(e).lower() or np.all(
            work_image == work_image.flat[0]
        ):
            # Return all zeros for constant images with original shape
            return np.zeros(original_shape, dtype=np.uint8)
        else:
            raise e

    # Apply threshold to original image
    binary_mask = image > threshold

    return binary_mask.astype(np.uint8)
zarrnii.plugins.segmentation.local_otsu.LocalOtsuSegmentation.segmentation_plugin_name()

Return the name of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/local_otsu.py 
109
110
111
112
@hookimpl
def segmentation_plugin_name(self) -> str:
    """Return the name of the segmentation algorithm."""
    return "Local Otsu Thresholding"
zarrnii.plugins.segmentation.local_otsu.LocalOtsuSegmentation.segmentation_plugin_description()

Return a description of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/local_otsu.py 
114
115
116
117
118
119
120
121
122
@hookimpl
def segmentation_plugin_description(self) -> str:
    """Return a description of the segmentation algorithm."""
    return (
        "Local Otsu's automatic threshold selection method for binary segmentation. "
        "Finds the threshold that minimizes intra-class variance assuming a "
        "bimodal intensity distribution. Threshold is computed locally for each "
        "processing block, suitable for images with varying illumination."
    )
zarrnii.plugins.segmentation.local_otsu.LocalOtsuSegmentation.get_threshold(image)

Get the Otsu threshold value without applying segmentation.

Parameters:

  • image (ndarray) –

    Input image as numpy array

Returns:

  • float –

    Computed Otsu threshold value

Source code in zarrnii/plugins/segmentation/local_otsu.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def get_threshold(self, image: np.ndarray) -> float:
    """
    Get the Otsu threshold value without applying segmentation.

    Args:
        image: Input image as numpy array

    Returns:
        Computed Otsu threshold value
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    # Handle multi-channel images by taking the first channel if needed
    if image.ndim > 3:
        while image.ndim > 3:
            image = image[0]

    if image.ndim == 3 and image.shape[0] <= 4:
        image = image[0]

    return threshold_otsu(image, nbins=self.nbins)

Threshold segmentation plugin.

This module implements threshold-based segmentation that can use either manual threshold values or computed thresholds (e.g., from Otsu analysis).

Classes

zarrnii.plugins.segmentation.threshold.ThresholdSegmentation(thresholds, inclusive=True)

Threshold-based segmentation plugin.

This plugin applies threshold-based segmentation using either a single threshold value or multiple threshold values to create labeled regions. It can use manually specified thresholds or thresholds computed from analysis functions like Otsu multi-thresholding.

Parameters:

  • thresholds (Union[float, List[float]]) –

    Single threshold value or list of threshold values. For single threshold, creates binary segmentation (0/1). For multiple thresholds, creates multi-class segmentation (0/1/2/...).

  • inclusive (bool, default: True ) –

    Whether thresholds are inclusive (default: True). If True, pixels >= threshold are labeled as foreground. If False, pixels > threshold are labeled as foreground.

Initialize threshold segmentation plugin.

Parameters:

  • thresholds (Union[float, List[float]]) –

    Single threshold or list of thresholds

  • inclusive (bool, default: True ) –

    Whether thresholds are inclusive (>= vs >)

Source code in zarrnii/plugins/segmentation/threshold.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, thresholds: Union[float, List[float]], inclusive: bool = True):
    """
    Initialize threshold segmentation plugin.

    Args:
        thresholds: Single threshold or list of thresholds
        inclusive: Whether thresholds are inclusive (>= vs >)
    """
    # Normalize thresholds to always be a list
    if isinstance(thresholds, (int, float)):
        self.thresholds = [float(thresholds)]
    else:
        self.thresholds = [float(t) for t in thresholds]

    # Sort thresholds to ensure proper ordering
    self.thresholds = sorted(self.thresholds)
    self.inclusive = inclusive

Functions

zarrnii.plugins.segmentation.threshold.ThresholdSegmentation.segment(image, metadata=None)

Segment image using threshold values.

Parameters:

  • image (ndarray) –

    Input image as numpy array

  • metadata (Optional[Dict[str, Any]], default: None ) –

    Optional metadata (unused in threshold method)

Returns:

  • ndarray –

    Labeled segmentation mask as numpy array with same shape as input.

  • ndarray –

    Values are 0 (background), 1, 2, ... up to len(thresholds) classes.

Raises:

  • ValueError –

    If input image is empty

Source code in zarrnii/plugins/segmentation/threshold.py 
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
@hookimpl
def segment(
    self, image: np.ndarray, metadata: Optional[Dict[str, Any]] = None
) -> np.ndarray:
    """
    Segment image using threshold values.

    Args:
        image: Input image as numpy array
        metadata: Optional metadata (unused in threshold method)

    Returns:
        Labeled segmentation mask as numpy array with same shape as input.
        Values are 0 (background), 1, 2, ... up to len(thresholds) classes.

    Raises:
        ValueError: If input image is empty
    """
    if image.size == 0:
        raise ValueError("Input image is empty")

    # Initialize result with zeros (background class)
    result = np.zeros(image.shape, dtype=np.uint8)

    # Apply thresholds to create labeled regions
    for i, threshold in enumerate(self.thresholds):
        if self.inclusive:
            mask = image >= threshold
        else:
            mask = image > threshold

        # Assign class label (i+1) to pixels above threshold
        result[mask] = i + 1

    return result
zarrnii.plugins.segmentation.threshold.ThresholdSegmentation.segmentation_plugin_name()

Return the name of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/threshold.py 
89
90
91
92
93
94
95
@hookimpl
def segmentation_plugin_name(self) -> str:
    """Return the name of the segmentation algorithm."""
    if len(self.thresholds) == 1:
        return "Binary Threshold"
    else:
        return "Multi-level Threshold"
zarrnii.plugins.segmentation.threshold.ThresholdSegmentation.segmentation_plugin_description()

Return a description of the segmentation algorithm.

Source code in zarrnii/plugins/segmentation/threshold.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@hookimpl
def segmentation_plugin_description(self) -> str:
    """Return a description of the segmentation algorithm."""
    if len(self.thresholds) == 1:
        op = ">=" if self.inclusive else ">"
        return (
            f"Binary threshold segmentation using threshold = {self.thresholds[0]}. "
            f"Pixels {op} threshold are labeled as foreground (1), others as background (0)."
        )
    else:
        op = ">=" if self.inclusive else ">"
        return (
            f"Multi-level threshold segmentation using {len(self.thresholds)} thresholds: "
            f"{self.thresholds}. Creates {len(self.thresholds) + 1} labeled regions based on "
            f"which thresholds each pixel exceeds (using {op} comparison)."
        )
zarrnii.plugins.segmentation.threshold.ThresholdSegmentation.get_thresholds()

Get the threshold values used by this plugin.

Returns:

  • List[float] –

    List of threshold values

Source code in zarrnii/plugins/segmentation/threshold.py 
118
119
120
121
122
123
124
125
def get_thresholds(self) -> List[float]:
    """
    Get the threshold values used by this plugin.

    Returns:
        List of threshold values
    """
    return self.thresholds.copy()

Built-in scaled-processing plugins

Gaussian Bias field correction plugin using multi-resolution processing.

This module implements a bias field correction plugin that estimates the bias field at low resolution and applies it to full resolution data.

Classes

zarrnii.plugins.scaled_processing.gaussian_biasfield.GaussianBiasFieldCorrection(sigma=5.0, mode='reflect')

Bias field correction plugin using multi-resolution processing.

This plugin estimates a smooth bias field at low resolution using simple smoothing (as a placeholder for more sophisticated methods like N4) and applies the correction to full resolution data by division.

Parameters:

  • sigma (float, default: 5.0 ) –

    Standard deviation for Gaussian smoothing (default: 5.0)

  • mode (str, default: 'reflect' ) –

    Boundary condition for smoothing (default: 'reflect')

Initialize bias field correction plugin.

Parameters:

  • sigma (float, default: 5.0 ) –

    Standard deviation for Gaussian smoothing

  • mode (str, default: 'reflect' ) –

    Boundary condition for smoothing

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
29
30
31
32
33
34
35
36
37
38
def __init__(self, sigma: float = 5.0, mode: str = "reflect"):
    """
    Initialize bias field correction plugin.

    Args:
        sigma: Standard deviation for Gaussian smoothing
        mode: Boundary condition for smoothing
    """
    self.sigma = sigma
    self.mode = mode

Functions

zarrnii.plugins.scaled_processing.gaussian_biasfield.GaussianBiasFieldCorrection.lowres_func(lowres_array)

Estimate bias field from low-resolution data.

This is a simplified bias field estimation using Gaussian smoothing. In practice, this could be replaced with more sophisticated methods like N4 bias field correction.

Parameters:

  • lowres_array (ndarray) –

    Downsampled input image

Returns:

  • ndarray –

    Estimated bias field at low resolution

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
 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
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Estimate bias field from low-resolution data.

    This is a simplified bias field estimation using Gaussian smoothing.
    In practice, this could be replaced with more sophisticated methods
    like N4 bias field correction.

    Args:
        lowres_array: Downsampled input image

    Returns:
        Estimated bias field at low resolution
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    # Handle different array dimensions
    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        leading_shape = work_array.shape[:-3]
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Ensure we're working with float data
    if work_array.dtype.kind in ["i", "u"]:  # integer types
        work_array = work_array.astype(np.float32)

    # Simple bias field estimation: smooth the image to get low-frequency components
    if work_array.ndim == 2:
        smoothed = ndimage.gaussian_filter(
            work_array, sigma=self.sigma, mode=self.mode
        )
    else:
        # Apply smoothing to each volume if we have multiple
        if work_array.ndim == 4:  # batched 3D volumes
            smoothed = np.zeros_like(work_array)
            for i in range(work_array.shape[0]):
                smoothed[i] = ndimage.gaussian_filter(
                    work_array[i], sigma=self.sigma, mode=self.mode
                )
        else:  # single 3D volume
            smoothed = ndimage.gaussian_filter(
                work_array, sigma=self.sigma, mode=self.mode
            )

    # Reshape back to original shape if needed
    if original_shape != smoothed.shape:
        smoothed = smoothed.reshape(original_shape)

    # Avoid division by zero by adding small epsilon
    smoothed = np.maximum(smoothed, np.finfo(smoothed.dtype).eps)

    return smoothed
zarrnii.plugins.scaled_processing.gaussian_biasfield.GaussianBiasFieldCorrection.highres_func(fullres_array, upsampled_output)

Apply bias field correction to full-resolution data.

This function takes the upsampled bias field (same size as fullres_array) and applies it to the full-resolution data by division.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution array (dask or NumPy)

  • upsampled_output –

    Upsampled bias field (same shape as fullres_array; dask or NumPy)

Returns:

  • –

    Bias-corrected full-resolution array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply bias field correction to full-resolution data.

    This function takes the upsampled bias field (same size as fullres_array)
    and applies it to the full-resolution data by division.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution array (dask or NumPy)
        upsampled_output: Upsampled bias field (same shape as fullres_array;
            dask or NumPy)

    Returns:
        Bias-corrected full-resolution array (same type as inputs)
    """
    # Avoid division by zero by clamping the bias field to a small epsilon.
    # np.maximum works for both dask arrays (via __array_ufunc__) and NumPy.
    epsilon = np.finfo(np.float32).eps
    corrected_array = fullres_array / np.maximum(upsampled_output, epsilon)

    return corrected_array
zarrnii.plugins.scaled_processing.gaussian_biasfield.GaussianBiasFieldCorrection.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
130
131
132
133
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "Gaussian Bias Field Correction"
zarrnii.plugins.scaled_processing.gaussian_biasfield.GaussianBiasFieldCorrection.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/gaussian_biasfield.py 
135
136
137
138
139
140
141
142
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution bias field correction. Estimates smooth bias field "
        "at low resolution using Gaussian smoothing and applies correction "
        "to full resolution data by division."
    )

N4 Bias field correction plugin using multi-resolution processing.

This module implements a bias field correction plugin that estimates the bias field at low resolution using N4 algorithm from ANTsPy and applies it to full resolution data.

Classes

zarrnii.plugins.scaled_processing.n4_biasfield.N4BiasFieldCorrection(spline_param=[2, 2, 2], convergence={'iters': [50, 50, 50, 50], 'tol': 1e-07}, shrink_factor=1)

N4 bias field correction plugin using multi-resolution processing.

This plugin estimates a smooth bias field at low resolution using the N4 bias field correction algorithm from ANTsPy and applies the correction to full resolution data by division.

Parameters:

  • spline_param (tuple[int, int, int], default: [2, 2, 2] ) –

    Spacing between knots for spline fitting (default: 200)

  • convergence (Optional[Dict[str, Any]], default: {'iters': [50, 50, 50, 50], 'tol': 1e-07} ) –

    Convergence criteria [iters, tol] (default: [50, 0.001])

  • shrink_factor (int, default: 1 ) –

    Shrink factor for processing (default: 1)

Initialize N4 bias field correction plugin.

Parameters:

  • spline_param (tuple[int, int, int], default: [2, 2, 2] ) –

    Spacing between knots for spline fitting

  • convergence (Optional[Dict[str, Any]], default: {'iters': [50, 50, 50, 50], 'tol': 1e-07} ) –

    Convergence criteria dict with 'iters' (list), 'tol'

  • shrink_factor (int, default: 1 ) –

    Shrink factor for processing

Raises:

  • ImportError –

    If antspyx is not installed

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
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
def __init__(
    self,
    spline_param: tuple[int, int, int] = [2, 2, 2],
    convergence: Optional[Dict[str, Any]] = {
        "iters": [50, 50, 50, 50],
        "tol": 1e-07,
    },
    shrink_factor: int = 1,
):
    """
    Initialize N4 bias field correction plugin.

    Args:
        spline_param: Spacing between knots for spline fitting
        convergence: Convergence criteria dict with 'iters' (list), 'tol'
        shrink_factor: Shrink factor for processing

    Raises:
        ImportError: If antspyx is not installed
    """
    if not HAS_ANTSPYX:
        raise ImportError(
            "antspyx is required for N4BiasFieldCorrection. "
            "Install it with: pip install 'zarrnii[n4]' "
            "or pip install antspyx"
        )

    self.spline_param = spline_param
    self.convergence = convergence
    self.shrink_factor = shrink_factor

Functions

zarrnii.plugins.scaled_processing.n4_biasfield.N4BiasFieldCorrection.lowres_func(lowres_array)

Estimate bias field from low-resolution data using N4 algorithm.

This function uses ANTsPy's N4 bias field correction to estimate the bias field at low resolution.

Parameters:

  • lowres_array (ndarray) –

    Downsampled input image

Returns:

  • ndarray –

    Estimated bias field at low resolution

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.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
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Estimate bias field from low-resolution data using N4 algorithm.

    This function uses ANTsPy's N4 bias field correction to estimate
    the bias field at low resolution.

    Args:
        lowres_array: Downsampled input image

    Returns:
        Estimated bias field at low resolution
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    # Handle different array dimensions
    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Ensure we're working with float data
    if work_array.dtype.kind in ["i", "u"]:  # integer types
        work_array = work_array.astype(np.float32)

    # Apply N4 bias field correction
    if work_array.ndim == 2:
        # For 2D data
        ants_img = ants.from_numpy(work_array)
        self.spline_param = [2, 2]
        # Use return_bias_field=True to get the bias field
        bias_result = ants.n4_bias_field_correction(
            ants_img,
            return_bias_field=True,
            spline_param=self.spline_param,
            convergence=self.convergence,
            shrink_factor=self.shrink_factor,
        )
        # Extract the bias field
        bias_field = bias_result.numpy()
    else:
        # Apply N4 to each volume if we have multiple or single 3D volume
        if work_array.ndim == 4:  # batched 3D volumes
            bias_field = np.zeros_like(work_array)
            for i in range(work_array.shape[0]):
                ants_img = ants.from_numpy(work_array[i])
                bias_result = ants.n4_bias_field_correction(
                    ants_img,
                    return_bias_field=True,
                    spline_param=self.spline_param,
                    convergence=self.convergence,
                    shrink_factor=self.shrink_factor,
                )
                bias_field[i] = bias_result.numpy()
        else:  # single 3D volume
            ants_img = ants.from_numpy(work_array)
            bias_result = ants.n4_bias_field_correction(
                ants_img,
                return_bias_field=True,
                spline_param=self.spline_param,
                convergence=self.convergence,
                shrink_factor=self.shrink_factor,
            )
            bias_field = bias_result.numpy()

    # Reshape back to original shape if needed
    if original_shape != bias_field.shape:
        bias_field = bias_field.reshape(original_shape)

    # Avoid division by zero by adding small epsilon
    bias_field = np.maximum(bias_field, np.finfo(bias_field.dtype).eps)

    return bias_field
zarrnii.plugins.scaled_processing.n4_biasfield.N4BiasFieldCorrection.highres_func(fullres_array, upsampled_output)

Apply bias field correction to full-resolution data.

This function takes the upsampled bias field (same size as fullres_array) and applies it to the full-resolution data by division.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution array (dask or NumPy)

  • upsampled_output –

    Upsampled bias field (same shape as fullres; dask or NumPy)

Returns:

  • –

    Bias-corrected full-resolution array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply bias field correction to full-resolution data.

    This function takes the upsampled bias field (same size as
    fullres_array) and applies it to the full-resolution data by division.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution array (dask or NumPy)
        upsampled_output: Upsampled bias field (same shape as fullres;
            dask or NumPy)

    Returns:
        Bias-corrected full-resolution array (same type as inputs)
    """
    # Avoid division by zero by clamping the bias field to a small epsilon.
    # np.maximum works for both dask arrays (via __array_ufunc__) and NumPy.
    epsilon = np.finfo(np.float32).eps
    corrected_array = fullres_array / np.maximum(upsampled_output, epsilon)

    return corrected_array
zarrnii.plugins.scaled_processing.n4_biasfield.N4BiasFieldCorrection.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
180
181
182
183
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "N4 Bias Field Correction"
zarrnii.plugins.scaled_processing.n4_biasfield.N4BiasFieldCorrection.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/n4_biasfield.py 
185
186
187
188
189
190
191
192
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution N4 bias field correction. Estimates smooth bias "
        "field at low resolution using ANTsPy N4 algorithm and applies "
        "correction to full resolution data by division."
    )

Segmentation cleaning plugin using multi-resolution processing.

This module implements a segmentation cleaning plugin that removes artifactual objects from a segmentation by performing connected components analysis at low resolution and filtering by extent, then applying the exclusion mask to full resolution data.

Classes

zarrnii.plugins.scaled_processing.segmentation_cleaner.SegmentationCleaner(mask_threshold=50, max_extent=0.15, exclusion_threshold=50)

Segmentation cleaning plugin using multi-resolution processing.

This plugin removes artifactual objects from segmentations by: 1. At low resolution: performing connected components analysis on a thresholded mask and identifying objects with extent below a threshold 2. At high resolution: applying an exclusion mask to remove these objects from the full-resolution segmentation

The extent of a region is defined as the ratio of pixels in the region to pixels in the total bounding box. Objects with low extent (e.g., < 0.15) are typically large artifactual objects with sparse coverage.

Parameters:

  • mask_threshold (float, default: 50 ) –

    Threshold for creating initial binary mask (default: 50)

  • max_extent (float, default: 0.15 ) –

    Maximum extent threshold for exclusion (default: 0.15)

  • exclusion_threshold (float, default: 50 ) –

    Threshold for upsampled exclusion mask (default: 50)

Initialize segmentation cleaner plugin.

Parameters:

  • mask_threshold (float, default: 50 ) –

    Threshold for creating initial binary mask from input

  • max_extent (float, default: 0.15 ) –

    Maximum extent to include in exclusion mask (objects with extent < max_extent are considered artifactual)

  • exclusion_threshold (float, default: 50 ) –

    Threshold for applying upsampled exclusion mask

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(
    self,
    mask_threshold: float = 50,
    max_extent: float = 0.15,
    exclusion_threshold: float = 50,
):
    """
    Initialize segmentation cleaner plugin.

    Args:
        mask_threshold: Threshold for creating initial binary mask from input
        max_extent: Maximum extent to include in exclusion mask (objects with
                   extent < max_extent are considered artifactual)
        exclusion_threshold: Threshold for applying upsampled exclusion mask
    """
    self.mask_threshold = mask_threshold
    self.max_extent = max_extent
    self.exclusion_threshold = exclusion_threshold

Functions

zarrnii.plugins.scaled_processing.segmentation_cleaner.SegmentationCleaner.lowres_func(lowres_array)

Create exclusion mask from low-resolution segmentation data.

This function performs connected components analysis on thresholded low-resolution data and creates an exclusion mask containing objects with extent below the threshold.

Parameters:

  • lowres_array (ndarray) –

    Downsampled segmentation image

Returns:

  • ndarray –

    Exclusion mask at low resolution (uint8, values 0 or 100)

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
 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
 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
@hookimpl
def lowres_func(self, lowres_array: np.ndarray) -> np.ndarray:
    """
    Create exclusion mask from low-resolution segmentation data.

    This function performs connected components analysis on thresholded
    low-resolution data and creates an exclusion mask containing objects
    with extent below the threshold.

    Args:
        lowres_array: Downsampled segmentation image

    Returns:
        Exclusion mask at low resolution (uint8, values 0 or 100)
    """
    if lowres_array.size == 0:
        raise ValueError("Input array is empty")

    if lowres_array.ndim < 2:
        raise ValueError("Input array must be at least 2D")

    # Store original shape
    original_shape = lowres_array.shape

    # For multi-dimensional arrays, work with the last 3 dimensions
    work_array = lowres_array
    if work_array.ndim > 3:
        # Flatten leading dimensions and work with spatial dimensions
        spatial_shape = work_array.shape[-3:]
        work_array = work_array.reshape(-1, *spatial_shape)

    # Process connected components and create exclusion mask
    if work_array.ndim == 2:
        # 2D case
        mask_img = work_array > self.mask_threshold
        conncomp, nlabels = label(mask_img, return_num=True)
        props = regionprops(conncomp)
        keep_labels = {r.label for r in props if r.extent < self.max_extent}
        exclude_mask = np.isin(conncomp, list(keep_labels))
        exclude_mask = exclude_mask.astype("uint8") * 100
    else:
        # 3D or batched 3D case
        if work_array.ndim == 4:
            # Batched 3D volumes - process each separately
            exclude_mask = np.zeros_like(work_array, dtype="uint8")
            for i in range(work_array.shape[0]):
                mask_img = work_array[i] > self.mask_threshold
                conncomp, nlabels = label(mask_img, return_num=True)
                props = regionprops(conncomp)
                keep_labels = {r.label for r in props if r.extent < self.max_extent}
                batch_exclude = np.isin(conncomp, list(keep_labels))
                exclude_mask[i] = batch_exclude.astype("uint8") * 100
        else:
            # Single 3D volume
            mask_img = work_array > self.mask_threshold
            conncomp, nlabels = label(mask_img, return_num=True)
            props = regionprops(conncomp)
            keep_labels = {r.label for r in props if r.extent < self.max_extent}
            exclude_mask = np.isin(conncomp, list(keep_labels))
            exclude_mask = exclude_mask.astype("uint8") * 100

    # Reshape back to original shape if needed
    if original_shape != exclude_mask.shape:
        exclude_mask = exclude_mask.reshape(original_shape)

    return exclude_mask
zarrnii.plugins.scaled_processing.segmentation_cleaner.SegmentationCleaner.highres_func(fullres_array, upsampled_output)

Apply exclusion mask to full-resolution segmentation data.

This function takes the upsampled exclusion mask and applies it to the full-resolution segmentation by zeroing out the excluded regions.

Works with both dask arrays ("default" method) and plain NumPy arrays ("map_blocks" method) because only NumPy-compatible operations are used.

Parameters:

  • fullres_array –

    Full-resolution segmentation array (dask or NumPy)

  • upsampled_output –

    Upsampled exclusion mask (same shape as fullres; dask or NumPy)

Returns:

  • –

    Cleaned full-resolution segmentation array (same type as inputs)

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
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
@hookimpl
def highres_func(self, fullres_array, upsampled_output):
    """
    Apply exclusion mask to full-resolution segmentation data.

    This function takes the upsampled exclusion mask and applies it to
    the full-resolution segmentation by zeroing out the excluded regions.

    Works with both dask arrays (``"default"`` method) and plain NumPy arrays
    (``"map_blocks"`` method) because only NumPy-compatible operations are used.

    Args:
        fullres_array: Full-resolution segmentation array (dask or NumPy)
        upsampled_output: Upsampled exclusion mask (same shape as fullres;
            dask or NumPy)

    Returns:
        Cleaned full-resolution segmentation array (same type as inputs)
    """
    # Threshold the upsampled exclusion mask.
    # Values >= exclusion_threshold (e.g., 50) indicate regions to exclude.
    exclusion_mask = upsampled_output >= self.exclusion_threshold

    # Apply mask: set excluded regions to zero.
    # np.where works for both dask arrays (via __array_ufunc__) and NumPy.
    cleaned_array = np.where(exclusion_mask, 0, fullres_array)

    return cleaned_array
zarrnii.plugins.scaled_processing.segmentation_cleaner.SegmentationCleaner.scaled_processing_plugin_name()

Return the name of the algorithm.

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
153
154
155
156
@hookimpl
def scaled_processing_plugin_name(self) -> str:
    """Return the name of the algorithm."""
    return "Segmentation Cleaner"
zarrnii.plugins.scaled_processing.segmentation_cleaner.SegmentationCleaner.scaled_processing_plugin_description()

Return a description of the algorithm.

Source code in zarrnii/plugins/scaled_processing/segmentation_cleaner.py 
158
159
160
161
162
163
164
165
166
@hookimpl
def scaled_processing_plugin_description(self) -> str:
    """Return a description of the algorithm."""
    return (
        "Multi-resolution segmentation cleaning. Identifies and removes "
        "artifactual objects by performing connected components analysis at "
        "low resolution, filtering by extent, and applying exclusion mask to "
        "full resolution data."
    )