Skip to content

API Reference

Core

Core data models and tile operations. This module contains the fundamental building blocks: Tile (a single image tile with position, size, and loader), TiledImage (a collection of tiles forming one image), and functions for parsing tiles from DataFrames or building them programmatically.

Key exports: Tile, TiledImage, TileSlice, TileFOVGroup, hcs_images_from_dataframe, single_images_from_dataframe, tiled_image_from_tiles, build_dummy_tile.

ome_zarr_converters_tools.core

Core utility module for OME-Zarr converters tools.

Tile

Bases: BaseModel, Generic[CollectionInterfaceType, ImageLoaderInterfaceType]

A tile representing a region of an image to be converted.

This model is a complete definition of a tile, including its position, size, how to load the image data, and additional metadata. This model is the basic entry point for defining what regions of an acquisition to convert.

Attributes:

  • fov_name (str) –

    Name of the field of view (FOV) this tile belongs to.

  • start_x (float) –

    Starting position in the X dimension.

  • start_y (float) –

    Starting position in the Y dimension.

  • start_z (float) –

    Starting position in the Z dimension.

  • start_c (int) –

    Starting position in the C (channel) dimension. Channel indices index into acquisition_details.channels when channel metadata is provided, so start_c + length_c must not exceed its length.

  • start_t (float) –

    Starting position in the T (time) dimension.

  • length_x (float) –

    Length of the tile in the X dimension.

  • length_y (float) –

    Length of the tile in the Y dimension.

  • length_z (float) –

    Length of the tile in the Z dimension.

  • length_c (int) –

    Length of the tile in the C (channel) dimension.

  • length_t (float) –

    Length of the tile in the T (time) dimension.

  • collection (CollectionInterfaceType) –

    Collection model defining how to build the path to the image(s).

  • image_loader (ImageLoaderInterfaceType) –

    Image loader model defining how to load the image data.

  • acquisition_details (AcquisitionDetails) –

    Acquisition specific details that will be used to validate and convert the tile.

  • attributes (dict[str, AttributeType]) –

    Additional attributes for the these will be passed to the fractal image list as key-value pairs.

Source code in ome_zarr_converters_tools/core/_tile.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
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class Tile(BaseModel, Generic[CollectionInterfaceType, ImageLoaderInterfaceType]):
    """A tile representing a region of an image to be converted.

    This model is a complete definition of a tile, including its position,
    size, how to load the image data, and additional metadata. This model is the
    basic entry point for defining what regions of an acquisition to convert.

    Attributes:
        fov_name: Name of the field of view (FOV) this tile belongs to.
        start_x: Starting position in the X dimension.
        start_y: Starting position in the Y dimension.
        start_z: Starting position in the Z dimension.
        start_c: Starting position in the C (channel) dimension. Channel indices
            index into `acquisition_details.channels` when channel metadata is
            provided, so `start_c + length_c` must not exceed its length.
        start_t: Starting position in the T (time) dimension.
        length_x: Length of the tile in the X dimension.
        length_y: Length of the tile in the Y dimension.
        length_z: Length of the tile in the Z dimension.
        length_c: Length of the tile in the C (channel) dimension.
        length_t: Length of the tile in the T (time) dimension.
        collection: Collection model defining how to build the path to the image(s).
        image_loader: Image loader model defining how to load the image data.
        acquisition_details: Acquisition specific details that will be used to validate
            and convert the tile.
        attributes: Additional attributes for the these will be passed to
            the fractal image list as key-value pairs.

    """

    fov_name: str
    # Positions
    start_x: float
    start_y: float
    start_z: float = 0.0
    start_c: int = 0
    start_t: float = 0.0

    # Sizes
    length_x: float = Field(gt=0)
    length_y: float = Field(gt=0)
    length_z: float = Field(default=1.0, gt=0)
    length_c: int = Field(default=1, gt=0)
    length_t: float = Field(default=1.0, gt=0)

    # Additional attribute for the tile
    attributes: dict[str, AttributeType] = Field(default_factory=dict)
    # Collection model defining how to build the path to the image(s)
    collection: CollectionInterfaceType
    # Image loader model defining how to load the image data
    # This model will need to wrap all the necessary context
    # to load the image data for this tile
    image_loader: ImageLoaderInterfaceType
    # Acquisition specific details that will be used to validate and convert
    # the tile
    acquisition_details: AcquisitionDetails

    # Pydantic configuration
    model_config = ConfigDict(extra="forbid")

    @model_validator(mode="after")
    def _validate_channel_range(self) -> "Tile":
        """Enforce that channel indices resolve into `acquisition_details.channels`."""
        if self.start_c < 0:
            raise ValueError(
                f"Tile '{self.fov_name}' has start_c={self.start_c}; "
                "channel indices must be >= 0."
            )
        channels = self.acquisition_details.channels
        if channels is not None and self.start_c + self.length_c > len(channels):
            max_index = self.start_c + self.length_c - 1
            raise ValueError(
                f"Tile '{self.fov_name}' references channel index {max_index} but "
                f"acquisition_details.channels has {len(channels)} entries. Provide "
                "one ChannelInfo per channel index (padding unused instrument slots "
                'with e.g. ChannelInfo(channel_label="unused_1")), or set '
                "channels=None to use auto-generated channel names."
            )
        return self

    def to_roi(self) -> Roi:
        """Convert the Tile to a Roi."""
        acquisition_details = self.acquisition_details
        stage_corrections = acquisition_details.stage_orientation
        spacing = {
            "x": acquisition_details.xy_pixel_size,
            "y": acquisition_details.xy_pixel_size,
            "z": acquisition_details.z_spacing,
            "t": acquisition_details.t_spacing,
        }
        origins = {}
        roi_slices = {}
        for ax in acquisition_details.axes:
            # `swap_xy` transposes the X and Y stage axes: the output x slice is
            # built from this tile's y position and vice versa. The output axis
            # label (`axis_name`) stays `ax`; only the source field is swapped.
            source_ax = ax
            if stage_corrections.swap_xy and ax == "x":
                source_ax = "y"
            elif stage_corrections.swap_xy and ax == "y":
                source_ax = "x"

            start_field = f"start_{source_ax}"
            start = getattr(self, start_field)
            start_space = getattr(acquisition_details, f"{start_field}_space", None)
            if start_space is not None:
                start = safe_to_world(
                    start=start,
                    spacing=spacing[source_ax],
                    space=start_space,
                )

            if ax == "x" and stage_corrections.flip_x:
                start = -start
            if ax == "y" and stage_corrections.flip_y:
                start = -start

            length_field = f"length_{source_ax}"
            length = getattr(self, length_field)
            length_space = getattr(acquisition_details, f"{length_field}_space", None)
            if length_space is not None:
                length = safe_to_world(
                    start=length,
                    spacing=spacing[source_ax],
                    space=length_space,
                )
            roi_slices[ax] = RoiSlice(start=start, length=length, axis_name=ax)
            if ax in ["x", "y", "z"]:
                origins[f"{ax}_micrometer_original"] = start

        return Roi(
            name=self.fov_name,
            slices=list(roi_slices.values()),
            space="world",
            **origins,
        )

    def find_data_type(self, resource: Any | None = None) -> str:
        """Find the data type of the image data."""
        if self.acquisition_details.data_type != DataTypeEnum.AUTODETECT:
            return self.acquisition_details.data_type
        return self.image_loader.find_data_type(resource)
find_data_type(resource: Any | None = None) -> str

Find the data type of the image data.

Source code in ome_zarr_converters_tools/core/_tile.py 
176
177
178
179
180
def find_data_type(self, resource: Any | None = None) -> str:
    """Find the data type of the image data."""
    if self.acquisition_details.data_type != DataTypeEnum.AUTODETECT:
        return self.acquisition_details.data_type
    return self.image_loader.find_data_type(resource)
to_roi() -> Roi

Convert the Tile to a Roi.

Source code in ome_zarr_converters_tools/core/_tile.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def to_roi(self) -> Roi:
    """Convert the Tile to a Roi."""
    acquisition_details = self.acquisition_details
    stage_corrections = acquisition_details.stage_orientation
    spacing = {
        "x": acquisition_details.xy_pixel_size,
        "y": acquisition_details.xy_pixel_size,
        "z": acquisition_details.z_spacing,
        "t": acquisition_details.t_spacing,
    }
    origins = {}
    roi_slices = {}
    for ax in acquisition_details.axes:
        # `swap_xy` transposes the X and Y stage axes: the output x slice is
        # built from this tile's y position and vice versa. The output axis
        # label (`axis_name`) stays `ax`; only the source field is swapped.
        source_ax = ax
        if stage_corrections.swap_xy and ax == "x":
            source_ax = "y"
        elif stage_corrections.swap_xy and ax == "y":
            source_ax = "x"

        start_field = f"start_{source_ax}"
        start = getattr(self, start_field)
        start_space = getattr(acquisition_details, f"{start_field}_space", None)
        if start_space is not None:
            start = safe_to_world(
                start=start,
                spacing=spacing[source_ax],
                space=start_space,
            )

        if ax == "x" and stage_corrections.flip_x:
            start = -start
        if ax == "y" and stage_corrections.flip_y:
            start = -start

        length_field = f"length_{source_ax}"
        length = getattr(self, length_field)
        length_space = getattr(acquisition_details, f"{length_field}_space", None)
        if length_space is not None:
            length = safe_to_world(
                start=length,
                spacing=spacing[source_ax],
                space=length_space,
            )
        roi_slices[ax] = RoiSlice(start=start, length=length, axis_name=ax)
        if ax in ["x", "y", "z"]:
            origins[f"{ax}_micrometer_original"] = start

    return Roi(
        name=self.fov_name,
        slices=list(roi_slices.values()),
        space="world",
        **origins,
    )

TileFOVGroup

Bases: BaseModel, Generic[ImageLoaderInterfaceType]

Group of TileSlices belonging to the same acquisition FOV.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
class TileFOVGroup(BaseModel, Generic[ImageLoaderInterfaceType]):
    """Group of TileSlices belonging to the same acquisition FOV."""

    fov_name: str
    regions: list[TileSlice[ImageLoaderInterfaceType]] = Field(default_factory=list)
    axes: list[CANONICAL_AXES_TYPE]
    pixel_size: PixelSize
    data_type: str

    model_config = ConfigDict(extra="forbid")

    def shape(self) -> tuple[int, ...]:
        """Get the shape of the FOV group by computing the union of all regions."""
        return shape_from_rois(
            [region.roi for region in self.regions],
            self.axes,
            self.pixel_size,
        )

    def roi(self) -> Roi:
        """Get the global ROI covering all TileSlices in the FOV group."""
        union_roi = bulk_roi_union([region.roi for region in self.regions])
        union_roi.name = self.fov_name
        return union_roi

    def ref_slice(self) -> TileSlice[ImageLoaderInterfaceType]:
        """Get a reference TileSlice for this FOV group."""
        point = {}
        for axis in self.axes:
            point[axis] = 0.0

        ref_region = self.regions[0]
        ref_distance = roi_to_point_distance(ref_region.roi, point)  # type: ignore
        for region in self.regions[1:]:
            distance = roi_to_point_distance(region.roi, point)  # type: ignore
            if distance < ref_distance:
                ref_region = region
                ref_distance = distance
        return ref_region

    def _prepare_slice_loading(
        self, resource: Any | None = None
    ) -> list[tuple[tuple[slice, ...], Callable[[], np.ndarray]]]:
        """Prepare the TileSlices and their corresponding slicing tuples for loading."""
        slices = []
        group_roi = self.roi()
        # Find the offset between the group ROI and the origin ROI
        offset = {}
        for axis in self.axes:
            group_slice = group_roi.get(axis)
            assert group_slice is not None
            ref_slice_axis = self.ref_slice().roi.get(axis)
            assert ref_slice_axis is not None
            start = ref_slice_axis.start
            assert start is not None
            offset[axis] = -start

        def make_loader(
            region: TileSlice, resource: Any | None
        ) -> Callable[[], np.ndarray]:
            return lambda: region.load_data(axes=self.axes, resource=resource)

        for region in self.regions:
            roi_zeroed = move_roi_by(region.roi, offset)  # type: ignore
            roi_slice = roi_zeroed.to_slicing_dict(pixel_size=self.pixel_size)
            slicing = []
            for axis in self.axes:
                _slice = roi_slice[axis]
                slicing.append(slice(math.floor(_slice.start), math.ceil(_slice.stop)))
            slices.append((tuple(slicing), make_loader(region, resource)))
        return slices

    def load_data(self, resource: Any | None = None) -> np.ndarray:
        """Load the full image data for this FOV group."""
        shape = self.shape()
        full_image = np.zeros(shape, dtype=np.dtype(self.data_type))
        slices = self._prepare_slice_loading(resource=resource)
        for slicing, loader in slices:
            full_image[slicing] = loader()
        return full_image

    def load_data_dask(
        self, resource: Any | None = None, chunks: tuple[int, ...] | None = None
    ) -> da.Array:
        """Load the full image data for this FOV group using Dask."""
        shape = self.shape()
        dtype = self.data_type
        slices = self._prepare_slice_loading(resource=resource)
        if chunks is None:
            chunks = shape
        return lazy_array_from_regions(
            slices, shape=shape, chunks=chunks, dtype=dtype, fill_value=0.0
        )
load_data(resource: Any | None = None) -> np.ndarray

Load the full image data for this FOV group.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
140
141
142
143
144
145
146
147
def load_data(self, resource: Any | None = None) -> np.ndarray:
    """Load the full image data for this FOV group."""
    shape = self.shape()
    full_image = np.zeros(shape, dtype=np.dtype(self.data_type))
    slices = self._prepare_slice_loading(resource=resource)
    for slicing, loader in slices:
        full_image[slicing] = loader()
    return full_image
load_data_dask(resource: Any | None = None, chunks: tuple[int, ...] | None = None) -> da.Array

Load the full image data for this FOV group using Dask.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
149
150
151
152
153
154
155
156
157
158
159
160
def load_data_dask(
    self, resource: Any | None = None, chunks: tuple[int, ...] | None = None
) -> da.Array:
    """Load the full image data for this FOV group using Dask."""
    shape = self.shape()
    dtype = self.data_type
    slices = self._prepare_slice_loading(resource=resource)
    if chunks is None:
        chunks = shape
    return lazy_array_from_regions(
        slices, shape=shape, chunks=chunks, dtype=dtype, fill_value=0.0
    )
ref_slice() -> TileSlice[ImageLoaderInterfaceType]

Get a reference TileSlice for this FOV group.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def ref_slice(self) -> TileSlice[ImageLoaderInterfaceType]:
    """Get a reference TileSlice for this FOV group."""
    point = {}
    for axis in self.axes:
        point[axis] = 0.0

    ref_region = self.regions[0]
    ref_distance = roi_to_point_distance(ref_region.roi, point)  # type: ignore
    for region in self.regions[1:]:
        distance = roi_to_point_distance(region.roi, point)  # type: ignore
        if distance < ref_distance:
            ref_region = region
            ref_distance = distance
    return ref_region
roi() -> Roi

Get the global ROI covering all TileSlices in the FOV group.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
87
88
89
90
91
def roi(self) -> Roi:
    """Get the global ROI covering all TileSlices in the FOV group."""
    union_roi = bulk_roi_union([region.roi for region in self.regions])
    union_roi.name = self.fov_name
    return union_roi
shape() -> tuple[int, ...]

Get the shape of the FOV group by computing the union of all regions.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
79
80
81
82
83
84
85
def shape(self) -> tuple[int, ...]:
    """Get the shape of the FOV group by computing the union of all regions."""
    return shape_from_rois(
        [region.roi for region in self.regions],
        self.axes,
        self.pixel_size,
    )

TileSlice

Bases: BaseModel, Generic[ImageLoaderInterfaceType]

The smallest unit of a tiled image.

Usually corresponds to the minimal unit in which the source data can be loaded (e.g., a single tiff file from the microscope).

Source code in ome_zarr_converters_tools/core/_tile_region.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class TileSlice(BaseModel, Generic[ImageLoaderInterfaceType]):
    """The smallest unit of a tiled image.

    Usually corresponds to the minimal unit in which the source data
    can be loaded (e.g., a single tiff file from the microscope).

    """

    roi: Roi
    image_loader: ImageLoaderInterfaceType
    model_config = ConfigDict(extra="forbid")

    @classmethod
    def from_tile(cls, tile: Tile) -> Self:
        """Create a TileSlice from a Tile."""
        return cls(
            roi=tile.to_roi(),
            image_loader=tile.image_loader,
        )

    def load_data(
        self, *, axes: list[CANONICAL_AXES_TYPE], resource: Any | None = None
    ) -> np.ndarray:
        """Load the image data for this TileSlice using the image loader."""
        data = self.image_loader.load_data(resource=resource)
        # Padding data to match the ROI shape if necessary
        n_axes = len(axes)
        data_axes = data.ndim
        if data_axes > n_axes:
            raise ValueError("Data has more axes than expected.")
        if data_axes < n_axes:
            data = data.reshape((1,) * (n_axes - data_axes) + data.shape)
        return data
from_tile(tile: Tile) -> Self classmethod

Create a TileSlice from a Tile.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
45
46
47
48
49
50
51
@classmethod
def from_tile(cls, tile: Tile) -> Self:
    """Create a TileSlice from a Tile."""
    return cls(
        roi=tile.to_roi(),
        image_loader=tile.image_loader,
    )
load_data(*, axes: list[CANONICAL_AXES_TYPE], resource: Any | None = None) -> np.ndarray

Load the image data for this TileSlice using the image loader.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def load_data(
    self, *, axes: list[CANONICAL_AXES_TYPE], resource: Any | None = None
) -> np.ndarray:
    """Load the image data for this TileSlice using the image loader."""
    data = self.image_loader.load_data(resource=resource)
    # Padding data to match the ROI shape if necessary
    n_axes = len(axes)
    data_axes = data.ndim
    if data_axes > n_axes:
        raise ValueError("Data has more axes than expected.")
    if data_axes < n_axes:
        data = data.reshape((1,) * (n_axes - data_axes) + data.shape)
    return data

TiledImage

Bases: BaseModel, Generic[CollectionInterfaceType, ImageLoaderInterfaceType]

A TiledImage is the unit that will be converted into an OME-Zarr image.

Can contain multiple TileFOVGroups, each containing multiple TileSlices or it can directly contain a single TileFOVGroup.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
class TiledImage(BaseModel, Generic[CollectionInterfaceType, ImageLoaderInterfaceType]):
    """A TiledImage is the unit that will be converted into an OME-Zarr image.

    Can contain multiple TileFOVGroups, each containing multiple TileSlices
    or it can directly contain a single TileFOVGroup.
    """

    regions: list[TileSlice[ImageLoaderInterfaceType]] = Field(default_factory=list)
    path: str
    name: str | None = None
    xy_pixel_size: float = 1.0
    z_spacing: float = 1.0
    t_spacing: float = 1.0
    data_type: str
    axes: list[CANONICAL_AXES_TYPE]
    collection: CollectionInterfaceType
    channels: list[ChannelInfo] | None = None
    translation: list[float | int] | None = None
    attributes: dict[str, AttributeType] = Field(default_factory=dict)

    model_config = ConfigDict(extra="forbid")

    @model_validator(mode="after")
    def _validate_channel_coverage(self) -> "TiledImage":
        """Enforce that every region's channel range fits the channel metadata.

        Tiles are validated at construction, but a TiledImage is also rebuilt
        from JSON at the fractal init/compute task boundary; this guards that
        path with the same contract.
        """
        if self.channels is None:
            return self
        num_channels = len(self.channels)
        for region in self.regions:
            c_slice = region.roi.get("c")
            if c_slice is None or c_slice.start is None:
                continue
            length = c_slice.length if c_slice.length is not None else 1
            if c_slice.start < 0 or c_slice.start + length > num_channels:
                raise ValueError(
                    f"TiledImage '{self.path}' region '{region.roi.name}' "
                    f"references channel range [{c_slice.start}, "
                    f"{c_slice.start + length}) but channels has {num_channels} "
                    "entries. Provide one ChannelInfo per channel index, or set "
                    "channels=None to use auto-generated channel names."
                )
        return self

    def group_by_fov(self) -> list[TileFOVGroup[ImageLoaderInterfaceType]]:
        """Group TileSlices by field of view name."""
        fov_dict: dict[str, list[TileSlice]] = {}
        for region in self.regions:
            fov_name = region.roi.name
            if fov_name is None:
                raise ValueError("TileSlice ROI must have a name to group by FOV.")
            if fov_name not in fov_dict:
                fov_dict[fov_name] = []
            fov_dict[fov_name].append(region)
        return [
            TileFOVGroup(
                fov_name=fov_name,
                regions=regions,
                axes=self.axes,
                pixel_size=self.pixel_size,
                data_type=self.data_type,
            )
            for fov_name, regions in fov_dict.items()
        ]

    @property
    def pixel_size(self) -> PixelSize:
        """Return the PixelSize of the TiledImage."""
        return PixelSize(
            x=self.xy_pixel_size,
            y=self.xy_pixel_size,
            z=self.z_spacing,
            t=self.t_spacing,
        )

    def add_tile(self, tile: Tile, add_translation: bool = False) -> None:
        """Add a Tile to the TiledImage as a TileSlice."""
        if self.channels != tile.acquisition_details.channels:
            raise ValueError("Tile channels do not match TiledImage channels.")
        if self.axes != tile.acquisition_details.axes:
            raise ValueError("Tile axes do not match TiledImage axes.")
        if self.xy_pixel_size != tile.acquisition_details.xy_pixel_size:
            raise ValueError(
                "Tile xy_pixel_size does not match TiledImage xy_pixel_size."
            )
        if self.z_spacing != tile.acquisition_details.z_spacing:
            raise ValueError("Tile z_spacing does not match TiledImage z_spacing.")
        if self.t_spacing != tile.acquisition_details.t_spacing:
            raise ValueError("Tile t_spacing does not match TiledImage t_spacing.")
        tile_slice = TileSlice.from_tile(tile)

        if add_translation:
            roi_extra = tile_slice.roi.model_extra or {}
            translation = []
            for ax in self.axes:
                o_ax = roi_extra.get(f"{ax}_micrometer_original")
                translation.append(o_ax if o_ax is not None else 0.0)

            if self.translation is None:
                self.translation = translation
            else:
                self.translation = [
                    min(t, tr)
                    for t, tr in zip(translation, self.translation, strict=True)
                ]
        self.regions.append(tile_slice)

    def shape(self) -> tuple[int, ...]:
        """Get the shape of the TiledImage by computing the union of all regions."""
        return shape_from_rois(
            [region.roi for region in self.regions],
            self.axes,
            self.pixel_size,
        )

    def output_shape(self) -> tuple[int, ...]:
        """Output-array shape anchored at pixel 0 (includes any left-padding).

        Equals `shape()` when the mosaic origin is 0 (the default); larger when a
        `remove_*_offset="Keep"` axis keeps a positive absolute origin.
        """
        return output_shape_from_rois(
            [region.roi for region in self.regions],
            self.axes,
            self.pixel_size,
        )

    def roi(self) -> Roi:
        """Get the global ROI covering all TileSlices in the TiledImage."""
        union_roi = bulk_roi_union([region.roi for region in self.regions])
        union_roi.name = self.name or self.path
        return union_roi

    def _prepare_slice_loading(
        self, resource: Any | None = None
    ) -> list[tuple[tuple[slice, ...], Callable[[], np.ndarray]]]:
        """Prepare the TileSlices and their corresponding slicing tuples for loading."""
        # Zero every region to the union origin so absolute stage coordinates map
        # into the union-bounding-box buffer allocated by `load_data`/`shape()`.
        # Without this, regions that do not start at pixel 0 index past the buffer
        # and their data is silently dropped (or raises a broadcast error).
        union_roi = self.roi()
        offset = {}
        for axis in self.axes:
            union_axis = union_roi.get(axis)
            assert union_axis is not None
            start = union_axis.start
            assert start is not None
            offset[axis] = -start

        def make_loader(
            region: TileSlice, resource: Any | None
        ) -> Callable[[], np.ndarray]:
            return lambda: region.load_data(axes=self.axes, resource=resource)

        slices = []
        for region in self.regions:
            roi_zeroed = move_roi_by(region.roi, offset)  # type: ignore
            roi_slice = roi_zeroed.to_slicing_dict(pixel_size=self.pixel_size)
            slicing = []
            for axis in self.axes:
                _slice = roi_slice[axis]
                slicing.append(slice(math.floor(_slice.start), math.ceil(_slice.stop)))
            slicing = tuple(slicing)
            slices.append((slicing, make_loader(region, resource)))
        return slices

    def load_data(self, resource: Any | None = None) -> np.ndarray:
        """Load the full image data for this TiledImage using the image loaders."""
        shape = self.shape()
        dtype = np.dtype(self.data_type)
        full_image = np.zeros(shape, dtype=dtype)
        slices = self._prepare_slice_loading(resource=resource)
        for slicing, loader in slices:
            full_image[slicing] = loader()
        return full_image

    def load_data_dask(
        self, resource: Any | None = None, chunks: tuple[int, ...] | None = None
    ) -> da.Array:
        """Load the full image data for this TiledImage using Dask."""
        shape = self.shape()
        dtype = self.data_type
        slices = self._prepare_slice_loading(resource=resource)
        if chunks is None:
            chunks = shape
        return lazy_array_from_regions(
            slices, shape=shape, chunks=chunks, dtype=dtype, fill_value=0.0
        )
pixel_size: PixelSize property

Return the PixelSize of the TiledImage.

add_tile(tile: Tile, add_translation: bool = False) -> None

Add a Tile to the TiledImage as a TileSlice.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def add_tile(self, tile: Tile, add_translation: bool = False) -> None:
    """Add a Tile to the TiledImage as a TileSlice."""
    if self.channels != tile.acquisition_details.channels:
        raise ValueError("Tile channels do not match TiledImage channels.")
    if self.axes != tile.acquisition_details.axes:
        raise ValueError("Tile axes do not match TiledImage axes.")
    if self.xy_pixel_size != tile.acquisition_details.xy_pixel_size:
        raise ValueError(
            "Tile xy_pixel_size does not match TiledImage xy_pixel_size."
        )
    if self.z_spacing != tile.acquisition_details.z_spacing:
        raise ValueError("Tile z_spacing does not match TiledImage z_spacing.")
    if self.t_spacing != tile.acquisition_details.t_spacing:
        raise ValueError("Tile t_spacing does not match TiledImage t_spacing.")
    tile_slice = TileSlice.from_tile(tile)

    if add_translation:
        roi_extra = tile_slice.roi.model_extra or {}
        translation = []
        for ax in self.axes:
            o_ax = roi_extra.get(f"{ax}_micrometer_original")
            translation.append(o_ax if o_ax is not None else 0.0)

        if self.translation is None:
            self.translation = translation
        else:
            self.translation = [
                min(t, tr)
                for t, tr in zip(translation, self.translation, strict=True)
            ]
    self.regions.append(tile_slice)
group_by_fov() -> list[TileFOVGroup[ImageLoaderInterfaceType]]

Group TileSlices by field of view name.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def group_by_fov(self) -> list[TileFOVGroup[ImageLoaderInterfaceType]]:
    """Group TileSlices by field of view name."""
    fov_dict: dict[str, list[TileSlice]] = {}
    for region in self.regions:
        fov_name = region.roi.name
        if fov_name is None:
            raise ValueError("TileSlice ROI must have a name to group by FOV.")
        if fov_name not in fov_dict:
            fov_dict[fov_name] = []
        fov_dict[fov_name].append(region)
    return [
        TileFOVGroup(
            fov_name=fov_name,
            regions=regions,
            axes=self.axes,
            pixel_size=self.pixel_size,
            data_type=self.data_type,
        )
        for fov_name, regions in fov_dict.items()
    ]
load_data(resource: Any | None = None) -> np.ndarray

Load the full image data for this TiledImage using the image loaders.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
334
335
336
337
338
339
340
341
342
def load_data(self, resource: Any | None = None) -> np.ndarray:
    """Load the full image data for this TiledImage using the image loaders."""
    shape = self.shape()
    dtype = np.dtype(self.data_type)
    full_image = np.zeros(shape, dtype=dtype)
    slices = self._prepare_slice_loading(resource=resource)
    for slicing, loader in slices:
        full_image[slicing] = loader()
    return full_image
load_data_dask(resource: Any | None = None, chunks: tuple[int, ...] | None = None) -> da.Array

Load the full image data for this TiledImage using Dask.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
344
345
346
347
348
349
350
351
352
353
354
355
def load_data_dask(
    self, resource: Any | None = None, chunks: tuple[int, ...] | None = None
) -> da.Array:
    """Load the full image data for this TiledImage using Dask."""
    shape = self.shape()
    dtype = self.data_type
    slices = self._prepare_slice_loading(resource=resource)
    if chunks is None:
        chunks = shape
    return lazy_array_from_regions(
        slices, shape=shape, chunks=chunks, dtype=dtype, fill_value=0.0
    )
output_shape() -> tuple[int, ...]

Output-array shape anchored at pixel 0 (includes any left-padding).

Equals shape() when the mosaic origin is 0 (the default); larger when a remove_*_offset="Keep" axis keeps a positive absolute origin.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
282
283
284
285
286
287
288
289
290
291
292
def output_shape(self) -> tuple[int, ...]:
    """Output-array shape anchored at pixel 0 (includes any left-padding).

    Equals `shape()` when the mosaic origin is 0 (the default); larger when a
    `remove_*_offset="Keep"` axis keeps a positive absolute origin.
    """
    return output_shape_from_rois(
        [region.roi for region in self.regions],
        self.axes,
        self.pixel_size,
    )
roi() -> Roi

Get the global ROI covering all TileSlices in the TiledImage.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
294
295
296
297
298
def roi(self) -> Roi:
    """Get the global ROI covering all TileSlices in the TiledImage."""
    union_roi = bulk_roi_union([region.roi for region in self.regions])
    union_roi.name = self.name or self.path
    return union_roi
shape() -> tuple[int, ...]

Get the shape of the TiledImage by computing the union of all regions.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
274
275
276
277
278
279
280
def shape(self) -> tuple[int, ...]:
    """Get the shape of the TiledImage by computing the union of all regions."""
    return shape_from_rois(
        [region.roi for region in self.regions],
        self.axes,
        self.pixel_size,
    )

build_dummy_tile(*, fov_name: str, start: StartPosition, shape: TileShape, collection: CollectionInterface, acquisition_details: AcquisitionDetails, font_scale: float = 0.22) -> Tile

Build a dummy tile with default parameters, allowing overrides.

Source code in ome_zarr_converters_tools/core/_dummy_tiles.py 
 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
def build_dummy_tile(
    *,
    fov_name: str,
    start: StartPosition,
    shape: TileShape,
    collection: CollectionInterface,
    acquisition_details: AcquisitionDetails,
    font_scale: float = 0.22,
) -> Tile:
    """Build a dummy tile with default parameters, allowing overrides."""
    return Tile(
        fov_name=fov_name,
        start_x=start.x,
        start_y=start.y,
        start_z=start.z,
        start_c=start.c,
        start_t=start.t,
        length_x=shape.x,
        length_y=shape.y,
        length_z=shape.z,
        length_c=shape.c,
        length_t=shape.t,
        collection=collection,
        image_loader=DummyLoader(shape=shape, text=fov_name, font_scale=font_scale),
        acquisition_details=acquisition_details,
    )

find_url_type(url: str) -> UrlType

Classify a URL as local, S3, or unsupported (see UrlType).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
39
40
41
42
43
44
45
def find_url_type(url: str) -> UrlType:
    """Classify a URL as local, S3, or unsupported (see `UrlType`)."""
    if url.startswith("s3://"):
        return UrlType.S3
    elif _is_local_absolute(url):
        return UrlType.LOCAL
    return UrlType.NOT_SUPPORTED

hcs_images_from_dataframe(*, tiles_table: pd.DataFrame, acquisition_details: AcquisitionDetails, plate_name: str | None = None, acquisition_id: int = 0) -> list[Tile]

Build a list of Tiles belonging to an HCS plate acquisition.

Parameters:

  • tiles_table (DataFrame) –

    DataFrame containing the tiles table.

  • acquisition_details (AcquisitionDetails) –

    AcquisitionDetails model for the acquisition.

  • plate_name (str | None, default: None ) –

    Optional name of the plate.

  • acquisition_id (int, default: 0 ) –

    Acquisition index.

Returns:

  • list[Tile] –

    One Tile per row of the tiles table, with ImageInPlate collections.

Source code in ome_zarr_converters_tools/core/_table.py 
 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
def hcs_images_from_dataframe(
    *,
    tiles_table: pd.DataFrame,
    acquisition_details: AcquisitionDetails,
    plate_name: str | None = None,
    acquisition_id: int = 0,
) -> list[Tile]:
    """Build a list of Tiles belonging to an HCS plate acquisition.

    Args:
        tiles_table: DataFrame containing the tiles table.
        acquisition_details: AcquisitionDetails model for the acquisition.
        plate_name: Optional name of the plate.
        acquisition_id: Acquisition index.

    Returns:
        One Tile per row of the tiles table, with `ImageInPlate` collections.
    """
    plate_name = plate_name or "Plate"
    tiles = []
    for _, row in tiles_table.iterrows():
        row_dict = row.to_dict()
        image_loader, row_dict = _build_default_image_loader(data=row_dict)
        collection, row_dict = _build_plate_collection(
            data=row_dict,
            plate_name=plate_name,
            acquisition=acquisition_id,
        )
        tile_data, attributes_data = build_tile_data_and_attributes_from_row(
            data=row_dict
        )
        tile = Tile(
            **tile_data,
            image_loader=image_loader,
            collection=collection,
            acquisition_details=acquisition_details,
            attributes=attributes_data,
        )
        tiles.append(tile)
    return tiles

join_url_paths(base_url: str, *paths: str) -> str

Join path components to a base URL, normalizing ./.. segments.

Used instead of os.path.join or pathlib.Path to support both local and S3 URLs. Resolves ./.. and collapses redundant slashes while staying protocol- and Windows-safe: it uses posixpath.normpath (never os.path.normpath, which on Windows would rewrite forward slashes to backslashes and corrupt S3 keys).

Raises:

  • ValueError –

    if .. segments would ascend above the network location of a protocol URL (e.g. join_url_paths("s3://bucket", "..", "x") would otherwise silently drop the bucket).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
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
def join_url_paths(base_url: str, *paths: str) -> str:
    """Join path components to a base URL, normalizing `.`/`..` segments.

    Used instead of `os.path.join` or `pathlib.Path` to support both local
    and S3 URLs. Resolves `.`/`..` and collapses redundant slashes while
    staying protocol- and Windows-safe: it uses `posixpath.normpath` (never
    `os.path.normpath`, which on Windows would rewrite forward slashes to
    backslashes and corrupt S3 keys).

    Raises:
        ValueError: if `..` segments would ascend above the network location
            of a protocol URL (e.g. `join_url_paths("s3://bucket", "..", "x")`
            would otherwise silently drop the bucket).
    """
    protocol, base = fsspec.core.split_protocol(base_url)
    combined = f"{base}/{'/'.join(str(p) for p in paths)}".replace("\\", "/")
    joined = posixpath.normpath(combined)
    if protocol is None:
        return joined
    # The first component after the protocol is the network location (e.g. the
    # S3 bucket) and is inviolable — reject any `..` that escapes it.
    netloc = base.replace("\\", "/").split("/", 1)[0]
    if joined != netloc and not joined.startswith(f"{netloc}/"):
        raise ValueError(
            f"Path escapes network location for URL: {base_url!r} + {paths!r}"
        )
    return f"{protocol}://{joined}"

local_url_to_path(url: str) -> Path

Convert a local URL to an absolute Path, expanding ~.

Does not touch the filesystem.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
48
49
50
51
52
53
def local_url_to_path(url: str) -> Path:
    """Convert a local URL to an absolute Path, expanding `~`.

    Does not touch the filesystem.
    """
    return Path(url).expanduser().resolve()

single_images_from_dataframe(*, tiles_table: pd.DataFrame, acquisition_details: AcquisitionDetails) -> list[Tile]

Build a list of Tiles belonging to stand-alone (non-plate) images.

Parameters:

  • tiles_table (DataFrame) –

    DataFrame containing the tiles table.

  • acquisition_details (AcquisitionDetails) –

    AcquisitionDetails model for the acquisition.

Returns:

  • list[Tile] –

    One Tile per row of the tiles table, with SingleImage collections.

Source code in ome_zarr_converters_tools/core/_table.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def single_images_from_dataframe(
    *,
    tiles_table: pd.DataFrame,
    acquisition_details: AcquisitionDetails,
) -> list[Tile]:
    """Build a list of Tiles belonging to stand-alone (non-plate) images.

    Args:
        tiles_table: DataFrame containing the tiles table.
        acquisition_details: AcquisitionDetails model for the acquisition.

    Returns:
        One Tile per row of the tiles table, with `SingleImage` collections.
    """
    tiles = []
    for _, row in tiles_table.iterrows():
        row_dict = row.to_dict()
        image_loader, row_dict = _build_default_image_loader(data=row_dict)
        collection, row_dict = _build_single_image_collection(
            data=row_dict,
        )
        tile_data, attributes_data = build_tile_data_and_attributes_from_row(
            data=row_dict
        )
        tile = Tile(
            **tile_data,
            image_loader=image_loader,
            collection=collection,
            acquisition_details=acquisition_details,
            attributes=attributes_data,
        )
        tiles.append(tile)
    return tiles

tiled_image_from_tiles(*, tiles: list[Tile], split_per_fov: bool = False, resource: Any | None = None) -> list[TiledImage]

Build a list of TiledImages from a list of Tiles.

Parameters:

  • tiles (list[Tile]) –

    List of Tile models to build the TiledImages from.

  • split_per_fov (bool, default: False ) –

    If True, each field of view becomes its own TiledImage; if False, all fields of view are aggregated into one mosaic image.

  • resource (Any | None, default: None ) –

    Optional resource to assist in processing.

Returns:

  • list[TiledImage] –

    A list of TiledImage models created from the tiles.

Source code in ome_zarr_converters_tools/core/_tile_to_tiled_images.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def tiled_image_from_tiles(
    *,
    tiles: list[Tile],
    split_per_fov: bool = False,
    resource: Any | None = None,
) -> list[TiledImage]:
    """Build a list of TiledImages from a list of Tiles.

    Args:
        tiles: List of Tile models to build the TiledImages from.
        split_per_fov: If True, each field of view becomes its own TiledImage;
            if False, all fields of view are aggregated into one mosaic image.
        resource: Optional resource to assist in processing.

    Returns:
        A list of TiledImage models created from the tiles.

    """
    tiled_images = {}

    if len(tiles) == 0:
        raise ValueError("No tiles provided to build TiledImage.")
    data_type = tiles[0].find_data_type(resource=resource)
    for tile in tiles:
        if split_per_fov:
            suffix = f"_{tile.fov_name}"
            add_translation = True
        else:
            suffix = ""
            add_translation = False
        tile.collection.set_suffix(suffix)
        path = tile.collection.path()
        if path not in tiled_images:
            acquisition_details = tile.acquisition_details
            tiled_images[path] = TiledImage(
                path=path,
                regions=[],
                data_type=data_type,
                channels=acquisition_details.channels,
                xy_pixel_size=acquisition_details.xy_pixel_size,
                z_spacing=acquisition_details.z_spacing,
                t_spacing=acquisition_details.t_spacing,
                axes=acquisition_details.axes,
                collection=tile.collection,
                attributes=tile.attributes,
            )
        tiled_images[path].add_tile(tile, add_translation=add_translation)

    return list(tiled_images.values())

Models

Configuration models, collection types, and image loaders. This module defines the Pydantic models used to configure the conversion pipeline (ConverterOptions, AcquisitionDetails), the collection types that determine output structure (ImageInPlate, SingleImage), and the image loader interface for custom formats.

Key exports: ConverterOptions, AcquisitionDetails, ChannelInfo, ImageInPlate, SingleImage, ImageLoaderInterface, DefaultImageLoader, Grouping, MosaicGrouping, PerFovGrouping, TilingStrategy, AutoTiling, SnapToGridTiling, SnapToCornersTiling, InplaceTiling, WriterMode, OverwriteMode, StagePositionCorrections, OmeZarrOptions.

ome_zarr_converters_tools.models

Models and types definitions for the ome_zarr_converters_tools.

AcquisitionDetails

Bases: BaseModel

Details about the acquisition.

These attributes are known and fixed prior to conversion. (Either parsed from metadata or manually serialized by the user beforehand.)

Source code in ome_zarr_converters_tools/models/_acquisition.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class AcquisitionDetails(BaseModel):
    """Details about the acquisition.

    These attributes are known and fixed prior to conversion.
    (Either parsed from metadata or manually serialized by the user beforehand.)
    """

    # Determine the coordinate space for start and length values
    start_x_space: SPACE_TYPE = "world"
    start_y_space: SPACE_TYPE = "world"
    start_z_space: SPACE_TYPE = "world"
    start_t_space: SPACE_TYPE = "world"
    length_x_space: SPACE_TYPE = "pixel"
    length_y_space: SPACE_TYPE = "pixel"
    length_z_space: SPACE_TYPE = "pixel"
    length_t_space: SPACE_TYPE = "pixel"
    # Spacing information
    xy_pixel_size: float = Field(default=1.0, gt=0.0)  # in micrometers
    z_spacing: float = Field(default=1.0, gt=0.0)  # in micrometers
    t_spacing: float = Field(default=1.0, gt=0.0)  # in micrometers

    # Channel information
    channels: list[ChannelInfo] | None = None

    # Axes order to be used for the data (should be a subset of canonical axes)
    axes: list[CANONICAL_AXES_TYPE] = Field(
        default_factory=lambda: canonical_axes.copy(), min_length=2, max_length=5
    )

    # Data type of the image data (if known)
    data_type: DataTypeEnum = DataTypeEnum.AUTODETECT

    # Condition table path (if applicable)
    condition_table_path: str | None = None

    # Stage orientation corrections
    stage_orientation: StageOrientation = Field(default_factory=StageOrientation)

    model_config = ConfigDict(extra="forbid")

    @field_validator("axes")
    @classmethod
    def validate_axes(cls, v: list[CANONICAL_AXES_TYPE]) -> list[CANONICAL_AXES_TYPE]:
        """Validate that axes are in canonical order."""
        for i in range(1, len(v)):
            if canonical_axes.index(v[i]) <= canonical_axes.index(v[i - 1]):
                raise ValueError("Axes must be in canonical order: t, c, z, y, x")
        return v
validate_axes(v: list[CANONICAL_AXES_TYPE]) -> list[CANONICAL_AXES_TYPE] classmethod

Validate that axes are in canonical order.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
200
201
202
203
204
205
206
207
@field_validator("axes")
@classmethod
def validate_axes(cls, v: list[CANONICAL_AXES_TYPE]) -> list[CANONICAL_AXES_TYPE]:
    """Validate that axes are in canonical order."""
    for i in range(1, len(v)):
        if canonical_axes.index(v[i]) <= canonical_axes.index(v[i - 1]):
            raise ValueError("Axes must be in canonical order: t, c, z, y, x")
    return v

AutoTiling

Bases: UserFacingModel

Automatically pick between Snap to Grid and Snap to Corners.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
28
29
30
31
32
33
34
35
36
37
38
class AutoTiling(UserFacingModel):
    """Automatically pick between `Snap to Grid` and `Snap to Corners`."""

    model_config = ConfigDict(title="Auto")

    mode: Literal["Auto"] = "Auto"
    """Automatically use `Snap to Grid` when the field of view positions
    align to a regular grid, `Snap to Corners` otherwise."""
    tolerance: float = Field(default=1, ge=0, title="Tiling Tolerance (in pixels)")
    """Maximum misalignment (in pixels) tolerated when checking whether the
    field of view positions fit a regular grid."""
mode: Literal['Auto'] = 'Auto' class-attribute instance-attribute

Automatically use Snap to Grid when the field of view positions align to a regular grid, Snap to Corners otherwise.

tolerance: float = Field(default=1, ge=0, title='Tiling Tolerance (in pixels)') class-attribute instance-attribute

Maximum misalignment (in pixels) tolerated when checking whether the field of view positions fit a regular grid.

BackendType

Bases: StrEnum

File format used to store tables (e.g. ROI tables) in the OME-Zarr.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
144
145
146
147
148
149
150
class BackendType(StrEnum):
    """File format used to store tables (e.g. ROI tables) in the OME-Zarr."""

    ANNDATA = "anndata"
    JSON = "json"
    CSV = "csv"
    PARQUET = "parquet"

ChannelInfo

Bases: UserFacingModel

Name, wavelength, and display color of a channel.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
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
class ChannelInfo(UserFacingModel):
    """Name, wavelength, and display color of a channel."""

    channel_label: str
    """Label of the channel."""
    wavelength_id: str | None = None
    """
    The wavelength ID of the channel.
    This field can be used in some tasks as alternative to channel_label,
    e.g. for multiplexed acquisitions it can be used for applying illumination
    correction based on wavelength ID instead of channel name.
    """
    color: str | None = Field(
        default=None, pattern=r"^#([A-Fa-f0-9]{6}|[A-Fa-f0-9]{3})$"
    )
    """
    The color associated with the channel in hex format,
    e.g. for visualization purposes.
    """

    @model_validator(mode="after")
    def validate_channel_info(self) -> "ChannelInfo":
        """Assign a default color if None is provided."""
        if self.color is not None:
            return self
        color = _color_from_wavelength_id(self.wavelength_id)
        if color is None:
            color = _color_from_channel_label(self.channel_label)
        self.color = color
        return self
channel_label: str instance-attribute

Label of the channel.

color: str | None = Field(default=None, pattern='^#([A-Fa-f0-9]{6}|[A-Fa-f0-9]{3})$') class-attribute instance-attribute

The color associated with the channel in hex format, e.g. for visualization purposes.

wavelength_id: str | None = None class-attribute instance-attribute

The wavelength ID of the channel. This field can be used in some tasks as alternative to channel_label, e.g. for multiplexed acquisitions it can be used for applying illumination correction based on wavelength ID instead of channel name.

validate_channel_info() -> ChannelInfo

Assign a default color if None is provided.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
133
134
135
136
137
138
139
140
141
142
@model_validator(mode="after")
def validate_channel_info(self) -> "ChannelInfo":
    """Assign a default color if None is provided."""
    if self.color is not None:
        return self
    color = _color_from_wavelength_id(self.wavelength_id)
    if color is None:
        color = _color_from_channel_label(self.channel_label)
    self.color = color
    return self

CollectionInterface

Bases: BaseModel, ABC

Base class defining how to build the output path(s) for a collection.

Subclasses implement path and inherit the tiling _suffix slot, which is set via set_suffix during per-FOV splitting (do not set it manually).

Source code in ome_zarr_converters_tools/models/_collection.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class CollectionInterface(BaseModel, ABC):
    """Base class defining how to build the output path(s) for a collection.

    Subclasses implement `path` and inherit the tiling `_suffix` slot, which is
    set via `set_suffix` during per-FOV splitting (do not set it manually).
    """

    model_config = ConfigDict(extra="ignore")

    # Auto-generated suffix used to disambiguate per-FOV output paths.
    _suffix: str = PrivateAttr("")

    @abstractmethod
    def path(self) -> str:
        """Return the output path for this collection."""
        ...

    def set_suffix(self, suffix: str) -> None:
        """Set the per-FOV path suffix (used when splitting tiles per FOV)."""
        self._suffix = suffix
path() -> str abstractmethod

Return the output path for this collection.

Source code in ome_zarr_converters_tools/models/_collection.py 
40
41
42
43
@abstractmethod
def path(self) -> str:
    """Return the output path for this collection."""
    ...
set_suffix(suffix: str) -> None

Set the per-FOV path suffix (used when splitting tiles per FOV).

Source code in ome_zarr_converters_tools/models/_collection.py 
45
46
47
def set_suffix(self, suffix: str) -> None:
    """Set the per-FOV path suffix (used when splitting tiles per FOV)."""
    self._suffix = suffix

ConverterOptions

Bases: UserFacingModel

Options for the OME-Zarr conversion process.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
class ConverterOptions(UserFacingModel):
    """Options for the OME-Zarr conversion process."""

    writer_mode: WriterMode = Field(default=WriterMode.BY_FOV, title="Writer Mode")
    """
    Mode for writing data during conversion.

    - `By Tile`: Write data one tile at a time. This consumes less memory, but may be
      slower.
    - `By Tile (Using Dask)`: Write tiles in parallel using Dask. This is
      usually faster than writing by tile sequentially, but may consume more
      memory.
    - `By FOV`: Write data one field of view at a time. Often the best compromise
      between speed and memory usage in most cases.
    - `By FOV (Using Dask)`: Write fields of view in parallel using Dask.
      This is usually faster than writing by FOV sequentially, but may consume
      more memory.
    - `In Memory`: Load all data into memory before writing.
    """
    grouping: Grouping = Field(default_factory=MosaicGrouping, title="Grouping")
    """
    How fields of view are grouped into output images.

    - `Mosaic`: Aggregate all fields of view of an acquisition into one OME-Zarr,
      arranged by the nested `tiling_strategy`.
    - `Per-FOV`: Write each field of view as its own OME-Zarr image (no mosaic,
      no tiling strategy).
    """
    stage_position_corrections: StagePositionCorrections = Field(
        default_factory=StagePositionCorrections,
        title="Stage Position Corrections",
    )
    """Stage position correction options."""
    omezarr_options: OmeZarrOptions = Field(
        default_factory=OmeZarrOptions, title="OME-Zarr Options"
    )
    """Options specific to OME-Zarr writing."""
    runtime_settings: RuntimeSettings = Field(
        default_factory=RuntimeSettings, title="Runtime Settings"
    )
    """Advanced performance settings: parallelism, storage codec, and
    temporary storage. The defaults are safe for most conversions."""
grouping: Grouping = Field(default_factory=MosaicGrouping, title='Grouping') class-attribute instance-attribute

How fields of view are grouped into output images.

  • Mosaic: Aggregate all fields of view of an acquisition into one OME-Zarr, arranged by the nested tiling_strategy.
  • Per-FOV: Write each field of view as its own OME-Zarr image (no mosaic, no tiling strategy).
omezarr_options: OmeZarrOptions = Field(default_factory=OmeZarrOptions, title='OME-Zarr Options') class-attribute instance-attribute

Options specific to OME-Zarr writing.

runtime_settings: RuntimeSettings = Field(default_factory=RuntimeSettings, title='Runtime Settings') class-attribute instance-attribute

Advanced performance settings: parallelism, storage codec, and temporary storage. The defaults are safe for most conversions.

stage_position_corrections: StagePositionCorrections = Field(default_factory=StagePositionCorrections, title='Stage Position Corrections') class-attribute instance-attribute

Stage position correction options.

writer_mode: WriterMode = Field(default=(WriterMode.BY_FOV), title='Writer Mode') class-attribute instance-attribute

Mode for writing data during conversion.

  • By Tile: Write data one tile at a time. This consumes less memory, but may be slower.
  • By Tile (Using Dask): Write tiles in parallel using Dask. This is usually faster than writing by tile sequentially, but may consume more memory.
  • By FOV: Write data one field of view at a time. Often the best compromise between speed and memory usage in most cases.
  • By FOV (Using Dask): Write fields of view in parallel using Dask. This is usually faster than writing by FOV sequentially, but may consume more memory.
  • In Memory: Load all data into memory before writing.

DataTypeEnum

Bases: StrEnum

Pixel data type of the output image; autodetect infers it.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
24
25
26
27
28
29
30
class DataTypeEnum(StrEnum):
    """Pixel data type of the output image; `autodetect` infers it."""

    AUTODETECT = "autodetect"
    UINT8 = "uint8"
    UINT16 = "uint16"
    UINT32 = "uint32"

DefaultImageLoader

Bases: ImageLoaderInterface

File-based image loader for common formats (TIFF, PNG/JPEG/BMP, NPY).

The file type is inferred from the extension; unrecognized extensions are attempted as TIFF with a warning.

Source code in ome_zarr_converters_tools/models/_loader.py 
 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
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
class DefaultImageLoader(ImageLoaderInterface):
    """File-based image loader for common formats (TIFF, PNG/JPEG/BMP, NPY).

    The file type is inferred from the extension; unrecognized extensions are
    attempted as TIFF with a warning.
    """

    file_path: str
    """Path to the image file. If relative, it is resolved against the
    `resource` passed to `load_data` (usually the acquisition base directory)."""

    def _resolve_path(self, resource: Any | None) -> str:
        """Resolve `file_path` against the optional `resource` base directory."""
        try:
            if resource is not None:
                # Ensure we can convert to str
                resource = str(resource)
        except Exception:
            raise ValueError(  # noqa: B904
                "DefaultImageLoader expects resource to be of type str, Path, or None."
            )
        if resource and isinstance(resource, str):
            return join_url_paths(resource, self.file_path)
        return self.file_path

    def preflight(self, resource: Any | None = None) -> None:
        """Warn if the source file does not exist, without reading it."""
        path = self._resolve_path(resource)
        try:
            fs = filesystem_for_url(path, error_msg_prefix="Preflight check")
            exists = fs.exists(path)
        except Exception as e:
            warnings.warn(
                f"Preflight check could not verify source file '{path}': {e}",
                stacklevel=2,
            )
            return
        if not exists:
            warnings.warn(
                f"Source file '{path}' does not exist. Check that the file "
                "was not moved or deleted, and that `file_path` (combined "
                "with the `resource` base directory, if any) points to it.",
                stacklevel=2,
            )

    def load_data(self, resource: Any | None = None) -> np.ndarray:
        """Load the image data as a NumPy array."""
        path = self._resolve_path(resource)

        suffix = basename_url(path).split(".")[-1].lower()
        if suffix in ["tiff", "tif", "tf2", "tf8", "btf"]:
            return self.load_tiff(path)
        elif suffix in ["png", "jpg", "jpeg", "bmp"]:
            return self.load_png(path)
        elif suffix == "npy":
            return self.load_npy(path)
        else:
            # Unknown extension: many files (e.g. custom/uncommon TIFF variants) are
            # still readable by tifffile, so warn and attempt a best-effort TIFF read.
            warnings.warn(
                f"DefaultImageLoader does not recognize file type {suffix!r}; "
                "attempting to load it as a TIFF file.",
                stacklevel=2,
            )
            try:
                return self.load_tiff(path)
            except Exception as e:
                raise ValueError(
                    f"DefaultImageLoader cannot handle file type {suffix!r}: "
                    "the TIFF fallback failed to read the file. Supported types are "
                    ".tiff, .tif, .tf2, .tf8, .btf, .png, .jpg, .jpeg, .bmp, .npy."
                ) from e

    def load_tiff(self, path: str) -> np.ndarray:
        fs = filesystem_for_url(path, error_msg_prefix="Loading image")
        with fs.open(path, "rb") as f:
            with tifffile.TiffFile(f) as tif:
                return tif.asarray()

    def load_png(self, path: str) -> np.ndarray:
        fs = filesystem_for_url(path, error_msg_prefix="Loading image")
        with fs.open(path, "rb") as f:
            return np.array(Image.open(f))

    def load_npy(self, path: str) -> np.ndarray:
        fs = filesystem_for_url(path, error_msg_prefix="Loading image")
        with fs.open(path, "rb") as f:
            return np.load(f)
file_path: str instance-attribute

Path to the image file. If relative, it is resolved against the resource passed to load_data (usually the acquisition base directory).

load_data(resource: Any | None = None) -> np.ndarray

Load the image data as a NumPy array.

Source code in ome_zarr_converters_tools/models/_loader.py 
 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
def load_data(self, resource: Any | None = None) -> np.ndarray:
    """Load the image data as a NumPy array."""
    path = self._resolve_path(resource)

    suffix = basename_url(path).split(".")[-1].lower()
    if suffix in ["tiff", "tif", "tf2", "tf8", "btf"]:
        return self.load_tiff(path)
    elif suffix in ["png", "jpg", "jpeg", "bmp"]:
        return self.load_png(path)
    elif suffix == "npy":
        return self.load_npy(path)
    else:
        # Unknown extension: many files (e.g. custom/uncommon TIFF variants) are
        # still readable by tifffile, so warn and attempt a best-effort TIFF read.
        warnings.warn(
            f"DefaultImageLoader does not recognize file type {suffix!r}; "
            "attempting to load it as a TIFF file.",
            stacklevel=2,
        )
        try:
            return self.load_tiff(path)
        except Exception as e:
            raise ValueError(
                f"DefaultImageLoader cannot handle file type {suffix!r}: "
                "the TIFF fallback failed to read the file. Supported types are "
                ".tiff, .tif, .tf2, .tf8, .btf, .png, .jpg, .jpeg, .bmp, .npy."
            ) from e
preflight(resource: Any | None = None) -> None

Warn if the source file does not exist, without reading it.

Source code in ome_zarr_converters_tools/models/_loader.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def preflight(self, resource: Any | None = None) -> None:
    """Warn if the source file does not exist, without reading it."""
    path = self._resolve_path(resource)
    try:
        fs = filesystem_for_url(path, error_msg_prefix="Preflight check")
        exists = fs.exists(path)
    except Exception as e:
        warnings.warn(
            f"Preflight check could not verify source file '{path}': {e}",
            stacklevel=2,
        )
        return
    if not exists:
        warnings.warn(
            f"Source file '{path}' does not exist. Check that the file "
            "was not moved or deleted, and that `file_path` (combined "
            "with the `resource` base directory, if any) points to it.",
            stacklevel=2,
        )

DefaultScheduler

Bases: UserFacingModel

Keep the environment's default parallelism settings unchanged.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
58
59
60
61
62
63
64
65
66
67
class DefaultScheduler(UserFacingModel):
    """Keep the environment's default parallelism settings unchanged."""

    mode: Literal["Default"] = "Default"
    """How the conversion work is parallelized."""

    model_config = ConfigDict(title="Default")

    def get_config(self) -> dict[str, object]:
        return {}
mode: Literal['Default'] = 'Default' class-attribute instance-attribute

How the conversion work is parallelized.

FixedSizeChunking

Bases: UserFacingModel

Store the image in chunks of a fixed size.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
class FixedSizeChunking(UserFacingModel):
    """Store the image in chunks of a fixed size."""

    model_config = ConfigDict(title="Fixed Size")

    mode: Literal["Fixed Size"] = "Fixed Size"
    """How the image is split into storage chunks."""
    xy_chunk: int = Field(default=4096, ge=1, title="Chunk Size for XY")
    """Chunk size for XY dimensions."""
    z_chunk: int = Field(default=10, ge=1, title="Chunk Size for Z")
    """Chunk size for Z dimension."""
    c_chunk: int = Field(default=1, ge=1, title="Chunk Size for C")
    """Chunk size for C dimension."""
    t_chunk: int = Field(default=1, ge=1, title="Chunk Size for T")
    """Chunk size for T dimension."""

    def get_xy_chunk(self, fov_xy_shape: int) -> int:
        return self.xy_chunk
c_chunk: int = Field(default=1, ge=1, title='Chunk Size for C') class-attribute instance-attribute

Chunk size for C dimension.

mode: Literal['Fixed Size'] = 'Fixed Size' class-attribute instance-attribute

How the image is split into storage chunks.

t_chunk: int = Field(default=1, ge=1, title='Chunk Size for T') class-attribute instance-attribute

Chunk size for T dimension.

xy_chunk: int = Field(default=4096, ge=1, title='Chunk Size for XY') class-attribute instance-attribute

Chunk size for XY dimensions.

z_chunk: int = Field(default=10, ge=1, title='Chunk Size for Z') class-attribute instance-attribute

Chunk size for Z dimension.

FovBasedChunking

Bases: UserFacingModel

Store the image in chunks sized like the fields of view.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
class FovBasedChunking(UserFacingModel):
    """Store the image in chunks sized like the fields of view."""

    model_config = ConfigDict(title="Same as FOV")

    mode: Literal["Same as FOV"] = "Same as FOV"
    """How the image is split into storage chunks."""
    xy_scaling: Scalings = Field(default=Scalings.ONE, title="XY Scaling Factor")
    """
    Scaling factor for the XY chunk size, relative to the field of view size.

    - `1`: chunks match the field of view size.
    - `0.5`: chunks are half the field of view (smaller chunks, more files).
    - `2`: chunks are double the field of view (larger chunks, fewer files).
    """
    z_chunk: int = Field(default=10, ge=1, title="Chunk Size for Z")
    """Chunk size for Z dimension."""
    c_chunk: int = Field(default=1, ge=1, title="Chunk Size for C")
    """Chunk size for C dimension."""
    t_chunk: int = Field(default=1, ge=1, title="Chunk Size for T")
    """Chunk size for T dimension."""

    def get_xy_chunk(self, fov_xy_shape: int) -> int:
        scaling_factor = self.xy_scaling.to_float()
        chunk_size = int(fov_xy_shape * scaling_factor)
        return max(1, chunk_size)
c_chunk: int = Field(default=1, ge=1, title='Chunk Size for C') class-attribute instance-attribute

Chunk size for C dimension.

mode: Literal['Same as FOV'] = 'Same as FOV' class-attribute instance-attribute

How the image is split into storage chunks.

t_chunk: int = Field(default=1, ge=1, title='Chunk Size for T') class-attribute instance-attribute

Chunk size for T dimension.

xy_scaling: Scalings = Field(default=(Scalings.ONE), title='XY Scaling Factor') class-attribute instance-attribute

Scaling factor for the XY chunk size, relative to the field of view size.

  • 1: chunks match the field of view size.
  • 0.5: chunks are half the field of view (smaller chunks, more files).
  • 2: chunks are double the field of view (larger chunks, fewer files).
z_chunk: int = Field(default=10, ge=1, title='Chunk Size for Z') class-attribute instance-attribute

Chunk size for Z dimension.

ImageInPlate

Bases: CollectionInterface

Collection for an image inside an HCS plate (plate/row/column layout).

Source code in ome_zarr_converters_tools/models/_collection.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class ImageInPlate(CollectionInterface):
    """Collection for an image inside an HCS plate (plate/row/column layout)."""

    plate_name: str
    """Name of the plate (`.zarr` is appended if missing)."""
    row: str
    """Well row label, e.g. `"A"`. Integer input is converted (1 → `"A"`)."""
    column: int = Field(ge=1)
    """Well column number, 1-based."""
    acquisition: int = Field(default=0, ge=0)
    """Acquisition index (used as the image path inside the well)."""

    @property
    def well(self) -> str:
        return f"{self.row}{self.column:02d}"

    def plate_path(self) -> str:
        return sanitize_path(self.plate_name)

    def well_path(self) -> str:
        return join_url_paths(self.row, f"{self.column:02d}")

    def path_in_well(self) -> str:
        return f"{self.acquisition}{self._suffix}"

    def image_in_well_path(self) -> str:
        return join_url_paths(self.well_path(), self.path_in_well())

    def path(self) -> str:
        return join_url_paths(self.plate_path(), self.well_path(), self.path_in_well())

    @field_validator("row", mode="before")
    @classmethod
    def row_to_str(cls, v: Any) -> Any:
        if isinstance(v, str):
            return v
        v = int(v)
        if v < 1 or v > len(ALPHABET):
            raise ValueError(
                f"Row index {v} out of range. Must be between 1 and {len(ALPHABET)}"
            )
        return ALPHABET[v - 1]
acquisition: int = Field(default=0, ge=0) class-attribute instance-attribute

Acquisition index (used as the image path inside the well).

column: int = Field(ge=1) class-attribute instance-attribute

Well column number, 1-based.

plate_name: str instance-attribute

Name of the plate (.zarr is appended if missing).

row: str instance-attribute

Well row label, e.g. "A". Integer input is converted (1 → "A").

ImageLoaderInterface

Bases: BaseModel, ABC

Base class for image loaders; subclass it to support a custom format.

Implement load_data to return the tile pixels as a NumPy array. The optional resource carries per-call context (e.g. a base directory or an open handle) and is threaded through the conversion pipeline unchanged.

Source code in ome_zarr_converters_tools/models/_loader.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class ImageLoaderInterface(BaseModel, ABC):
    """Base class for image loaders; subclass it to support a custom format.

    Implement `load_data` to return the tile pixels as a NumPy array. The
    optional `resource` carries per-call context (e.g. a base directory or an
    open handle) and is threaded through the conversion pipeline unchanged.
    """

    model_config = ConfigDict(extra="ignore")

    @abstractmethod
    def load_data(self, resource: Any | None = None) -> np.ndarray:
        """Load the image data as a NumPy array."""
        pass

    def preflight(self, resource: Any | None = None) -> None:
        """Cheaply verify the source data is reachable, without loading it.

        The default implementation is a no-op. Override it to emit a warning
        on missing or unreadable sources (e.g. a file-existence check) so
        that pre-flight validators can surface such problems at init time,
        before compute jobs are dispatched. Implementations should warn, not
        raise: only loading the data decides whether it is truly unreadable.
        """

    def find_data_type(self, resource: Any | None = None) -> str:
        """Find the data type of the image data."""
        return str(self.load_data(resource).dtype)
find_data_type(resource: Any | None = None) -> str

Find the data type of the image data.

Source code in ome_zarr_converters_tools/models/_loader.py 
44
45
46
def find_data_type(self, resource: Any | None = None) -> str:
    """Find the data type of the image data."""
    return str(self.load_data(resource).dtype)
load_data(resource: Any | None = None) -> np.ndarray abstractmethod

Load the image data as a NumPy array.

Source code in ome_zarr_converters_tools/models/_loader.py 
29
30
31
32
@abstractmethod
def load_data(self, resource: Any | None = None) -> np.ndarray:
    """Load the image data as a NumPy array."""
    pass
preflight(resource: Any | None = None) -> None

Cheaply verify the source data is reachable, without loading it.

The default implementation is a no-op. Override it to emit a warning on missing or unreadable sources (e.g. a file-existence check) so that pre-flight validators can surface such problems at init time, before compute jobs are dispatched. Implementations should warn, not raise: only loading the data decides whether it is truly unreadable.

Source code in ome_zarr_converters_tools/models/_loader.py 
34
35
36
37
38
39
40
41
42
def preflight(self, resource: Any | None = None) -> None:
    """Cheaply verify the source data is reachable, without loading it.

    The default implementation is a no-op. Override it to emit a warning
    on missing or unreadable sources (e.g. a file-existence check) so
    that pre-flight validators can surface such problems at init time,
    before compute jobs are dispatched. Implementations should warn, not
    raise: only loading the data decides whether it is truly unreadable.
    """

InplaceTiling

Bases: UserFacingModel

Keep every field of view at its original stage position.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
64
65
66
67
68
69
70
71
72
class InplaceTiling(UserFacingModel):
    """Keep every field of view at its original stage position."""

    model_config = ConfigDict(title="Inplace")

    mode: Literal["Inplace"] = "Inplace"
    """Keep every field of view at its original stage position. Imprecise
    stage positions **may produce artifacts**: where tiles overlap, the last
    written tile wins."""
mode: Literal['Inplace'] = 'Inplace' class-attribute instance-attribute

Keep every field of view at its original stage position. Imprecise stage positions may produce artifacts: where tiles overlap, the last written tile wins.

MosaicGrouping

Bases: UserFacingModel

Aggregate all fields of view of an acquisition into one mosaic OME-Zarr.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
 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
class MosaicGrouping(UserFacingModel):
    """Aggregate all fields of view of an acquisition into one mosaic OME-Zarr."""

    mode: Literal["Mosaic"] = "Mosaic"
    """How fields of view are grouped into output images."""
    tiling_strategy: TilingStrategy = Field(
        default_factory=AutoTiling, title="Tiling Strategy"
    )
    """
    How the fields of view are arranged within the mosaic.

    - `Auto`: Automatically determine if `Snap to Grid` is possible,
      otherwise use `Snap to Corners`. Accepts an optional tolerance
      (in pixels) for grid alignment.
    - `Snap to Grid`: Tile images to fit a regular grid. This is only
      possible if image positions align to a grid (potentially with
      overlap). Accepts an optional tolerance (in pixels).
    - `Snap to Corners`: Tile images to fit a grid defined by the corner
      positions.
    - `Inplace`: Write tiles in their original positions without tiling.
      This may lead to artifacts if microscope stage positions are not
      precise.
    """
    model_config = ConfigDict(title="Mosaic")

    @property
    def split_per_fov(self) -> bool:
        """Whether each field of view becomes its own OME-Zarr image."""
        return False

    def tiling_for_registration(self) -> TilingStrategy:
        """Within-image arrangement strategy for the registration pipeline."""
        return self.tiling_strategy
mode: Literal['Mosaic'] = 'Mosaic' class-attribute instance-attribute

How fields of view are grouped into output images.

split_per_fov: bool property

Whether each field of view becomes its own OME-Zarr image.

tiling_strategy: TilingStrategy = Field(default_factory=AutoTiling, title='Tiling Strategy') class-attribute instance-attribute

How the fields of view are arranged within the mosaic.

  • Auto: Automatically determine if Snap to Grid is possible, otherwise use Snap to Corners. Accepts an optional tolerance (in pixels) for grid alignment.
  • Snap to Grid: Tile images to fit a regular grid. This is only possible if image positions align to a grid (potentially with overlap). Accepts an optional tolerance (in pixels).
  • Snap to Corners: Tile images to fit a grid defined by the corner positions.
  • Inplace: Write tiles in their original positions without tiling. This may lead to artifacts if microscope stage positions are not precise.
tiling_for_registration() -> TilingStrategy

Within-image arrangement strategy for the registration pipeline.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
111
112
113
def tiling_for_registration(self) -> TilingStrategy:
    """Within-image arrangement strategy for the registration pipeline."""
    return self.tiling_strategy

NamedLevels

Bases: UserFacingModel

Name each resolution level explicitly, from highest to lowest resolution.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
class NamedLevels(UserFacingModel):
    """Name each resolution level explicitly, from highest to lowest resolution."""

    model_config = ConfigDict(title="Custom Names")

    mode: Literal["Custom Names"] = "Custom Names"
    """How the resolution levels of the pyramid are defined."""
    level_names: list[str] = Field(min_length=1, title="Level Names")
    """Names of the resolution levels, from highest to lowest resolution,
    e.g. `["s0", "s1", "s2"]`. Each name becomes a path inside the OME-Zarr."""

    @field_validator("level_names")
    @classmethod
    def _validate_level_names(cls, level_names: list[str]) -> list[str]:
        for name in level_names:
            if not name or "/" in name or name != name.strip():
                raise ValueError(
                    f"Invalid level name {name!r}: each level name must be a "
                    "non-empty path segment without '/' or leading/trailing "
                    "whitespace, e.g. ['s0', 's1', 's2']."
                )
        duplicates = sorted({n for n in level_names if level_names.count(n) > 1})
        if duplicates:
            raise ValueError(
                f"Duplicate level names {duplicates}: each resolution level "
                "needs a unique name, e.g. ['s0', 's1', 's2']."
            )
        return level_names

    def to_ngio_levels(self) -> int | list[str]:
        """Levels in the form ngio's `create_empty_ome_zarr` accepts."""
        return self.level_names
level_names: list[str] = Field(min_length=1, title='Level Names') class-attribute instance-attribute

Names of the resolution levels, from highest to lowest resolution, e.g. ["s0", "s1", "s2"]. Each name becomes a path inside the OME-Zarr.

mode: Literal['Custom Names'] = 'Custom Names' class-attribute instance-attribute

How the resolution levels of the pyramid are defined.

to_ngio_levels() -> int | list[str]

Levels in the form ngio's create_empty_ome_zarr accepts.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
319
320
321
def to_ngio_levels(self) -> int | list[str]:
    """Levels in the form ngio's `create_empty_ome_zarr` accepts."""
    return self.level_names

NumberOfLevels

Bases: UserFacingModel

Create a number of resolution levels with default names (0, 1, ...).

Source code in ome_zarr_converters_tools/models/_converter_options.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
class NumberOfLevels(UserFacingModel):
    """Create a number of resolution levels with default names (`0`, `1`, ...)."""

    model_config = ConfigDict(title="Number of Levels")

    mode: Literal["Number of Levels"] = "Number of Levels"
    """How the resolution levels of the pyramid are defined."""
    num_levels: int = Field(default=5, ge=1, title="Number of Resolution Levels")
    """Number of resolution levels in the pyramid of the output image."""

    def to_ngio_levels(self) -> int | list[str]:
        """Levels in the form ngio's `create_empty_ome_zarr` accepts."""
        return self.num_levels
mode: Literal['Number of Levels'] = 'Number of Levels' class-attribute instance-attribute

How the resolution levels of the pyramid are defined.

num_levels: int = Field(default=5, ge=1, title='Number of Resolution Levels') class-attribute instance-attribute

Number of resolution levels in the pyramid of the output image.

to_ngio_levels() -> int | list[str]

Levels in the form ngio's create_empty_ome_zarr accepts.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
285
286
287
def to_ngio_levels(self) -> int | list[str]:
    """Levels in the form ngio's `create_empty_ome_zarr` accepts."""
    return self.num_levels

OmeZarrOptions

Bases: UserFacingModel

Options controlling the layout of the output OME-Zarr.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
class OmeZarrOptions(UserFacingModel):
    """Options controlling the layout of the output OME-Zarr."""

    levels: PyramidLevels = Field(
        default_factory=NumberOfLevels, title="Resolution Levels"
    )
    """
    How many resolution levels the output pyramid has and how they are named.

    - `Number of Levels`: a number of levels with the default names
      (`0`, `1`, ...).
    - `Custom Names`: an explicit list of level names, e.g. `["s0", "s1"]`.
    """
    chunks: ChunkingStrategy = Field(
        default_factory=FovBasedChunking, title="Chunking Strategy"
    )
    """How the image is split into storage chunks: sized like the fields of
    view (`Same as FOV`) or with fixed sizes (`Fixed Size`)."""
    ngff_version: NgffVersions = Field(default=DefaultNgffVersion, title="NGFF Version")
    """Version of the OME-NGFF specification to target."""
    table_backend: BackendType = Field(default=BackendType.CSV, title="Table Backend")
    """File format used to store tables (e.g. ROI tables) inside the
    OME-Zarr."""
chunks: ChunkingStrategy = Field(default_factory=FovBasedChunking, title='Chunking Strategy') class-attribute instance-attribute

How the image is split into storage chunks: sized like the fields of view (Same as FOV) or with fixed sizes (Fixed Size).

levels: PyramidLevels = Field(default_factory=NumberOfLevels, title='Resolution Levels') class-attribute instance-attribute

How many resolution levels the output pyramid has and how they are named.

  • Number of Levels: a number of levels with the default names (0, 1, ...).
  • Custom Names: an explicit list of level names, e.g. ["s0", "s1"].
ngff_version: NgffVersions = Field(default=DefaultNgffVersion, title='NGFF Version') class-attribute instance-attribute

Version of the OME-NGFF specification to target.

table_backend: BackendType = Field(default=(BackendType.CSV), title='Table Backend') class-attribute instance-attribute

File format used to store tables (e.g. ROI tables) inside the OME-Zarr.

OverwriteMode

Bases: StrEnum

How to handle existing output data: fail, replace, or extend it.

No Overwrite fails instead of touching existing data, Overwrite removes and replaces it, and Extend adds new wells or acquisitions while leaving existing ones untouched.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
15
16
17
18
19
20
21
22
23
24
25
class OverwriteMode(StrEnum):
    """How to handle existing output data: fail, replace, or extend it.

    `No Overwrite` fails instead of touching existing data, `Overwrite`
    removes and replaces it, and `Extend` adds new wells or acquisitions
    while leaving existing ones untouched.
    """

    NO_OVERWRITE = "No Overwrite"
    OVERWRITE = "Overwrite"
    EXTEND = "Extend"

PerFovGrouping

Bases: UserFacingModel

Write each field of view as its own OME-Zarr image (no mosaic).

Source code in ome_zarr_converters_tools/models/_converter_options.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class PerFovGrouping(UserFacingModel):
    """Write each field of view as its own OME-Zarr image (no mosaic)."""

    mode: Literal["Per-FOV"] = "Per-FOV"
    """How fields of view are grouped into output images."""
    model_config = ConfigDict(title="Per-FOV")

    @property
    def split_per_fov(self) -> bool:
        """Whether each field of view becomes its own OME-Zarr image."""
        return True

    def tiling_for_registration(self) -> TilingStrategy:
        """Within-image arrangement strategy for the registration pipeline.

        Per-FOV images contain a single field of view, so there is nothing to
        arrange; `InplaceTiling` is a no-op here (identical to the removed
        `NoTiling` path in the tiling step).
        """
        return InplaceTiling()
mode: Literal['Per-FOV'] = 'Per-FOV' class-attribute instance-attribute

How fields of view are grouped into output images.

split_per_fov: bool property

Whether each field of view becomes its own OME-Zarr image.

tiling_for_registration() -> TilingStrategy

Within-image arrangement strategy for the registration pipeline.

Per-FOV images contain a single field of view, so there is nothing to arrange; InplaceTiling is a no-op here (identical to the removed NoTiling path in the tiling step).

Source code in ome_zarr_converters_tools/models/_converter_options.py 
128
129
130
131
132
133
134
135
def tiling_for_registration(self) -> TilingStrategy:
    """Within-image arrangement strategy for the registration pipeline.

    Per-FOV images contain a single field of view, so there is nothing to
    arrange; `InplaceTiling` is a no-op here (identical to the removed
    `NoTiling` path in the tiling step).
    """
    return InplaceTiling()

ProcessScheduler

Bases: UserFacingModel

Parallelize the conversion using multiple processes.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
class ProcessScheduler(UserFacingModel):
    """Parallelize the conversion using multiple processes."""

    mode: Literal["Processes"] = "Processes"
    """How the conversion work is parallelized."""

    num_workers: int = Field(default=8, ge=1, title="Number of Processes")
    """Number of worker processes to use. Must be at least 1."""

    model_config = ConfigDict(title="Processes")

    def get_config(self) -> dict[str, object]:
        return {"scheduler": "processes", "num_workers": self.num_workers}
mode: Literal['Processes'] = 'Processes' class-attribute instance-attribute

How the conversion work is parallelized.

num_workers: int = Field(default=8, ge=1, title='Number of Processes') class-attribute instance-attribute

Number of worker processes to use. Must be at least 1.

RuntimeSettings

Bases: UserFacingModel

Advanced performance settings; the defaults suit most conversions.

Covers parallelism, the storage codec, and temporary storage.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
class RuntimeSettings(UserFacingModel):
    """Advanced performance settings; the defaults suit most conversions.

    Covers parallelism, the storage codec, and temporary storage.
    """

    use_zarrs_codec: bool = Field(default=False, title="Use Zarrs Codec Pipeline")
    """Read and write image data with the `zarrs` Rust backend, which is
    usually faster. **Requires** the optional `zarrs` dependency to be
    installed.
    """
    dask_scheduler: DaskScheduler = Field(
        default_factory=DefaultScheduler, title="Dask Scheduler"
    )
    """
    How the conversion work is parallelized.

    - `Threads`: parallelize using multiple threads.
    - `Processes`: parallelize using multiple processes.
    - `Synchronous`: run sequentially, without parallelism.
    - `Default`: keep the environment's parallelism settings unchanged.
    """
    temp_json_options: TempJsonOptions = Field(
        default_factory=TempJsonOptions, title="Temporary JSON Options"
    )
    """Where and when intermediate conversion data is stored on disk."""

    @model_validator(mode="after")
    def _validate(self) -> "RuntimeSettings":
        if self.use_zarrs_codec and find_spec("zarrs") is None:
            raise ImportError(
                "use_zarrs_codec=True but the 'zarrs' package is not installed. "
                "Install it with: pip install ome-zarr-converters-tools[zarrs]"
            )
        return self

    @contextmanager
    def apply(self) -> Iterator[None]:
        """Apply settings as a scoped context manager.

        Mutates `zarr.config` and/or `dask.config` only for the duration of
        the `with` block. Default-constructed settings produce a no-op
        (no zarr config mutation, and `dask.config.set({})` for the default
        scheduler).
        """
        with ExitStack() as stack:
            if self.use_zarrs_codec:
                stack.enter_context(
                    zarr.config.set({"codec_pipeline.path": "zarrs.ZarrsCodecPipeline"})
                )

            stack.enter_context(dask.config.set(self.dask_scheduler.get_config()))
            yield
dask_scheduler: DaskScheduler = Field(default_factory=DefaultScheduler, title='Dask Scheduler') class-attribute instance-attribute

How the conversion work is parallelized.

  • Threads: parallelize using multiple threads.
  • Processes: parallelize using multiple processes.
  • Synchronous: run sequentially, without parallelism.
  • Default: keep the environment's parallelism settings unchanged.
temp_json_options: TempJsonOptions = Field(default_factory=TempJsonOptions, title='Temporary JSON Options') class-attribute instance-attribute

Where and when intermediate conversion data is stored on disk.

use_zarrs_codec: bool = Field(default=False, title='Use Zarrs Codec Pipeline') class-attribute instance-attribute

Read and write image data with the zarrs Rust backend, which is usually faster. Requires the optional zarrs dependency to be installed.

apply() -> Iterator[None]

Apply settings as a scoped context manager.

Mutates zarr.config and/or dask.config only for the duration of the with block. Default-constructed settings produce a no-op (no zarr config mutation, and dask.config.set({}) for the default scheduler).

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
@contextmanager
def apply(self) -> Iterator[None]:
    """Apply settings as a scoped context manager.

    Mutates `zarr.config` and/or `dask.config` only for the duration of
    the `with` block. Default-constructed settings produce a no-op
    (no zarr config mutation, and `dask.config.set({})` for the default
    scheduler).
    """
    with ExitStack() as stack:
        if self.use_zarrs_codec:
            stack.enter_context(
                zarr.config.set({"codec_pipeline.path": "zarrs.ZarrsCodecPipeline"})
            )

        stack.enter_context(dask.config.set(self.dask_scheduler.get_config()))
        yield

Scalings

Bases: StrEnum

Scaling factor applied to the field of view size to get the chunk size.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
153
154
155
156
157
158
159
160
161
162
163
class Scalings(StrEnum):
    """Scaling factor applied to the field of view size to get the chunk size."""

    QUARTER = "0.25"
    HALF = "0.5"
    ONE = "1"
    DOUBLE = "2"
    QUADRUPLE = "4"

    def to_float(self) -> float:
        return float(self.value)

SingleImage

Bases: CollectionInterface

Collection for a stand-alone OME-Zarr image (not part of a plate).

Source code in ome_zarr_converters_tools/models/_collection.py 
61
62
63
64
65
66
67
68
69
70
class SingleImage(CollectionInterface):
    """Collection for a stand-alone OME-Zarr image (not part of a plate)."""

    image_path: str
    """Output path of the image, relative to the zarr directory
    (`.zarr` is appended if missing)."""

    def path(self) -> str:
        """Return the output path for this image."""
        return sanitize_path(f"{self.image_path}{self._suffix}")
image_path: str instance-attribute

Output path of the image, relative to the zarr directory (.zarr is appended if missing).

path() -> str

Return the output path for this image.

Source code in ome_zarr_converters_tools/models/_collection.py 
68
69
70
def path(self) -> str:
    """Return the output path for this image."""
    return sanitize_path(f"{self.image_path}{self._suffix}")

SnapToCornersTiling

Bases: UserFacingModel

Arrange the fields of view on a grid defined by their corners.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
54
55
56
57
58
59
60
61
class SnapToCornersTiling(UserFacingModel):
    """Arrange the fields of view on a grid defined by their corners."""

    model_config = ConfigDict(title="Snap to Corners")

    mode: Literal["Snap to Corners"] = "Snap to Corners"
    """Arrange the fields of view on a grid defined by their corner
    positions."""
mode: Literal['Snap to Corners'] = 'Snap to Corners' class-attribute instance-attribute

Arrange the fields of view on a grid defined by their corner positions.

SnapToGridTiling

Bases: UserFacingModel

Arrange the fields of view on a regular grid.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
41
42
43
44
45
46
47
48
49
50
51
class SnapToGridTiling(UserFacingModel):
    """Arrange the fields of view on a regular grid."""

    model_config = ConfigDict(title="Snap to Grid")

    mode: Literal["Snap to Grid"] = "Snap to Grid"
    """Arrange the fields of view on a regular grid. Only possible when
    their positions already align to a grid (potentially with overlap)."""
    tolerance: float = Field(default=1, ge=0, title="Tiling Tolerance (in pixels)")
    """Maximum misalignment (in pixels) tolerated when checking whether the
    field of view positions fit a regular grid."""
mode: Literal['Snap to Grid'] = 'Snap to Grid' class-attribute instance-attribute

Arrange the fields of view on a regular grid. Only possible when their positions already align to a grid (potentially with overlap).

tolerance: float = Field(default=1, ge=0, title='Tiling Tolerance (in pixels)') class-attribute instance-attribute

Maximum misalignment (in pixels) tolerated when checking whether the field of view positions fit a regular grid.

StageOrientation

Bases: UserFacingModel

Corrections for the stage orientation relative to the image axes.

Use these when the fields of view appear mirrored or arranged wrongly in the output image.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
class StageOrientation(UserFacingModel):
    """Corrections for the stage orientation relative to the image axes.

    Use these when the fields of view appear mirrored or arranged wrongly
    in the output image.
    """

    flip_x: bool = Field(default=False, title="Flip X")
    """Whether to flip the position along the X axis."""
    flip_y: bool = Field(default=False, title="Flip Y")
    """Whether to flip the position along the Y axis."""
    swap_xy: bool = Field(default=False, title="Swap XY")
    """Whether to swap the positions along the X and Y axes."""
flip_x: bool = Field(default=False, title='Flip X') class-attribute instance-attribute

Whether to flip the position along the X axis.

flip_y: bool = Field(default=False, title='Flip Y') class-attribute instance-attribute

Whether to flip the position along the Y axis.

swap_xy: bool = Field(default=False, title='Swap XY') class-attribute instance-attribute

Whether to swap the positions along the X and Y axes.

StagePositionCorrections

Bases: UserFacingModel

Corrections applied to the stage positions before writing the image.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class StagePositionCorrections(UserFacingModel):
    """Corrections applied to the stage positions before writing the image."""

    remove_xy_offset: Literal["Keep", "Global"] = Field(
        default="Global", title="Remove XY Offset"
    )
    """
    Whether to shift the image so its XY origin is 0.

    - `Global`: Shift all positions together so the image origin is 0.
    - `Keep`: Use the stage positions as-is. **Fails** if any position is
      negative; positive positions produce empty padding at the image origin.
    """
    remove_z_offset: Literal["Keep", "Per-FOV", "Global"] = Field(
        default="Global", title="Remove Z Offset"
    )
    """
    Whether to shift the image so its Z origin is 0.

    - `Global`: Shift all positions together so the Z origin is 0.
    - `Per-FOV`: Shift each field of view independently to Z origin 0.
    - `Keep`: Use the stage positions as-is. **Fails** if any position is
      negative; positive positions produce empty padding at the image origin.
    """
    remove_t_offset: Literal["Keep", "Global"] = Field(
        default="Global", title="Remove T Offset"
    )
    """
    Whether to shift the image so its time origin is 0.

    - `Global`: Shift all positions together so the time origin is 0.
    - `Keep`: Use the stage positions as-is. **Fails** if any position is
      negative; positive positions produce empty padding at the image origin.
    """
    remove_xy_jitter: bool = Field(default=True, title="Remove XY Jitter")
    """
    Remove small stage position inconsistencies within a field of view by
    snapping its sub-tiles to a shared origin.
    """
    reindex_channels: bool = Field(default=True, title="Reindex Channels")
    """
    If enabled, only the channels actually present are converted; if
    disabled, missing channels are stored as empty images.
    """
reindex_channels: bool = Field(default=True, title='Reindex Channels') class-attribute instance-attribute

If enabled, only the channels actually present are converted; if disabled, missing channels are stored as empty images.

remove_t_offset: Literal['Keep', 'Global'] = Field(default='Global', title='Remove T Offset') class-attribute instance-attribute

Whether to shift the image so its time origin is 0.

  • Global: Shift all positions together so the time origin is 0.
  • Keep: Use the stage positions as-is. Fails if any position is negative; positive positions produce empty padding at the image origin.
remove_xy_jitter: bool = Field(default=True, title='Remove XY Jitter') class-attribute instance-attribute

Remove small stage position inconsistencies within a field of view by snapping its sub-tiles to a shared origin.

remove_xy_offset: Literal['Keep', 'Global'] = Field(default='Global', title='Remove XY Offset') class-attribute instance-attribute

Whether to shift the image so its XY origin is 0.

  • Global: Shift all positions together so the image origin is 0.
  • Keep: Use the stage positions as-is. Fails if any position is negative; positive positions produce empty padding at the image origin.
remove_z_offset: Literal['Keep', 'Per-FOV', 'Global'] = Field(default='Global', title='Remove Z Offset') class-attribute instance-attribute

Whether to shift the image so its Z origin is 0.

  • Global: Shift all positions together so the Z origin is 0.
  • Per-FOV: Shift each field of view independently to Z origin 0.
  • Keep: Use the stage positions as-is. Fails if any position is negative; positive positions produce empty padding at the image origin.

SynchronousScheduler

Bases: UserFacingModel

Run the conversion sequentially, without parallelism.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
46
47
48
49
50
51
52
53
54
55
class SynchronousScheduler(UserFacingModel):
    """Run the conversion sequentially, without parallelism."""

    mode: Literal["Synchronous"] = "Synchronous"
    """How the conversion work is parallelized."""

    model_config = ConfigDict(title="Synchronous")

    def get_config(self) -> dict[str, object]:
        return {"scheduler": "synchronous"}
mode: Literal['Synchronous'] = 'Synchronous' class-attribute instance-attribute

How the conversion work is parallelized.

TempJsonOptions

Bases: UserFacingModel

Where and when intermediate conversion data is stored on disk.

The data is handed over between the init and compute phases of the task.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
 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
class TempJsonOptions(UserFacingModel):
    """Where and when intermediate conversion data is stored on disk.

    The data is handed over between the init and compute phases of the task.
    """

    temp_url: str = Field(default="{zarr_dir}/_tmp_json", title="Temporary Storage URL")
    """Where intermediate JSON files are written. The `{zarr_dir}`
    placeholder is replaced by the task's output directory."""
    serialization: Literal["Auto", "Memory", "JSON"] = "Auto"
    """Serialization mode for tiled image data between init and compute phases.

    - `Memory`: always keep data in-memory (skips all filesystem I/O).
    - `JSON`: always write to a temporary JSON file on disk (**required** for
      distributed Fractal runs where init and compute execute on different
      machines).
    - `Auto`: use in-memory when the total serialized payload is within
      `max_in_memory_bytes` (default 10 MiB), otherwise fall back to JSON files
      on disk.
    """
    max_in_memory_bytes: int = Field(
        default=10 * 1024 * 1024,
        ge=1,
        title="Max In-Memory Bytes",
    )
    """Maximum total size of serialized tiled image data to keep in-memory
    between init and compute phases when `serialization` is `Auto`.
    If the total size exceeds this threshold, data will be written to temporary
    JSON files on disk instead. Default is 10 MiB.
    """

    def format_temp_url(self, zarr_dir: str) -> str:
        # Route through join_url_paths (no extra parts) so a zarr_dir with a
        # trailing/duplicate/back-slash is normalized and the protocol preserved.
        return join_url_paths(self.temp_url.format(zarr_dir=zarr_dir))

    def use_in_memory(self, total_bytes: int) -> bool:
        """Resolve whether to skip disk I/O for the given total serialized size."""
        if self.serialization == "Memory":
            return True
        if self.serialization == "JSON":
            return False
        return total_bytes <= self.max_in_memory_bytes
max_in_memory_bytes: int = Field(default=(10 * 1024 * 1024), ge=1, title='Max In-Memory Bytes') class-attribute instance-attribute

Maximum total size of serialized tiled image data to keep in-memory between init and compute phases when serialization is Auto. If the total size exceeds this threshold, data will be written to temporary JSON files on disk instead. Default is 10 MiB.

serialization: Literal['Auto', 'Memory', 'JSON'] = 'Auto' class-attribute instance-attribute

Serialization mode for tiled image data between init and compute phases.

  • Memory: always keep data in-memory (skips all filesystem I/O).
  • JSON: always write to a temporary JSON file on disk (required for distributed Fractal runs where init and compute execute on different machines).
  • Auto: use in-memory when the total serialized payload is within max_in_memory_bytes (default 10 MiB), otherwise fall back to JSON files on disk.
temp_url: str = Field(default='{zarr_dir}/_tmp_json', title='Temporary Storage URL') class-attribute instance-attribute

Where intermediate JSON files are written. The {zarr_dir} placeholder is replaced by the task's output directory.

use_in_memory(total_bytes: int) -> bool

Resolve whether to skip disk I/O for the given total serialized size.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
112
113
114
115
116
117
118
def use_in_memory(self, total_bytes: int) -> bool:
    """Resolve whether to skip disk I/O for the given total serialized size."""
    if self.serialization == "Memory":
        return True
    if self.serialization == "JSON":
        return False
    return total_bytes <= self.max_in_memory_bytes

ThreadScheduler

Bases: UserFacingModel

Parallelize the conversion using multiple threads.

Source code in ome_zarr_converters_tools/models/_runtime_settings.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
class ThreadScheduler(UserFacingModel):
    """Parallelize the conversion using multiple threads."""

    mode: Literal["Threads"] = "Threads"
    """How the conversion work is parallelized."""

    num_workers: int = Field(default=8, ge=1, title="Number of Threads")
    """Number of worker threads to use. Must be at least 1."""

    model_config = ConfigDict(title="Threads")

    def get_config(self) -> dict[str, object]:
        return {"scheduler": "threads", "num_workers": self.num_workers}
mode: Literal['Threads'] = 'Threads' class-attribute instance-attribute

How the conversion work is parallelized.

num_workers: int = Field(default=8, ge=1, title='Number of Threads') class-attribute instance-attribute

Number of worker threads to use. Must be at least 1.

WriterMode

Bases: StrEnum

How image data is written, trading off speed against memory usage.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
166
167
168
169
170
171
172
173
class WriterMode(StrEnum):
    """How image data is written, trading off speed against memory usage."""

    BY_TILE = "By Tile"
    BY_FOV = "By FOV"
    BY_FOV_DASK = "By FOV (Using Dask)"
    BY_TILE_DASK = "By Tile (Using Dask)"
    IN_MEMORY = "In Memory"

basename_url(url: str) -> str

Return the last path component of a URL.

A trailing slash is stripped so a directory yields its own name.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
106
107
108
109
110
111
def basename_url(url: str) -> str:
    """Return the last path component of a URL.

    A trailing slash is stripped so a directory yields its own name.
    """
    return posixpath.basename(url.replace("\\", "/").rstrip("/"))

default_axes_builder(is_time_series: bool) -> list[CANONICAL_AXES_TYPE]

Build default axes list.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
33
34
35
36
37
38
def default_axes_builder(is_time_series: bool) -> list[CANONICAL_AXES_TYPE]:
    """Build default axes list."""
    if is_time_series:
        return ["t", "c", "z", "y", "x"]
    else:
        return ["c", "z", "y", "x"]

filesystem_for_url(url: str, error_msg_prefix: str = 'File handling') -> fsspec.AbstractFileSystem

Return the fsspec filesystem for a URL; raise if its type is unsupported.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def filesystem_for_url(
    url: str, error_msg_prefix: str = "File handling"
) -> fsspec.AbstractFileSystem:
    """Return the fsspec filesystem for a URL; raise if its type is unsupported."""
    url_type = find_url_type(url)
    if url_type == UrlType.NOT_SUPPORTED:
        raise NotImplementedError(
            f"{error_msg_prefix} for URL {url} "
            f"with detected type {url_type} is not implemented yet."
        )
    try:
        return fsspec.filesystem(url_type.value)
    except ImportError as e:
        raise ImportError(
            f"Accessing {url} requires the '{url_type.value}' fsspec backend, "
            "which is not installed. Install it with: "
            "pip install 'ome-zarr-converters-tools[s3]'"
        ) from e

find_url_type(url: str) -> UrlType

Classify a URL as local, S3, or unsupported (see UrlType).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
39
40
41
42
43
44
45
def find_url_type(url: str) -> UrlType:
    """Classify a URL as local, S3, or unsupported (see `UrlType`)."""
    if url.startswith("s3://"):
        return UrlType.S3
    elif _is_local_absolute(url):
        return UrlType.LOCAL
    return UrlType.NOT_SUPPORTED

glob_url_paths(*, base_url: str | None, pattern: str) -> list[str]

Glob a URL pattern, re-prefixing the protocol on each match.

Uses filesystem_for_url so it works for local and S3 URLs, and re-prefixes the protocol on each result so matches stay usable URLs. If base_url is None, pattern is treated as absolute — it must then itself be absolute (/…, ~/…, s3://…, or a Windows path), since a relative pattern classifies as NOT_SUPPORTED and raises.

A literal, non-existent pattern yields an empty list (fsspec behaviour that downstream code relies on for existence checks).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def glob_url_paths(*, base_url: str | None, pattern: str) -> list[str]:
    """Glob a URL pattern, re-prefixing the protocol on each match.

    Uses `filesystem_for_url` so it works for local and S3 URLs, and
    re-prefixes the protocol on each result so matches stay usable URLs. If
    `base_url` is `None`, `pattern` is treated as absolute — it must then
    itself be absolute (`/…`, `~/…`, `s3://…`, or a Windows path), since a
    relative pattern classifies as `NOT_SUPPORTED` and raises.

    A literal, non-existent pattern yields an empty list (fsspec behaviour that
    downstream code relies on for existence checks).
    """
    if base_url is not None:
        pattern = join_url_paths(base_url, pattern)
    fs = filesystem_for_url(pattern)
    protocol = fsspec.core.split_protocol(pattern)[0]
    matched = fs.glob(pattern)
    if protocol is not None:
        return [f"{protocol}://{p}" for p in matched]
    return matched

is_absolute_url(url: str) -> bool

Return True if the URL is absolute (protocol, root, home, or Windows).

Host-independent: shares _is_local_absolute with find_url_type so a Windows drive/UNC path or a ~-anchored home path is treated as absolute even when the suite runs on POSIX. Falls back to Path(path).is_absolute() for anything else. Used to decide whether a path from a manifest/CSV needs to be resolved against a base directory.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
def is_absolute_url(url: str) -> bool:
    """Return True if the URL is absolute (protocol, root, home, or Windows).

    Host-independent: shares `_is_local_absolute` with `find_url_type` so a
    Windows drive/UNC path or a `~`-anchored home path is treated as absolute
    even when the suite runs on POSIX. Falls back to `Path(path).is_absolute()`
    for anything else. Used to decide whether a path from a manifest/CSV needs
    to be resolved against a base directory.
    """
    protocol, path = fsspec.core.split_protocol(url)
    if protocol is not None or _is_local_absolute(path):
        return True
    return Path(path).is_absolute()

join_url_paths(base_url: str, *paths: str) -> str

Join path components to a base URL, normalizing ./.. segments.

Used instead of os.path.join or pathlib.Path to support both local and S3 URLs. Resolves ./.. and collapses redundant slashes while staying protocol- and Windows-safe: it uses posixpath.normpath (never os.path.normpath, which on Windows would rewrite forward slashes to backslashes and corrupt S3 keys).

Raises:

  • ValueError –

    if .. segments would ascend above the network location of a protocol URL (e.g. join_url_paths("s3://bucket", "..", "x") would otherwise silently drop the bucket).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
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
def join_url_paths(base_url: str, *paths: str) -> str:
    """Join path components to a base URL, normalizing `.`/`..` segments.

    Used instead of `os.path.join` or `pathlib.Path` to support both local
    and S3 URLs. Resolves `.`/`..` and collapses redundant slashes while
    staying protocol- and Windows-safe: it uses `posixpath.normpath` (never
    `os.path.normpath`, which on Windows would rewrite forward slashes to
    backslashes and corrupt S3 keys).

    Raises:
        ValueError: if `..` segments would ascend above the network location
            of a protocol URL (e.g. `join_url_paths("s3://bucket", "..", "x")`
            would otherwise silently drop the bucket).
    """
    protocol, base = fsspec.core.split_protocol(base_url)
    combined = f"{base}/{'/'.join(str(p) for p in paths)}".replace("\\", "/")
    joined = posixpath.normpath(combined)
    if protocol is None:
        return joined
    # The first component after the protocol is the network location (e.g. the
    # S3 bucket) and is inviolable — reject any `..` that escapes it.
    netloc = base.replace("\\", "/").split("/", 1)[0]
    if joined != netloc and not joined.startswith(f"{netloc}/"):
        raise ValueError(
            f"Path escapes network location for URL: {base_url!r} + {paths!r}"
        )
    return f"{protocol}://{joined}"

local_url_to_path(url: str) -> Path

Convert a local URL to an absolute Path, expanding ~.

Does not touch the filesystem.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
48
49
50
51
52
53
def local_url_to_path(url: str) -> Path:
    """Convert a local URL to an absolute Path, expanding `~`.

    Does not touch the filesystem.
    """
    return Path(url).expanduser().resolve()

parent_url(url: str) -> str

Return the parent directory of a URL path.

Robust to forward/back slashes and URL protocols, and always returns POSIX / separators regardless of host OS (uses posixpath for the local branch too, never OS-native pathlib, which would emit \ on Windows). For remote protocols the first component after the protocol is the network location (e.g. the S3 bucket), which has no parent.

Raises:

  • ValueError –

    if url is a filesystem/network-location root with no parent (e.g. /, s3://bucket, or an empty string).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def parent_url(url: str) -> str:
    r"""Return the parent directory of a URL path.

    Robust to forward/back slashes and URL protocols, and always returns POSIX
    `/` separators regardless of host OS (uses `posixpath` for the local
    branch too, never OS-native `pathlib`, which would emit `\` on Windows).
    For remote protocols the first component after the protocol is the network
    location (e.g. the S3 bucket), which has no parent.

    Raises:
        ValueError: if `url` is a filesystem/network-location root with no
            parent (e.g. `/`, `s3://bucket`, or an empty string).
    """
    protocol, path = fsspec.core.split_protocol(url)
    path = path.replace("\\", "/").rstrip("/")
    parent = posixpath.dirname(path)
    if parent == "":
        raise ValueError(f"No parent directory for URL: {url}")
    return parent if protocol is None else f"{protocol}://{parent}"

URL & Path Utilities

Protocol-aware path helpers that work transparently for local paths and remote s3:// URLs, and are robust to Windows backslash separators (they always emit POSIX /). Use these instead of os.path / pathlib whenever a location may point at object storage -- they are the URL equivalents of os.path.join / dirname / basename / isabs / glob, plus fsspec filesystem resolution.

url vs path naming

Fields and helpers named *_url (e.g. filesystem_for_url, join_url_paths) take an fsspec-style location that may carry a protocol (s3://…) or be a local path. Fields named *_path (e.g. SingleImage.image_path, DefaultImageLoader.file_path) are collection-relative output locations resolved against the zarr directory or a resource base. s3:// support requires the optional s3 extra: pip install ome-zarr-converters-tools[s3].

Key exports: join_url_paths, parent_url, basename_url, is_absolute_url, glob_url_paths, filesystem_for_url, find_url_type, local_url_to_path, UrlType.

ome_zarr_converters_tools.models

Models and types definitions for the ome_zarr_converters_tools.

UrlType

Bases: Enum

Source code in ome_zarr_converters_tools/models/_url_utils.py 
12
13
14
15
class UrlType(Enum):
    LOCAL = "local"
    S3 = "s3"
    NOT_SUPPORTED = "not_supported"

join_url_paths(base_url: str, *paths: str) -> str

Join path components to a base URL, normalizing ./.. segments.

Used instead of os.path.join or pathlib.Path to support both local and S3 URLs. Resolves ./.. and collapses redundant slashes while staying protocol- and Windows-safe: it uses posixpath.normpath (never os.path.normpath, which on Windows would rewrite forward slashes to backslashes and corrupt S3 keys).

Raises:

  • ValueError –

    if .. segments would ascend above the network location of a protocol URL (e.g. join_url_paths("s3://bucket", "..", "x") would otherwise silently drop the bucket).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
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
def join_url_paths(base_url: str, *paths: str) -> str:
    """Join path components to a base URL, normalizing `.`/`..` segments.

    Used instead of `os.path.join` or `pathlib.Path` to support both local
    and S3 URLs. Resolves `.`/`..` and collapses redundant slashes while
    staying protocol- and Windows-safe: it uses `posixpath.normpath` (never
    `os.path.normpath`, which on Windows would rewrite forward slashes to
    backslashes and corrupt S3 keys).

    Raises:
        ValueError: if `..` segments would ascend above the network location
            of a protocol URL (e.g. `join_url_paths("s3://bucket", "..", "x")`
            would otherwise silently drop the bucket).
    """
    protocol, base = fsspec.core.split_protocol(base_url)
    combined = f"{base}/{'/'.join(str(p) for p in paths)}".replace("\\", "/")
    joined = posixpath.normpath(combined)
    if protocol is None:
        return joined
    # The first component after the protocol is the network location (e.g. the
    # S3 bucket) and is inviolable — reject any `..` that escapes it.
    netloc = base.replace("\\", "/").split("/", 1)[0]
    if joined != netloc and not joined.startswith(f"{netloc}/"):
        raise ValueError(
            f"Path escapes network location for URL: {base_url!r} + {paths!r}"
        )
    return f"{protocol}://{joined}"

parent_url(url: str) -> str

Return the parent directory of a URL path.

Robust to forward/back slashes and URL protocols, and always returns POSIX / separators regardless of host OS (uses posixpath for the local branch too, never OS-native pathlib, which would emit \ on Windows). For remote protocols the first component after the protocol is the network location (e.g. the S3 bucket), which has no parent.

Raises:

  • ValueError –

    if url is a filesystem/network-location root with no parent (e.g. /, s3://bucket, or an empty string).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def parent_url(url: str) -> str:
    r"""Return the parent directory of a URL path.

    Robust to forward/back slashes and URL protocols, and always returns POSIX
    `/` separators regardless of host OS (uses `posixpath` for the local
    branch too, never OS-native `pathlib`, which would emit `\` on Windows).
    For remote protocols the first component after the protocol is the network
    location (e.g. the S3 bucket), which has no parent.

    Raises:
        ValueError: if `url` is a filesystem/network-location root with no
            parent (e.g. `/`, `s3://bucket`, or an empty string).
    """
    protocol, path = fsspec.core.split_protocol(url)
    path = path.replace("\\", "/").rstrip("/")
    parent = posixpath.dirname(path)
    if parent == "":
        raise ValueError(f"No parent directory for URL: {url}")
    return parent if protocol is None else f"{protocol}://{parent}"

basename_url(url: str) -> str

Return the last path component of a URL.

A trailing slash is stripped so a directory yields its own name.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
106
107
108
109
110
111
def basename_url(url: str) -> str:
    """Return the last path component of a URL.

    A trailing slash is stripped so a directory yields its own name.
    """
    return posixpath.basename(url.replace("\\", "/").rstrip("/"))

is_absolute_url(url: str) -> bool

Return True if the URL is absolute (protocol, root, home, or Windows).

Host-independent: shares _is_local_absolute with find_url_type so a Windows drive/UNC path or a ~-anchored home path is treated as absolute even when the suite runs on POSIX. Falls back to Path(path).is_absolute() for anything else. Used to decide whether a path from a manifest/CSV needs to be resolved against a base directory.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
def is_absolute_url(url: str) -> bool:
    """Return True if the URL is absolute (protocol, root, home, or Windows).

    Host-independent: shares `_is_local_absolute` with `find_url_type` so a
    Windows drive/UNC path or a `~`-anchored home path is treated as absolute
    even when the suite runs on POSIX. Falls back to `Path(path).is_absolute()`
    for anything else. Used to decide whether a path from a manifest/CSV needs
    to be resolved against a base directory.
    """
    protocol, path = fsspec.core.split_protocol(url)
    if protocol is not None or _is_local_absolute(path):
        return True
    return Path(path).is_absolute()

glob_url_paths(*, base_url: str | None, pattern: str) -> list[str]

Glob a URL pattern, re-prefixing the protocol on each match.

Uses filesystem_for_url so it works for local and S3 URLs, and re-prefixes the protocol on each result so matches stay usable URLs. If base_url is None, pattern is treated as absolute — it must then itself be absolute (/…, ~/…, s3://…, or a Windows path), since a relative pattern classifies as NOT_SUPPORTED and raises.

A literal, non-existent pattern yields an empty list (fsspec behaviour that downstream code relies on for existence checks).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def glob_url_paths(*, base_url: str | None, pattern: str) -> list[str]:
    """Glob a URL pattern, re-prefixing the protocol on each match.

    Uses `filesystem_for_url` so it works for local and S3 URLs, and
    re-prefixes the protocol on each result so matches stay usable URLs. If
    `base_url` is `None`, `pattern` is treated as absolute — it must then
    itself be absolute (`/…`, `~/…`, `s3://…`, or a Windows path), since a
    relative pattern classifies as `NOT_SUPPORTED` and raises.

    A literal, non-existent pattern yields an empty list (fsspec behaviour that
    downstream code relies on for existence checks).
    """
    if base_url is not None:
        pattern = join_url_paths(base_url, pattern)
    fs = filesystem_for_url(pattern)
    protocol = fsspec.core.split_protocol(pattern)[0]
    matched = fs.glob(pattern)
    if protocol is not None:
        return [f"{protocol}://{p}" for p in matched]
    return matched

filesystem_for_url(url: str, error_msg_prefix: str = 'File handling') -> fsspec.AbstractFileSystem

Return the fsspec filesystem for a URL; raise if its type is unsupported.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def filesystem_for_url(
    url: str, error_msg_prefix: str = "File handling"
) -> fsspec.AbstractFileSystem:
    """Return the fsspec filesystem for a URL; raise if its type is unsupported."""
    url_type = find_url_type(url)
    if url_type == UrlType.NOT_SUPPORTED:
        raise NotImplementedError(
            f"{error_msg_prefix} for URL {url} "
            f"with detected type {url_type} is not implemented yet."
        )
    try:
        return fsspec.filesystem(url_type.value)
    except ImportError as e:
        raise ImportError(
            f"Accessing {url} requires the '{url_type.value}' fsspec backend, "
            "which is not installed. Install it with: "
            "pip install 'ome-zarr-converters-tools[s3]'"
        ) from e

find_url_type(url: str) -> UrlType

Classify a URL as local, S3, or unsupported (see UrlType).

Source code in ome_zarr_converters_tools/models/_url_utils.py 
39
40
41
42
43
44
45
def find_url_type(url: str) -> UrlType:
    """Classify a URL as local, S3, or unsupported (see `UrlType`)."""
    if url.startswith("s3://"):
        return UrlType.S3
    elif _is_local_absolute(url):
        return UrlType.LOCAL
    return UrlType.NOT_SUPPORTED

local_url_to_path(url: str) -> Path

Convert a local URL to an absolute Path, expanding ~.

Does not touch the filesystem.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
48
49
50
51
52
53
def local_url_to_path(url: str) -> Path:
    """Convert a local URL to an absolute Path, expanding `~`.

    Does not touch the filesystem.
    """
    return Path(url).expanduser().resolve()

Pipelines

Pipeline functions for aggregation, registration, filtering, validation, and writing. This module orchestrates the full conversion flow: aggregating tiles into images, running registration steps, applying filters, and writing the final OME-Zarr datasets. It also provides extension points for custom filters, validators, and registration steps.

Key exports: tiles_aggregation_pipeline, tiled_image_creation_pipeline, build_default_registration_pipeline, apply_registration_pipeline, apply_filter_pipeline, add_filter, add_registration_func, add_validator.

ome_zarr_converters_tools.pipelines

Pipeline modules for OME-Zarr converters tools.

add_collection_handler(*, function: SetupCollectionFunction, collection_type: str | None = None, overwrite: bool = False) -> None

Register a new collection setup handler.

The collection setup handler is responsible for setting up the collection structure and metadata in the Zarr group.

Note

Registrations are process-global: under MultiprocessingRunner, worker processes re-import the consumer's modules, so custom handlers must be registered at import time of the module that defines them to be visible in workers.

Parameters:

  • collection_type (str | None, default: None ) –

    Name of the collection setup handler. By convention, the name of the CollectionInterfaceType, e.g., 'SingleImage' or 'ImageInPlate'. Defaults to function.__name__.

  • function (SetupCollectionFunction) –

    Function that performs the collection setup step.

  • overwrite (bool, default: False ) –

    Whether to overwrite an existing collection setup step with the same name.

Source code in ome_zarr_converters_tools/pipelines/_collection_setup.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def add_collection_handler(
    *,
    function: SetupCollectionFunction,
    collection_type: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new collection setup handler.

    The collection setup handler is responsible for setting up the
    collection structure and metadata in the Zarr group.

    Note:
        Registrations are process-global: under `MultiprocessingRunner`,
        worker processes re-import the consumer's modules, so custom handlers
        must be registered at import time of the module that defines them to
        be visible in workers.

    Args:
        collection_type: Name of the collection setup handler. By convention,
            the name of the CollectionInterfaceType, e.g., 'SingleImage'
            or 'ImageInPlate'. Defaults to `function.__name__`.
        function: Function that performs the collection setup step.
        overwrite: Whether to overwrite an existing collection setup step
            with the same name.
    """
    _collection_setup_registry.add(
        function=function, name=collection_type, overwrite=overwrite
    )

add_filter(*, function: FilterFunctionProtocol, name: str | None = None, overwrite: bool = False) -> None

Register a new filter.

Note

Registrations are process-global: under MultiprocessingRunner, worker processes re-import the consumer's modules, so custom filters must be registered at import time of the module that defines them to be visible in workers.

Parameters:

  • function (FilterFunctionProtocol) –

    Function that performs the filter step.

  • name (str | None, default: None ) –

    Name of the filter step. Defaults to function.__name__.

  • overwrite (bool, default: False ) –

    Whether to overwrite an existing filter step with the same name.

Source code in ome_zarr_converters_tools/pipelines/_filters.py 
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def add_filter(
    *,
    function: FilterFunctionProtocol,
    name: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new filter.

    Note:
        Registrations are process-global: under `MultiprocessingRunner`,
        worker processes re-import the consumer's modules, so custom filters
        must be registered at import time of the module that defines them to
        be visible in workers.

    Args:
        function: Function that performs the filter step.
        name: Name of the filter step. Defaults to `function.__name__`.
        overwrite: Whether to overwrite an existing filter step
            with the same name.
    """
    _filter_registry.add(function=function, name=name, overwrite=overwrite)

add_registration_func(*, function: Callable[..., TiledImage], name: str | None = None, overwrite: bool = False) -> None

Register a new registration step function.

Note

Registrations are process-global: under MultiprocessingRunner, worker processes re-import the consumer's modules, so custom registration steps must be registered at import time of the module that defines them to be visible in workers.

Parameters:

  • function (Callable[..., TiledImage]) –

    Function that performs the registration step.

  • name (str | None, default: None ) –

    Name of the registration step. Defaults to function.__name__.

  • overwrite (bool, default: False ) –

    Whether to overwrite an existing registration step.

Source code in ome_zarr_converters_tools/pipelines/_registration_pipeline.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def add_registration_func(
    *,
    function: Callable[..., TiledImage],
    name: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new registration step function.

    Note:
        Registrations are process-global: under `MultiprocessingRunner`,
        worker processes re-import the consumer's modules, so custom
        registration steps must be registered at import time of the module
        that defines them to be visible in workers.

    Args:
        function: Function that performs the registration step.
        name: Name of the registration step. Defaults to `function.__name__`.
        overwrite: Whether to overwrite an existing registration step.
    """
    _registration_registry.add(function=function, name=name, overwrite=overwrite)

add_validator(*, function: ValidatorFunctionProtocol, name: str | None = None, overwrite: bool = False) -> None

Register a new validator function.

Validators are pre-flight checks that run on each TiledImage after aggregation and raise on failure; use them to front-load errors that would otherwise only surface during compute.

Note

Registrations are process-global: under MultiprocessingRunner, worker processes re-import the consumer's modules, so custom validators must be registered at import time of the module that defines them to be visible in workers.

Parameters:

  • function (ValidatorFunctionProtocol) –

    Function that performs the validation step. It is called as function(tiled_image, validator_params=step, resource=resource) and must raise on failure.

  • name (str | None, default: None ) –

    Name of the validation step. Defaults to function.__name__.

  • overwrite (bool, default: False ) –

    Whether to overwrite an existing validation step with the same name.

Source code in ome_zarr_converters_tools/pipelines/_validators.py 
 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
def add_validator(
    *,
    function: ValidatorFunctionProtocol,
    name: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new validator function.

    Validators are pre-flight checks that run on each `TiledImage` after
    aggregation and raise on failure; use them to front-load errors that
    would otherwise only surface during compute.

    Note:
        Registrations are process-global: under `MultiprocessingRunner`,
        worker processes re-import the consumer's modules, so custom
        validators must be registered at import time of the module that
        defines them to be visible in workers.

    Args:
        function: Function that performs the validation step. It is called
            as `function(tiled_image, validator_params=step, resource=resource)`
            and must raise on failure.
        name: Name of the validation step. Defaults to `function.__name__`.
        overwrite: Whether to overwrite an existing validation step
            with the same name.
    """
    _validator_registry.add(function=function, name=name, overwrite=overwrite)

setup_ome_zarr_collection(*, tiled_images: list[TiledImage], collection_type: str, zarr_dir: str, ngff_version: NgffVersions = DefaultNgffVersion, overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE) -> None

Set up the collection in the Zarr group using the specified handler.

Parameters:

  • tiled_images (list[TiledImage]) –

    List of TiledImage to set up the collection for.

  • collection_type (str) –

    Type of collection setup handler to use.

  • zarr_dir (str) –

    The base directory for the zarr data.

  • ngff_version (NgffVersions, default: DefaultNgffVersion ) –

    NGFF version to use for the collection setup.

  • overwrite_mode (OverwriteMode, default: NO_OVERWRITE ) –

    Overwrite mode to use for the collection setup.

Source code in ome_zarr_converters_tools/pipelines/_collection_setup.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def setup_ome_zarr_collection(
    *,
    tiled_images: list[TiledImage],
    collection_type: str,
    zarr_dir: str,
    ngff_version: NgffVersions = DefaultNgffVersion,
    overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE,
) -> None:
    """Set up the collection in the Zarr group using the specified handler.

    Args:
        tiled_images: List of TiledImage to set up the collection for.
        collection_type: Type of collection setup handler to use.
        zarr_dir: The base directory for the zarr data.
        ngff_version: NGFF version to use for the collection setup.
        overwrite_mode: Overwrite mode to use for the collection setup.
    """
    setup_function = _collection_setup_registry.get(collection_type)
    return setup_function(
        tiled_images=tiled_images,
        zarr_dir=zarr_dir,
        ngff_version=ngff_version,
        overwrite_mode=overwrite_mode,
    )

setup_singleimage(zarr_dir: str, tiled_images: list[TiledImage], ngff_version: NgffVersions = DefaultNgffVersion, overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE) -> None

Set up a SingleImage collection (overwrite-mode enforcement only).

SingleImage outputs do not need an upfront zarr skeleton — the zarr group is created during the compute task. This handler only enforces the OverwriteMode contract.

Source code in ome_zarr_converters_tools/pipelines/_collection_setup.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def setup_singleimage(
    zarr_dir: str,
    tiled_images: list[TiledImage],
    ngff_version: NgffVersions = DefaultNgffVersion,
    overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE,
) -> None:
    """Set up a SingleImage collection (overwrite-mode enforcement only).

    SingleImage outputs do not need an upfront zarr skeleton — the zarr
    group is created during the compute task. This handler only enforces
    the OverwriteMode contract.
    """
    for tiled_image in tiled_images:
        collection = tiled_image.collection
        if not isinstance(collection, SingleImage):
            raise ValueError(f"Expected SingleImage collection, got {type(collection)}")
        zarr_url = join_url_paths(zarr_dir, collection.path())
        if overwrite_mode == OverwriteMode.NO_OVERWRITE:
            fs = filesystem_for_url(zarr_url)
            if fs.exists(zarr_url):
                raise FileExistsError(
                    f"A zarr already exists at {zarr_url} "
                    f"(overwrite_mode={OverwriteMode.NO_OVERWRITE.value}). "
                    f"Set overwrite_mode={OverwriteMode.OVERWRITE.value} to "
                    f"replace it."
                )

tiled_image_creation_pipeline(*, zarr_url: str, tiled_image: TiledImage, registration_pipeline: list[RegistrationStep], converter_options: ConverterOptions, writer_mode: WriterMode, overwrite_mode: OverwriteMode, resource: Any | None = None) -> OmeZarrContainer

Write a TiledImage from a dictionary.

Source code in ome_zarr_converters_tools/pipelines/_tiled_image_creation_pipeline.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def tiled_image_creation_pipeline(
    *,
    zarr_url: str,
    tiled_image: TiledImage,
    registration_pipeline: list[RegistrationStep],
    converter_options: ConverterOptions,
    writer_mode: WriterMode,
    overwrite_mode: OverwriteMode,
    resource: Any | None = None,
) -> OmeZarrContainer:
    """Write a TiledImage from a dictionary."""
    with converter_options.runtime_settings.apply():
        logger.info("Applying registration pipeline to TiledImage.")
        tiled_image = apply_registration_pipeline(tiled_image, registration_pipeline)
        logger.info("Starting to write TiledImage as OME-Zarr.")
        omezarr = write_tiled_image_as_zarr(
            zarr_url=zarr_url,
            tiled_image=tiled_image,
            converter_options=converter_options,
            writer_mode=writer_mode,
            overwrite_mode=overwrite_mode,
            resource=resource,
        )
        return omezarr

tiles_aggregation_pipeline(tiles: list[Tile], *, converter_options: ConverterOptions, filters: Sequence[FilterModel] | None = None, validators: Sequence[ValidatorModel] | None = None, resource: Any | None = None) -> list[TiledImage]

Process tiles and aggregates them into TiledImages.

This function applies optional filters to the input tiles and then constructs TiledImage models from the processed tiles.

Parameters:

  • tiles (list[Tile]) –

    List of Tile models to process.

  • converter_options (ConverterOptions) –

    ConverterOptions model for the conversion.

  • filters (Sequence[FilterModel] | None, default: None ) –

    Optional sequence of filter steps to apply to the tiles.

  • validators (Sequence[ValidatorModel] | None, default: None ) –

    Optional sequence of validator steps to apply to the tiles.

  • resource (Any | None, default: None ) –

    Optional resource to assist in processing.

Returns:

  • list[TiledImage] –

    A list of TiledImage models created from the processed tiles.

Source code in ome_zarr_converters_tools/pipelines/_tiles_aggregation_pipeline.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def tiles_aggregation_pipeline(
    tiles: list[Tile],
    *,
    converter_options: ConverterOptions,
    filters: Sequence[FilterModel] | None = None,
    validators: Sequence[ValidatorModel] | None = None,
    resource: Any | None = None,
) -> list[TiledImage]:
    """Process tiles and aggregates them into TiledImages.

    This function applies optional filters to the input tiles and then
    constructs TiledImage models from the processed tiles.

    Args:
        tiles: List of Tile models to process.
        converter_options: ConverterOptions model for the conversion.
        filters: Optional sequence of filter steps to apply to the tiles.
        validators: Optional sequence of validator steps to apply to the tiles.
        resource: Optional resource to assist in processing.

    Returns:
        A list of TiledImage models created from the processed tiles.
    """
    if filters is not None:
        tiles = apply_filter_pipeline(tiles, filters_config=filters)
    tiled_images = tiled_image_from_tiles(
        tiles=tiles,
        split_per_fov=converter_options.grouping.split_per_fov,
        resource=resource,
    )
    if validators is not None:
        tiled_images = apply_validator_pipeline(
            tiled_images, validators_config=validators, resource=resource
        )
    return tiled_images

write_tiled_image_as_zarr(*, zarr_url: str, tiled_image: TiledImage, converter_options: ConverterOptions, writer_mode: WriterMode, overwrite_mode: OverwriteMode, resource: Any | None = None) -> OmeZarrContainer

Write a TiledImage as a Zarr file.

Parameters:

  • zarr_url (str) –

    URL to write the Zarr file to.

  • tiled_image (TiledImage) –

    TiledImage model to write.

  • converter_options (ConverterOptions) –

    Options for the OME-Zarr conversion.

  • writer_mode (WriterMode) –

    Mode for writing the data.

  • overwrite_mode (OverwriteMode) –

    Mode to handle existing data.

  • resource (Any | None, default: None ) –

    Optional resource to pass to the image loaders.

Returns:

  • OmeZarrContainer –

    The written OME-Zarr container.

Source code in ome_zarr_converters_tools/pipelines/_write_ome_zarr.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def write_tiled_image_as_zarr(
    *,
    zarr_url: str,
    tiled_image: TiledImage,
    converter_options: ConverterOptions,
    writer_mode: WriterMode,
    overwrite_mode: OverwriteMode,
    resource: Any | None = None,
) -> OmeZarrContainer:
    """Write a TiledImage as a Zarr file.

    Args:
        zarr_url: URL to write the Zarr file to.
        tiled_image: TiledImage model to write.
        converter_options: Options for the OME-Zarr conversion.
        writer_mode: Mode for writing the data.
        overwrite_mode: Mode to handle existing data.
        resource: Optional resource to pass to the image loaders.

    Returns:
        The written OME-Zarr container.
    """
    if overwrite_mode == OverwriteMode.NO_OVERWRITE:
        mode = "w-"
    elif overwrite_mode == OverwriteMode.OVERWRITE:
        mode = "w"
    else:  # extend
        mode = "a"
    zarr_format = 2 if converter_options.omezarr_options.ngff_version == "0.4" else 3
    # Whether a store already exists must be checked before `zarr.open_group`,
    # since mode "w" would truncate it and mode "a" would create it.
    store_pre_exists = filesystem_for_url(zarr_url).exists(zarr_url)
    base_group = zarr.open_group(store=zarr_url, mode=mode, zarr_format=zarr_format)
    omezarr_options = converter_options.omezarr_options
    if overwrite_mode == OverwriteMode.EXTEND and store_pre_exists:
        # Reuse the existing container. A failure to open here (corrupt/partial
        # store, version mismatch) is a real error that must surface, never be
        # silently overwritten with a fresh, empty container.
        return open_ome_zarr_container(base_group, cache=True)

    tiled_image.regions = _region_to_pixel_coordinates(
        tiled_image.regions,
        tiled_image.pixel_size,
    )
    channels_meta = build_channels_meta(tiled_image)
    ome_zarr = create_empty_ome_zarr(
        store=base_group,
        axes_names=tiled_image.axes,
        shape=tiled_image.output_shape(),
        chunks=_compute_chunk_size(tiled_image, omezarr_options),
        pixelsize=tiled_image.xy_pixel_size,
        z_spacing=tiled_image.z_spacing,
        time_spacing=tiled_image.t_spacing,
        levels=omezarr_options.levels.to_ngio_levels(),
        channels_meta=channels_meta,
        translation=tiled_image.translation,
        overwrite=True,
        ngff_version=omezarr_options.ngff_version,
        dtype=tiled_image.data_type,
    )
    image = ome_zarr.get_image()
    write_to_zarr(
        image=image,
        tiled_image=tiled_image,
        resource=resource,
        writer_mode=writer_mode,
    )
    image.consolidate()
    ome_zarr.set_channel_windows_with_percentiles()
    logger.info("OME-Zarr image creation and data writing complete.")

    fov_tiles = tiled_image.group_by_fov()
    if len(fov_tiles) > 1:
        rois = []
        for fov_tile in fov_tiles:
            roi_union = fov_tile.roi().to_world(pixel_size=tiled_image.pixel_size)
            rois.append(roi_union)

        roi_table = RoiTable(rois=rois)
        ome_zarr.add_table(
            "FOV_ROI_table", roi_table, backend=omezarr_options.table_backend
        )

    well_roi = ome_zarr.build_image_roi_table()
    ome_zarr.add_table(
        "well_ROI_table", well_roi, backend=omezarr_options.table_backend
    )
    condition_table = _attribute_to_condition_table(tiled_image.attributes)
    if condition_table is not None:
        ome_zarr.add_table("condition_table", condition_table, backend="csv")
    logger.info("Finished writing OME-Zarr Tables and metadata.")
    return ome_zarr

Fractal Integration

Utilities for building Fractal platform tasks. This module provides setup_images_for_conversion() (init task) and generic_compute_task() (compute task factory) for parallelizing conversions across a Fractal cluster.

Key exports: setup_images_for_conversion, generic_compute_task, ConvertParallelInitArgs, AcquisitionOptions. Lower-level JSON plumbing (tiled_image_from_json, dump_to_json, …) is also exported from this module for task authors who need to serialize TiledImages across the init/compute boundary; it is intentionally namespaced under fractal rather than the package root.

ome_zarr_converters_tools.fractal

API for building OME-Zarr converters tasks for Fractal.

AcquisitionOptions

Bases: UserFacingModel

Per-acquisition settings: channels, pixel sizes, axes, and filters.

In Fractal tasks these settings override/update the acquisition details parsed from the raw metadata (AcquisitionDetails).

Source code in ome_zarr_converters_tools/fractal/_models.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
class AcquisitionOptions(UserFacingModel):
    """Per-acquisition settings: channels, pixel sizes, axes, and filters.

    In Fractal tasks these settings override/update the acquisition details
    parsed from the raw metadata (`AcquisitionDetails`).
    """

    channels: list[ChannelInfoUI] | None = None
    """Names, wavelengths, and display colors of the channels. If left
    empty, the channel information parsed from the raw metadata is used."""
    pixel_info: PixelSizeModel | None = Field(
        default=None, title="Pixel Size Information"
    )
    """Override the pixel size and the Z/time spacing of the images. If left
    empty, the values parsed from the raw metadata are used."""
    condition_table_path: str | None = None
    """Optional path to a condition table CSV file to store in the plate
    metadata."""
    axes: str | None = None
    """Axes of the image data, e.g. `czyx`. If left empty, the axes parsed
    from the raw metadata are used."""
    data_type: DataTypeEnum = Field(default=DataTypeEnum.AUTODETECT, title="Data Type")
    """Pixel data type of the output image. `autodetect` infers it from the
    input images."""
    stage_orientation: StageOrientation = Field(
        default_factory=StageOrientation, title="Stage Orientation"
    )
    """Corrections for the orientation of the microscope stage relative to
    the image axes."""
    filters: list[ImplementedFilters] = Field(default_factory=list)
    """Filters selecting which tiles of the acquisition are converted."""

    def to_axes_list(self) -> list[CANONICAL_AXES_TYPE] | None:
        """Convert axes string to list of axes."""
        if self.axes is None:
            return None
        _axes = []
        for ax in self.axes:
            if ax not in canonical_axes:
                raise ValueError(f"Invalid axis '{ax}' in axes string.")
            _axes.append(ax)
        return _axes  # type: ignore

    def update_acquisition_details(
        self,
        acquisition_details: AcquisitionDetails,
    ) -> AcquisitionDetails:
        """Update AcquisitionDetails model with options from this model.

        Args:
            acquisition_details: AcquisitionDetails model to update.

        Returns:
            Updated AcquisitionDetails model.

        """
        updated_details = acquisition_details.model_copy()
        if self.channels is not None:
            _updated_channels = []
            for channel in self.channels:
                _updated_channels.append(
                    ChannelInfo(
                        channel_label=channel.channel_label,
                        wavelength_id=channel.wavelength_id,
                        color=channel.color.to_hexstr(),
                    )
                )
            updated_details.channels = _updated_channels
        if self.pixel_info is not None:
            updated_details.xy_pixel_size = self.pixel_info.xy_pixel_size
            updated_details.z_spacing = self.pixel_info.z_spacing
            updated_details.t_spacing = self.pixel_info.t_spacing
        axes = self.to_axes_list()
        if axes is not None:
            updated_details.axes = axes
        if self.data_type != DataTypeEnum.AUTODETECT:
            updated_details.data_type = self.data_type
        if self.condition_table_path is not None:
            updated_details.condition_table_path = self.condition_table_path
        return updated_details
axes: str | None = None class-attribute instance-attribute

Axes of the image data, e.g. czyx. If left empty, the axes parsed from the raw metadata are used.

channels: list[ChannelInfoUI] | None = None class-attribute instance-attribute

Names, wavelengths, and display colors of the channels. If left empty, the channel information parsed from the raw metadata is used.

condition_table_path: str | None = None class-attribute instance-attribute

Optional path to a condition table CSV file to store in the plate metadata.

data_type: DataTypeEnum = Field(default=(DataTypeEnum.AUTODETECT), title='Data Type') class-attribute instance-attribute

Pixel data type of the output image. autodetect infers it from the input images.

filters: list[ImplementedFilters] = Field(default_factory=list) class-attribute instance-attribute

Filters selecting which tiles of the acquisition are converted.

pixel_info: PixelSizeModel | None = Field(default=None, title='Pixel Size Information') class-attribute instance-attribute

Override the pixel size and the Z/time spacing of the images. If left empty, the values parsed from the raw metadata are used.

stage_orientation: StageOrientation = Field(default_factory=StageOrientation, title='Stage Orientation') class-attribute instance-attribute

Corrections for the orientation of the microscope stage relative to the image axes.

to_axes_list() -> list[CANONICAL_AXES_TYPE] | None

Convert axes string to list of axes.

Source code in ome_zarr_converters_tools/fractal/_models.py 
164
165
166
167
168
169
170
171
172
173
def to_axes_list(self) -> list[CANONICAL_AXES_TYPE] | None:
    """Convert axes string to list of axes."""
    if self.axes is None:
        return None
    _axes = []
    for ax in self.axes:
        if ax not in canonical_axes:
            raise ValueError(f"Invalid axis '{ax}' in axes string.")
        _axes.append(ax)
    return _axes  # type: ignore
update_acquisition_details(acquisition_details: AcquisitionDetails) -> AcquisitionDetails

Update AcquisitionDetails model with options from this model.

Parameters:

Returns:

Source code in ome_zarr_converters_tools/fractal/_models.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def update_acquisition_details(
    self,
    acquisition_details: AcquisitionDetails,
) -> AcquisitionDetails:
    """Update AcquisitionDetails model with options from this model.

    Args:
        acquisition_details: AcquisitionDetails model to update.

    Returns:
        Updated AcquisitionDetails model.

    """
    updated_details = acquisition_details.model_copy()
    if self.channels is not None:
        _updated_channels = []
        for channel in self.channels:
            _updated_channels.append(
                ChannelInfo(
                    channel_label=channel.channel_label,
                    wavelength_id=channel.wavelength_id,
                    color=channel.color.to_hexstr(),
                )
            )
        updated_details.channels = _updated_channels
    if self.pixel_info is not None:
        updated_details.xy_pixel_size = self.pixel_info.xy_pixel_size
        updated_details.z_spacing = self.pixel_info.z_spacing
        updated_details.t_spacing = self.pixel_info.t_spacing
    axes = self.to_axes_list()
    if axes is not None:
        updated_details.axes = axes
    if self.data_type != DataTypeEnum.AUTODETECT:
        updated_details.data_type = self.data_type
    if self.condition_table_path is not None:
        updated_details.condition_table_path = self.condition_table_path
    return updated_details

ChannelInfoUI

Bases: UserFacingModel

Set the name, wavelength, and display color of a channel.

Source code in ome_zarr_converters_tools/fractal/_models.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
class ChannelInfoUI(UserFacingModel):
    """Set the name, wavelength, and display color of a channel."""

    channel_label: str
    """Name of the channel, e.g. `DAPI` or `GFP`."""
    wavelength_id: str | None = Field(default=None, title="Wavelength ID")
    """
    The wavelength ID of the channel.
    Some tasks can use it instead of the channel name, e.g. to apply
    illumination correction per wavelength in multiplexed acquisitions.
    """
    color: ColorMenu = ColorMenu.Auto
    """Display color of the channel, e.g. for visualization purposes."""
channel_label: str instance-attribute

Name of the channel, e.g. DAPI or GFP.

color: ColorMenu = ColorMenu.Auto class-attribute instance-attribute

Display color of the channel, e.g. for visualization purposes.

wavelength_id: str | None = Field(default=None, title='Wavelength ID') class-attribute instance-attribute

The wavelength ID of the channel. Some tasks can use it instead of the channel name, e.g. to apply illumination correction per wavelength in multiplexed acquisitions.

ConvertParallelInitArgs

Bases: UserFacingModel

Internal data handed from the init phase to compute; filled automatically.

Source code in ome_zarr_converters_tools/fractal/_models.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ConvertParallelInitArgs(UserFacingModel):
    """Internal data handed from the init phase to compute; *filled automatically*."""

    tiled_image_json_dump_url: str | None = None
    """Location of the temporary file describing the image to convert."""
    tiled_image_json_str: str | None = None
    """Inline description of the image to convert (used instead of a
    temporary file for small conversions)."""
    converter_options: ConverterOptions
    """Converter options forwarded from the init phase."""
    overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE
    """Overwrite mode forwarded from the init phase."""

    @model_validator(mode="after")
    def _validate_exactly_one_source(self) -> "ConvertParallelInitArgs":
        if (self.tiled_image_json_dump_url is None) == (
            self.tiled_image_json_str is None
        ):
            raise ValueError(
                "Exactly one of tiled_image_json_dump_url or "
                "tiled_image_json_str must be set."
            )
        return self
converter_options: ConverterOptions instance-attribute

Converter options forwarded from the init phase.

overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE class-attribute instance-attribute

Overwrite mode forwarded from the init phase.

tiled_image_json_dump_url: str | None = None class-attribute instance-attribute

Location of the temporary file describing the image to convert.

tiled_image_json_str: str | None = None class-attribute instance-attribute

Inline description of the image to convert (used instead of a temporary file for small conversions).

MultiprocessingRunner

Bases: BaseModel

Runner for executing compound tasks using multiprocessing.

Source code in ome_zarr_converters_tools/fractal/_compound_task_wrapper.py 
26
27
28
29
30
class MultiprocessingRunner(BaseModel):
    """Runner for executing compound tasks using multiprocessing."""

    mode: Literal["Multiprocessing"] = "Multiprocessing"
    num_processes: Annotated[int, Field(gt=0)] = 4

PixelSizeModel

Bases: UserFacingModel

Override the pixel size and the Z/time spacing of the images.

Source code in ome_zarr_converters_tools/fractal/_models.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class PixelSizeModel(UserFacingModel):
    """Override the pixel size and the Z/time spacing of the images."""

    xy_pixel_size: float = Field(title="XY Pixel Size")
    """
    XY pixel size in micrometers.
    """
    z_spacing: float = Field(title="Z Spacing")
    """
    Z spacing in micrometers.
    """
    t_spacing: float = Field(title="Time Spacing")
    """
    Time spacing in seconds.
    """
t_spacing: float = Field(title='Time Spacing') class-attribute instance-attribute

Time spacing in seconds.

xy_pixel_size: float = Field(title='XY Pixel Size') class-attribute instance-attribute

XY pixel size in micrometers.

z_spacing: float = Field(title='Z Spacing') class-attribute instance-attribute

Z spacing in micrometers.

SequentialRunner

Bases: BaseModel

Runner for executing compound tasks sequentially.

Source code in ome_zarr_converters_tools/fractal/_compound_task_wrapper.py 
13
14
15
16
class SequentialRunner(BaseModel):
    """Runner for executing compound tasks sequentially."""

    mode: Literal["Synchronous"] = "Synchronous"

ThreadedRunner

Bases: BaseModel

Runner for executing compound tasks using threads.

Source code in ome_zarr_converters_tools/fractal/_compound_task_wrapper.py 
19
20
21
22
23
class ThreadedRunner(BaseModel):
    """Runner for executing compound tasks using threads."""

    mode: Literal["Threaded"] = "Threaded"
    num_threads: Annotated[int, Field(gt=0)] = 4

cleanup_if_exists(temp_json_url: str)

Clean up the temporary JSON directory if it exists.

If cleaning up is not possible, log an error message, but do not raise.

Parameters:

  • temp_json_url (str) –

    The URL to the temporary JSON directory.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def cleanup_if_exists(temp_json_url: str):
    """Clean up the temporary JSON directory if it exists.

    If cleaning up is not possible, log an error message, but do not raise.

    Args:
        temp_json_url: The URL to the temporary JSON directory.
    """
    fs = filesystem_for_url(temp_json_url, error_msg_prefix="Cleanup")

    if not fs.exists(temp_json_url):
        return
    if not fs.isdir(temp_json_url):
        raise ValueError(
            f"Expected a directory for a cleanup, but got a file: {temp_json_url}"
        )
    try:
        # Limit to depth 1: temp JSON store is flat (files only, no subdirs)
        fs.rm(temp_json_url, recursive=True, maxdepth=1)
    except Exception as e:
        logger.error(
            f"An error occurred while cleaning up the temporary JSON directory: {e}. "
            f"You can safely remove the store: {temp_json_url}"
        )

converters_tools_models(base: str = 'ome_zarr_converters_tools') -> list[tuple[str, str, str]]

Get all input models for Fractal tasks API.

Returns:

  • list[tuple[str, str, str]] –

    List of input models.

Source code in ome_zarr_converters_tools/fractal/_models.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def converters_tools_models(
    base: str = "ome_zarr_converters_tools",
) -> list[tuple[str, str, str]]:
    """Get all input models for Fractal tasks API.

    Returns:
        List of input models.
    """
    return [
        (
            base,
            "fractal/_models.py",
            "AcquisitionOptions",
        ),
        (
            base,
            "pipelines/_filters.py",
            "WellFilter",
        ),
        (
            base,
            "pipelines/_filters.py",
            "RegexFilter",
        ),
        (
            base,
            "models/_converter_options.py",
            "ConverterOptions",
        ),
        (
            base,
            "models/_acquisition.py",
            "StageOrientation",
        ),
        (
            base,
            "models/_converter_options.py",
            "StagePositionCorrections",
        ),
        (
            base,
            "models/_converter_options.py",
            "OmeZarrOptions",
        ),
        (
            base,
            "models/_runtime_settings.py",
            "TempJsonOptions",
        ),
        (
            base,
            "models/_converter_options.py",
            "FovBasedChunking",
        ),
        (
            base,
            "models/_converter_options.py",
            "FixedSizeChunking",
        ),
        (
            base,
            "models/_acquisition.py",
            "ChannelInfo",
        ),
    ]

dump_json_str(temp_json_url: str, json_str: str) -> str

Write a pre-serialized JSON string to a unique file in temp_json_url.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
22
23
24
25
26
27
28
29
30
31
def dump_json_str(temp_json_url: str, json_str: str) -> str:
    """Write a pre-serialized JSON string to a unique file in temp_json_url."""
    fs = filesystem_for_url(temp_json_url, error_msg_prefix="Dumping JSON")
    fs.makedirs(temp_json_url, exist_ok=True)
    unique_json_filename = f"{uuid4()}.json"
    tile_json_name = join_url_paths(temp_json_url, unique_json_filename)
    with fs.open(tile_json_name, "w") as f:
        f.write(json_str)
    logger.debug(f"JSON file created: {tile_json_name}")
    return tile_json_name

dump_to_json(temp_json_url: str, tiled_image: TiledImage) -> str

Create a JSON file for the tiled image.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
34
35
36
37
38
def dump_to_json(temp_json_url: str, tiled_image: TiledImage) -> str:
    """Create a JSON file for the tiled image."""
    return dump_json_str(
        temp_json_url=temp_json_url, json_str=tiled_image.model_dump_json()
    )

exec_compound_task(init_task_fn: Callable[..., dict], compute_task_fn: Callable[..., ImageListUpdateDict], init_task_kwargs: dict, compute_task_kwargs: dict | None = None, runner: RunnerType | None = None) -> list[ImageListUpdateDict]

Execute a Fractal compound task.

Runs init_task_fn once to obtain a parallelization_list, then calls compute_task_fn for each item, merging compute_task_kwargs into each call. Results are returned in the same order as parallelization_list.

Parameters:

  • init_task_fn (Callable[..., dict]) –

    Returns a dict with a "parallelization_list" key.

  • compute_task_fn (Callable[..., ImageListUpdateDict]) –

    Invoked once per parallelization item. When using MultiprocessingRunner, this must be a picklable module-level function.

  • init_task_kwargs (dict) –

    Keyword arguments forwarded to init_task_fn.

  • compute_task_kwargs (dict | None, default: None ) –

    Additional kwargs merged into each compute call.

  • runner (RunnerType | None, default: None ) –

    Execution strategy. Defaults to sequential when None.

Returns:

  • list[ImageListUpdateDict] –

    List of ImageListUpdateDict in parallelization_list order.

Source code in ome_zarr_converters_tools/fractal/_compound_task_wrapper.py 
 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
def exec_compound_task(
    init_task_fn: Callable[..., dict],
    compute_task_fn: Callable[..., ImageListUpdateDict],
    init_task_kwargs: dict,
    compute_task_kwargs: dict | None = None,
    runner: RunnerType | None = None,
) -> list[ImageListUpdateDict]:
    """Execute a Fractal compound task.

    Runs `init_task_fn` once to obtain a `parallelization_list`, then calls
    `compute_task_fn` for each item, merging `compute_task_kwargs` into each
    call. Results are returned in the same order as `parallelization_list`.

    Args:
        init_task_fn: Returns a dict with a `"parallelization_list"` key.
        compute_task_fn: Invoked once per parallelization item. When using
            `MultiprocessingRunner`, this must be a picklable module-level
            function.
        init_task_kwargs: Keyword arguments forwarded to `init_task_fn`.
        compute_task_kwargs: Additional kwargs merged into each compute call.
        runner: Execution strategy. Defaults to sequential when `None`.

    Returns:
        List of `ImageListUpdateDict` in `parallelization_list` order.
    """
    match runner:
        case None | SequentialRunner():
            return _exec_compound_task_synchronous(
                init_task_fn=init_task_fn,
                compute_task_fn=compute_task_fn,
                init_task_kwargs=init_task_kwargs,
                compute_task_kwargs=compute_task_kwargs,
            )
        case ThreadedRunner(num_threads=n):
            return _exec_compound_task_threaded(
                init_task_fn=init_task_fn,
                compute_task_fn=compute_task_fn,
                init_task_kwargs=init_task_kwargs,
                compute_task_kwargs=compute_task_kwargs,
                num_threads=n,
            )
        case MultiprocessingRunner(num_processes=n):
            return _exec_compound_task_multiprocessing(
                init_task_fn=init_task_fn,
                compute_task_fn=compute_task_fn,
                init_task_kwargs=init_task_kwargs,
                compute_task_kwargs=compute_task_kwargs,
                num_processes=n,
            )
        case _:
            raise ValueError(
                f"Unsupported runner {runner!r}. Use one of SequentialRunner, "
                "ThreadedRunner, MultiprocessingRunner, or None for sequential "
                "execution."
            )

generic_compute_task(*, zarr_url: str, init_args: ConvertParallelInitArgs | dict, collection_type: type[CollectionInterfaceType], image_loader_type: type[ImageLoaderInterfaceType], resource: Any | None = None) -> ImageListUpdateDict

Initialize the task to convert a LIF plate to OME-Zarr.

Parameters:

  • zarr_url (str) –

    URL to the OME-Zarr file.

  • init_args (ConvertParallelInitArgs | dict) –

    Arguments from the initialization task. Accepts either a ConvertParallelInitArgs instance or a plain dict (as produced by Fractal's orchestration layer).

  • collection_type (type[CollectionInterfaceType]) –

    The collection type to use when loading the TiledImage.

  • image_loader_type (type[ImageLoaderInterfaceType]) –

    The image loader type to use when loading the TiledImage.

  • resource (Any | None, default: None ) –

    The resource to associate with the context model.

Returns:

  • ImageListUpdateDict –

    The Fractal image-list update entry for the converted image.

Source code in ome_zarr_converters_tools/fractal/_compute_task.py 
 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
def generic_compute_task(
    *,
    # Fractal parameters
    zarr_url: str,
    init_args: ConvertParallelInitArgs | dict,
    collection_type: type[CollectionInterfaceType],
    image_loader_type: type[ImageLoaderInterfaceType],
    resource: Any | None = None,
) -> ImageListUpdateDict:
    """Initialize the task to convert a LIF plate to OME-Zarr.

    Args:
        zarr_url: URL to the OME-Zarr file.
        init_args: Arguments from the initialization task. Accepts either a
            `ConvertParallelInitArgs` instance or a plain dict (as produced by
            Fractal's orchestration layer).
        collection_type: The collection type to use when loading the `TiledImage`.
        image_loader_type: The image loader type to use when loading the `TiledImage`.
        resource: The resource to associate with the context model.

    Returns:
        The Fractal image-list update entry for the converted image.
    """
    logger.info(f"Starting conversion for Zarr URL: {zarr_url}")
    parsed_args = ConvertParallelInitArgs.model_validate(init_args)

    if parsed_args.tiled_image_json_str is not None:
        tiled_image_loaded = tiled_image_from_json_str(
            json_str=parsed_args.tiled_image_json_str,
            collection_type=collection_type,
            image_loader_type=image_loader_type,
        )
    else:
        json_url = parsed_args.tiled_image_json_dump_url
        assert json_url is not None
        # `tiled_image_from_json` already retries with backoff (see
        # CONVERTERS_TOOLS_NUM_RETRIES) and raises FileNotFoundError when the
        # file never appears; no outer retry loop is needed.
        tiled_image_loaded = tiled_image_from_json(
            tiled_image_json_dump_url=json_url,
            collection_type=collection_type,
            image_loader_type=image_loader_type,
        )
        logger.info(f"Successfully loaded JSON file: {json_url}")

    registration_pipeline = build_default_registration_pipeline(
        alignment_corrections=parsed_args.converter_options.stage_position_corrections,
        tiling_strategy=parsed_args.converter_options.grouping.tiling_for_registration(),
    )
    ome_zarr = tiled_image_creation_pipeline(
        zarr_url=zarr_url,
        tiled_image=tiled_image_loaded,
        registration_pipeline=registration_pipeline,
        converter_options=parsed_args.converter_options,
        writer_mode=parsed_args.converter_options.writer_mode,
        overwrite_mode=parsed_args.overwrite_mode,
        resource=resource,
    )
    if parsed_args.tiled_image_json_dump_url is not None:
        remove_json(parsed_args.tiled_image_json_dump_url)
    logger.info("Conversion complete")
    return _build_image_list_update(
        zarr_url=zarr_url,
        ome_zarr=ome_zarr,
        collection=tiled_image_loaded.collection,
        attributes=tiled_image_loaded.attributes,
    )

remove_json(tiled_image_json_dump_url: str)

Clean up the JSON file and the directory if it is empty.

Parameters:

  • tiled_image_json_dump_url (str) –

    The URL to the json file.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
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
def remove_json(
    tiled_image_json_dump_url: str,
):
    """Clean up the JSON file and the directory if it is empty.

    Args:
        tiled_image_json_dump_url: The URL to the json file.
    """
    fs = filesystem_for_url(
        tiled_image_json_dump_url, error_msg_prefix="Cleaning up JSON"
    )

    try:
        fs.rm(tiled_image_json_dump_url)
        parent_dir = parent_url(tiled_image_json_dump_url)
        try:
            # no-op if non-empty; avoids a listdir on potentially large directories
            # for distributed filesystems like CephFS where listdir can be expensive
            # also avoids a race condition where another process has already removed
            # the file and directory
            fs.rmdir(parent_dir)
        except OSError:
            pass
    except Exception as e:
        logger.error(
            f"An error occurred while cleaning up the JSON file: {e}. "
            f"You can safely remove the store: {tiled_image_json_dump_url}"
        )

setup_images_for_conversion(tiled_images: list[TiledImage], *, zarr_dir: str, collection_type: str, converter_options: ConverterOptions, overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE, ngff_version: NgffVersions = DefaultNgffVersion) -> list[dict]

Setup the OME-Zarr collection from converted tiled images.

This function run all the necessary steps to setup before parallel conversion. - Build the OME-Zarr collection structure. - Build the parallelization list (used by the fractal compute task).

Parameters:

  • tiled_images (list[TiledImage]) –

    List of tiled images that have been converted.

  • zarr_dir (str) –

    The base directory for the zarr data.

  • collection_type (str) –

    The type of collection to set up.

  • converter_options (ConverterOptions) –

    The converter options to use during conversion.

  • overwrite_mode (OverwriteMode, default: NO_OVERWRITE ) –

    The overwrite mode to use when writing the data.

  • ngff_version (NgffVersions, default: DefaultNgffVersion ) –

    The NGFF version to use when setting up the collection.

Source code in ome_zarr_converters_tools/fractal/_init_task.py 
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
def setup_images_for_conversion(
    tiled_images: list[TiledImage],
    *,
    zarr_dir: str,
    collection_type: str,
    converter_options: ConverterOptions,
    overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE,
    ngff_version: NgffVersions = DefaultNgffVersion,
) -> list[dict]:
    """Setup the OME-Zarr collection from converted tiled images.

    This function run all the necessary steps to setup before parallel conversion.
        - Build the OME-Zarr collection structure.
        - Build the parallelization list (used by the fractal compute task).

    Args:
        tiled_images: List of tiled images that have been converted.
        zarr_dir: The base directory for the zarr data.
        collection_type: The type of collection to set up.
        converter_options: The converter options to use during conversion.
        overwrite_mode: The overwrite mode to use when writing the data.
        ngff_version: The NGFF version to use when setting up the collection.
    """
    _check_path_collisions(tiled_images)
    setup_ome_zarr_collection(
        tiled_images=tiled_images,
        collection_type=collection_type,
        zarr_dir=zarr_dir,
        ngff_version=ngff_version,
        overwrite_mode=overwrite_mode,
    )
    return build_parallelization_list(
        zarr_dir=zarr_dir,
        tiled_images=tiled_images,
        converter_options=converter_options,
        overwrite_mode=overwrite_mode,
    )

tiled_image_from_json(tiled_image_json_dump_url: str, collection_type: type[CollectionInterfaceType], image_loader_type: type[ImageLoaderInterfaceType]) -> TiledImage

Load the json TiledImage object.

Since TiledImage is a generic model, we need to specify the concrete types when loading it from json otherwise pydantic cannot infer them.

Parameters:

  • tiled_image_json_dump_url (str) –

    The URL to the json file.

  • collection_type (type[CollectionInterfaceType]) –

    The concrete collection type of the TiledImage.

  • image_loader_type (type[ImageLoaderInterfaceType]) –

    The concrete image loader type of the TiledImage.

Returns:

Source code in ome_zarr_converters_tools/fractal/_json_utils.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
def tiled_image_from_json(
    tiled_image_json_dump_url: str,
    collection_type: type[CollectionInterfaceType],
    image_loader_type: type[ImageLoaderInterfaceType],
) -> TiledImage:
    """Load the json TiledImage object.

    Since TiledImage is a generic model, we need to specify the concrete types
    when loading it from json otherwise pydantic cannot infer them.

    Args:
        tiled_image_json_dump_url: The URL to the json file.
        collection_type: The concrete collection type of the `TiledImage`.
        image_loader_type: The concrete image loader type of the `TiledImage`.

    Returns:
        The loaded `TiledImage` object.
    """
    num_retries = int(os.getenv("CONVERTERS_TOOLS_NUM_RETRIES", 5))

    if num_retries < 1:
        raise ValueError("NUM_RETRIES must be greater than 0")

    for t in range(num_retries):
        try:
            fs = filesystem_for_url(
                tiled_image_json_dump_url, error_msg_prefix="Loading JSON"
            )
            with fs.open(tiled_image_json_dump_url, "r") as f:
                # Concretely specify the types to load the generic TiledImage
                tiled_image = TiledImage[
                    collection_type, image_loader_type  # ty:ignore[invalid-type-form]
                ].model_validate_json(f.read())

            return tiled_image

        except FileNotFoundError:
            if t == num_retries - 1:
                break  # Last attempt failed; don't sleep before raising.
            logger.error(
                f"JSON file does not exist: {tiled_image_json_dump_url}, retrying..."
            )
            time.sleep(2 ** (t + 1))

    raise FileNotFoundError(
        f"JSON file does not exist after {num_retries} "
        f"retries: {tiled_image_json_dump_url}"
    )

tiled_image_from_json_str(json_str: str, collection_type: type[CollectionInterfaceType], image_loader_type: type[ImageLoaderInterfaceType]) -> TiledImage

Deserialize a TiledImage from a JSON string (no filesystem I/O).

Parameters:

  • json_str (str) –

    The JSON string to deserialize.

  • collection_type (type[CollectionInterfaceType]) –

    The concrete collection type of the TiledImage.

  • image_loader_type (type[ImageLoaderInterfaceType]) –

    The concrete image loader type of the TiledImage.

Returns:

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def tiled_image_from_json_str(
    json_str: str,
    collection_type: type[CollectionInterfaceType],
    image_loader_type: type[ImageLoaderInterfaceType],
) -> TiledImage:
    """Deserialize a TiledImage from a JSON string (no filesystem I/O).

    Args:
        json_str: The JSON string to deserialize.
        collection_type: The concrete collection type of the TiledImage.
        image_loader_type: The concrete image loader type of the TiledImage.

    Returns:
        The loaded `TiledImage` object.
    """
    return TiledImage[
        collection_type, image_loader_type  # ty:ignore[invalid-type-form]
    ].model_validate_json(json_str)

Testing

Snapshot-testing helpers shipped for downstream converter test suites. Load the pytest plugin from a consumer's tests/conftest.py with pytest_plugins = ["ome_zarr_converters_tools.testing.plugin"]; it provides the --update-snapshots / --extended options, the extended marker, and the update_snapshots fixture.

ome_zarr_converters_tools.testing

Snapshot testing helpers for OME-Zarr converters.

The heavy _snapshot module (numpy/ngio/pydantic) is imported lazily so that loading the pytest plugin (ome_zarr_converters_tools.testing.plugin, wired via pytest_plugins in each consumer's conftest, not a pytest11 entry point) does not pull it in — this keeps plugin load cheap and lets coverage instrument _snapshot when tests first import it.

FingerprintModel

Bases: BaseModel

Compact pixel-content signature for a single ROI.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class FingerprintModel(BaseModel):
    """Compact pixel-content signature for a single ROI."""

    mean: float
    std: float
    min: float
    max: float
    hash: str

    @classmethod
    def from_array(cls, arr: np.ndarray, decimals: int = _FINGERPRINT_DECIMALS):
        """Build a fingerprint from a pixel array.

        Args:
            arr: The pixel data read back from a ROI.
            decimals: Decimals to round the stored stats to, and to round the
                array to before hashing.
        """
        rounded = np.round(arr, decimals)
        return cls(
            mean=round(float(np.mean(arr)), decimals),
            std=round(float(np.std(arr)), decimals),
            min=round(float(np.min(arr)), decimals),
            max=round(float(np.max(arr)), decimals),
            hash=hashlib.sha256(rounded.tobytes()).hexdigest(),
        )
from_array(arr: np.ndarray, decimals: int = _FINGERPRINT_DECIMALS) classmethod

Build a fingerprint from a pixel array.

Parameters:

  • arr (ndarray) –

    The pixel data read back from a ROI.

  • decimals (int, default: _FINGERPRINT_DECIMALS ) –

    Decimals to round the stored stats to, and to round the array to before hashing.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@classmethod
def from_array(cls, arr: np.ndarray, decimals: int = _FINGERPRINT_DECIMALS):
    """Build a fingerprint from a pixel array.

    Args:
        arr: The pixel data read back from a ROI.
        decimals: Decimals to round the stored stats to, and to round the
            array to before hashing.
    """
    rounded = np.round(arr, decimals)
    return cls(
        mean=round(float(np.mean(arr)), decimals),
        std=round(float(np.std(arr)), decimals),
        min=round(float(np.min(arr)), decimals),
        max=round(float(np.max(arr)), decimals),
        hash=hashlib.sha256(rounded.tobytes()).hexdigest(),
    )

ImageAssertionModel

Bases: BaseModel

Expected metadata and per-table ROI fingerprints for one image.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
125
126
127
128
129
130
131
132
133
134
135
class ImageAssertionModel(BaseModel):
    """Expected metadata and per-table ROI fingerprints for one image."""

    axes: tuple[str, ...]
    shape: tuple[int, ...]
    pixelsize: tuple[float, ...]
    channel_labels: list[str] = Field(default_factory=list)
    wavelength_ids: list[str | None] = Field(default_factory=list)
    types: dict[str, bool] = Field(default_factory=dict)
    attributes: dict[str, str | int | float] = Field(default_factory=dict)
    tables: dict[str, TableAssertionModel | None] = Field(default_factory=dict)

MultiPlateAssertionModel

Bases: BaseModel

Snapshot for HCS plate output: one or more plates keyed by zarr name.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
179
180
181
182
183
184
class MultiPlateAssertionModel(BaseModel):
    """Snapshot for HCS plate output: one or more plates keyed by zarr name."""

    # Informational only; recorded at generation time, never compared.
    versions: dict[str, str | None] = Field(default_factory=dict)
    plates: dict[str, PlateAssertionModel]

MultiSingleImageAssertionModel

Bases: BaseModel

Snapshot for single-image output: images keyed by zarr name.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
187
188
189
190
191
192
class MultiSingleImageAssertionModel(BaseModel):
    """Snapshot for single-image output: images keyed by zarr name."""

    # Informational only; recorded at generation time, never compared.
    versions: dict[str, str | None] = Field(default_factory=dict)
    images: dict[str, ImageAssertionModel]

PlateAssertionModel

Bases: BaseModel

Expected wells and images for a single plate.

A top-level images_common mapping may be supplied on input to factor out assertions shared by every image; it is deep-merged into each image and not retained on the model.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class PlateAssertionModel(BaseModel):
    """Expected wells and images for a single plate.

    A top-level `images_common` mapping may be supplied on input to factor out
    assertions shared by every image; it is deep-merged into each image and not
    retained on the model.
    """

    wells: list[str]
    images: dict[str, ImageAssertionModel]

    @model_validator(mode="before")
    @classmethod
    def validate_images(cls, values):
        """Deep-merge `images_common` into every image entry.

        `images_common` provides shared defaults; per-image values override them.
        """
        common_assertions = values.pop("images_common", {})
        images = values.get("images", {})
        updated_image_assertions = {}
        for image_path, image_assertions in images.items():
            # Common first (base), image second (overrides), for every image —
            # including keys the image does not itself specify.
            updated_image_assertions[image_path] = _deep_merge(
                common_assertions, image_assertions
            )
        values["images"] = updated_image_assertions
        return values
validate_images(values) classmethod

Deep-merge images_common into every image entry.

images_common provides shared defaults; per-image values override them.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@model_validator(mode="before")
@classmethod
def validate_images(cls, values):
    """Deep-merge `images_common` into every image entry.

    `images_common` provides shared defaults; per-image values override them.
    """
    common_assertions = values.pop("images_common", {})
    images = values.get("images", {})
    updated_image_assertions = {}
    for image_path, image_assertions in images.items():
        # Common first (base), image second (overrides), for every image —
        # including keys the image does not itself specify.
        updated_image_assertions[image_path] = _deep_merge(
            common_assertions, image_assertions
        )
    values["images"] = updated_image_assertions
    return values

RoiAssertionModel

Bases: BaseModel

Expected fingerprint and geometry for a single ROI.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
111
112
113
114
115
116
class RoiAssertionModel(BaseModel):
    """Expected fingerprint and geometry for a single ROI."""

    slice_repr: str
    finger_print: FingerprintModel
    yx_origin: list[float] | None = None

TableAssertionModel

Bases: BaseModel

Expected ROIs for a single table.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
119
120
121
122
class TableAssertionModel(BaseModel):
    """Expected ROIs for a single table."""

    rois: dict[str, RoiAssertionModel] = Field(default_factory=dict)

__getattr__(name: str)

Lazily resolve public names from _snapshot on first access.

Source code in ome_zarr_converters_tools/testing/__init__.py 
41
42
43
44
45
46
def __getattr__(name: str):
    """Lazily resolve public names from `_snapshot` on first access."""
    if name in __all__:
        module = importlib.import_module("ome_zarr_converters_tools.testing._snapshot")
        return getattr(module, name)
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

build_snapshot(*, zarr_dir: Path, image_list_updates: list[dict], output_type: Literal['plate', 'single_image'] = 'plate') -> MultiPlateAssertionModel | MultiSingleImageAssertionModel

Build a snapshot model from converted OME-Zarr output.

This is the single source of truth for both snapshot generation and validation: the returned model is either written to disk or compared against the on-disk expectation. The returned model also carries an informational versions block (dependency versions at generation time), which is never compared.

Parameters:

  • zarr_dir (Path) –

    Directory the converter wrote its output into.

  • image_list_updates (list[dict]) –

    The list returned by the converter's init task.

  • output_type (Literal['plate', 'single_image'], default: 'plate' ) –

    "plate" for HCS plate output, "single_image" for individual containers at the top level of zarr_dir.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
def build_snapshot(
    *,
    zarr_dir: Path,
    image_list_updates: list[dict],
    output_type: Literal["plate", "single_image"] = "plate",
) -> MultiPlateAssertionModel | MultiSingleImageAssertionModel:
    """Build a snapshot model from converted OME-Zarr output.

    This is the single source of truth for both snapshot generation and
    validation: the returned model is either written to disk or compared against
    the on-disk expectation. The returned model also carries an informational
    `versions` block (dependency versions at generation time), which is never
    compared.

    Args:
        zarr_dir: Directory the converter wrote its output into.
        image_list_updates: The list returned by the converter's init task.
        output_type: `"plate"` for HCS plate output, `"single_image"` for
            individual containers at the top level of `zarr_dir`.
    """
    if output_type == "plate":
        model: MultiPlateAssertionModel | MultiSingleImageAssertionModel = (
            _build_plate_snapshot(
                zarr_dir=zarr_dir, image_list_updates=image_list_updates
            )
        )
    else:
        model = _build_single_image_snapshot(
            zarr_dir=zarr_dir, image_list_updates=image_list_updates
        )
    model.versions = _collect_versions()
    return model

compare_snapshots(expected: MultiPlateAssertionModel | MultiSingleImageAssertionModel, actual: MultiPlateAssertionModel | MultiSingleImageAssertionModel) -> list[str]

Compare an expected snapshot against a freshly built one.

Returns a list of human-readable, fully-pathed difference messages (empty when the snapshots match). Floating-point stats use a tolerance; the sha256 pixel hash and structural fields are compared exactly.

The versions block is deliberately not compared: it is informational only (a dependency version drift must never fail a snapshot test).

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
def compare_snapshots(
    expected: MultiPlateAssertionModel | MultiSingleImageAssertionModel,
    actual: MultiPlateAssertionModel | MultiSingleImageAssertionModel,
) -> list[str]:
    """Compare an expected snapshot against a freshly built one.

    Returns a list of human-readable, fully-pathed difference messages (empty
    when the snapshots match). Floating-point stats use a tolerance; the sha256
    pixel hash and structural fields are compared exactly.

    The `versions` block is deliberately not compared: it is informational only
    (a dependency version drift must never fail a snapshot test).
    """
    diffs: list[str] = []
    if isinstance(expected, MultiPlateAssertionModel) and isinstance(
        actual, MultiPlateAssertionModel
    ):
        exp_plates, act_plates = set(expected.plates), set(actual.plates)
        if exp_plates != act_plates:
            diffs.append(
                f"plates: expected {sorted(exp_plates)} != actual {sorted(act_plates)}"
            )
        for pname in sorted(exp_plates & act_plates):
            exp_plate, act_plate = expected.plates[pname], actual.plates[pname]
            if set(exp_plate.wells) != set(act_plate.wells):
                diffs.append(
                    f"plates['{pname}'].wells: expected {sorted(exp_plate.wells)} "
                    f"!= actual {sorted(act_plate.wells)}"
                )
            exp_imgs, act_imgs = set(exp_plate.images), set(act_plate.images)
            if exp_imgs != act_imgs:
                diffs.append(
                    f"plates['{pname}'].images: expected {sorted(exp_imgs)} "
                    f"!= actual {sorted(act_imgs)}"
                )
            for iname in sorted(exp_imgs & act_imgs):
                _compare_image(
                    f"plates['{pname}'].images['{iname}']",
                    exp_plate.images[iname],
                    act_plate.images[iname],
                    diffs,
                )
    elif isinstance(expected, MultiSingleImageAssertionModel) and isinstance(
        actual, MultiSingleImageAssertionModel
    ):
        exp_imgs, act_imgs = set(expected.images), set(actual.images)
        if exp_imgs != act_imgs:
            diffs.append(
                f"images: expected {sorted(exp_imgs)} != actual {sorted(act_imgs)}"
            )
        for iname in sorted(exp_imgs & act_imgs):
            _compare_image(
                f"images['{iname}']",
                expected.images[iname],
                actual.images[iname],
                diffs,
            )
    else:
        diffs.append(
            f"snapshot type mismatch: expected {type(expected).__name__} "
            f"!= actual {type(actual).__name__}"
        )
    return diffs

run_converter_test(*, tmp_path: Path, api_fn: Callable, api_kwargs: dict, snapshot_path: Path, update_snapshots: bool, converter_options: ConverterOptions | None = None, output_type: Literal['plate', 'single_image'] = 'plate') -> None

Run a converter end-to-end and check it against a snapshot.

Runs api_fn into a temporary (or snapshot-adjacent) zarr directory and builds a snapshot from the output. With update_snapshots the snapshot is written to snapshot_path as JSON; otherwise it is compared against the on-disk snapshot and a single AssertionError listing every difference is raised on mismatch.

Parameters:

  • tmp_path (Path) –

    Pytest tmp_path for zarr output (validation mode).

  • api_fn (Callable) –

    The high-level converter API function.

  • api_kwargs (dict) –

    Keyword arguments for api_fn (e.g. acquisitions).

  • snapshot_path (Path) –

    Path to the snapshot JSON file.

  • update_snapshots (bool) –

    If True, (re)generate the snapshot instead of checking.

  • converter_options (ConverterOptions | None, default: None ) –

    Options controlling the conversion; passed to api_fn only when provided.

  • output_type (Literal['plate', 'single_image'], default: 'plate' ) –

    "plate" for HCS plate output, "single_image" for individual zarr containers at the top level of the output directory.

Source code in ome_zarr_converters_tools/testing/_snapshot.py 
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
def run_converter_test(
    *,
    tmp_path: Path,
    api_fn: Callable,
    api_kwargs: dict,
    snapshot_path: Path,
    update_snapshots: bool,
    converter_options: ConverterOptions | None = None,
    output_type: Literal["plate", "single_image"] = "plate",
) -> None:
    """Run a converter end-to-end and check it against a snapshot.

    Runs `api_fn` into a temporary (or snapshot-adjacent) zarr directory and
    builds a snapshot from the output. With `update_snapshots` the snapshot is
    written to `snapshot_path` as JSON; otherwise it is compared against the
    on-disk snapshot and a single `AssertionError` listing every difference is
    raised on mismatch.

    Args:
        tmp_path: Pytest `tmp_path` for zarr output (validation mode).
        api_fn: The high-level converter API function.
        api_kwargs: Keyword arguments for `api_fn` (e.g. `acquisitions`).
        snapshot_path: Path to the snapshot JSON file.
        update_snapshots: If True, (re)generate the snapshot instead of checking.
        converter_options: Options controlling the conversion; passed to `api_fn`
            only when provided.
        output_type: `"plate"` for HCS plate output, `"single_image"` for
            individual zarr containers at the top level of the output directory.
    """
    if update_snapshots:
        zarr_dir = snapshot_path.parent.parent / "output"
        api_kwargs = api_kwargs | {"overwrite": OverwriteMode.OVERWRITE}
    else:
        zarr_dir = tmp_path / "output"
    zarr_dir.mkdir(parents=True, exist_ok=True)

    call_kwargs = dict(api_kwargs)
    if converter_options is not None:
        call_kwargs["converter_options"] = converter_options
    updates_list = api_fn(zarr_dir=str(zarr_dir), **call_kwargs)

    actual = build_snapshot(
        zarr_dir=zarr_dir,
        image_list_updates=updates_list,
        output_type=output_type,
    )

    if update_snapshots:
        _write_snapshot(actual, snapshot_path)
        return

    if not snapshot_path.exists():
        raise FileNotFoundError(
            f"Snapshot file {snapshot_path} not found. "
            "Run with --update-snapshots to generate it."
        )

    expected = _load_snapshot(snapshot_path, output_type=output_type)
    diffs = compare_snapshots(expected, actual)
    if diffs:
        joined = "\n".join(f"  - {d}" for d in diffs)
        raise AssertionError(
            f"Snapshot mismatch for {snapshot_path.name} "
            f"({len(diffs)} difference(s)):\n{joined}"
        )