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.

  • 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 
 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
 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
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.
        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")

    def to_roi(self) -> Roi:
        """Convert the Tile to a Roi."""
        acquisition_details = self.acquisition_details
        stage_corrections = acquisition_details.stage_corrections
        spacing = {
            "x": acquisition_details.pixelsize,
            "y": acquisition_details.pixelsize,
            "z": acquisition_details.z_spacing,
            "t": acquisition_details.t_spacing,
        }
        origins = {}
        roi_slices = {}
        for ax in acquisition_details.axes:
            if ax == "x" and stage_corrections.swap_xy:
                ax = "y"
            elif ax == "y" and stage_corrections.swap_xy:
                ax = "x"

            start_field = f"start_{ax}"
            start = getattr(self, start_field)
            start_coo_system = getattr(acquisition_details, f"{start_field}_coo", None)
            if start_coo_system is not None:
                start = safe_to_world(
                    start=start, spacing=spacing[ax], coo_system=start_coo_system
                )

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

            length_field = f"length_{ax}"
            length = getattr(self, length_field)
            length_coo_system = getattr(
                acquisition_details, f"{length_field}_coo", None
            )
            if length_coo_system is not None:
                length = safe_to_world(
                    start=length, spacing=spacing[ax], coo_system=length_coo_system
                )
            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 is not None:
            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 
147
148
149
150
151
def find_data_type(self, resource: Any | None = None) -> str:
    """Find the data type of the image data."""
    if self.acquisition_details.data_type is not None:
        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 
 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
def to_roi(self) -> Roi:
    """Convert the Tile to a Roi."""
    acquisition_details = self.acquisition_details
    stage_corrections = acquisition_details.stage_corrections
    spacing = {
        "x": acquisition_details.pixelsize,
        "y": acquisition_details.pixelsize,
        "z": acquisition_details.z_spacing,
        "t": acquisition_details.t_spacing,
    }
    origins = {}
    roi_slices = {}
    for ax in acquisition_details.axes:
        if ax == "x" and stage_corrections.swap_xy:
            ax = "y"
        elif ax == "y" and stage_corrections.swap_xy:
            ax = "x"

        start_field = f"start_{ax}"
        start = getattr(self, start_field)
        start_coo_system = getattr(acquisition_details, f"{start_field}_coo", None)
        if start_coo_system is not None:
            start = safe_to_world(
                start=start, spacing=spacing[ax], coo_system=start_coo_system
            )

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

        length_field = f"length_{ax}"
        length = getattr(self, length_field)
        length_coo_system = getattr(
            acquisition_details, f"{length_field}_coo", None
        )
        if length_coo_system is not None:
            length = safe_to_world(
                start=length, spacing=spacing[ax], coo_system=length_coo_system
            )
        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
161
162
163
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

    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)
        for region in self.regions[1:]:
            distance = roi_to_point_distance(region.roi, point)
            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)
            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 using."""
        shape = self.shape()
        ref_slice = self.ref_slice()
        ref_data = ref_slice.load_data(axes=self.axes, resource=resource)
        full_image = np.zeros(shape, dtype=ref_data.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 FOV group using Dask."""
        shape = self.shape()
        ref_slice = self.ref_slice()
        ref_data = ref_slice.load_data(axes=self.axes, resource=resource)
        dtype = str(ref_data.dtype)
        slices = self._prepare_slice_loading(resource=resource)
        if chunks is None:
            chunks = ref_data.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 using.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
139
140
141
142
143
144
145
146
147
148
def load_data(self, resource: Any | None = None) -> np.ndarray:
    """Load the full image data for this FOV group using."""
    shape = self.shape()
    ref_slice = self.ref_slice()
    ref_data = ref_slice.load_data(axes=self.axes, resource=resource)
    full_image = np.zeros(shape, dtype=ref_data.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 FOV group using Dask.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
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()
    ref_slice = self.ref_slice()
    ref_data = ref_slice.load_data(axes=self.axes, resource=resource)
    dtype = str(ref_data.dtype)
    slices = self._prepare_slice_loading(resource=resource)
    if chunks is None:
        chunks = ref_data.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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
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)
    for region in self.regions[1:]:
        distance = roi_to_point_distance(region.roi, point)
        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 
86
87
88
89
90
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 
78
79
80
81
82
83
84
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 
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
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(),
            # collection=tile.collection,
            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 
44
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(),
        # collection=tile.collection,
        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 
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
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
    pixelsize: 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")

    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,
            )
            for fov_name, regions in fov_dict.items()
        ]

    @property
    def pixel_size(self) -> PixelSize:
        """Return the PixelSize of the TiledImage."""
        return PixelSize(
            x=self.pixelsize,
            y=self.pixelsize,
            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 TileRegion."""
        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.pixelsize != tile.acquisition_details.pixelsize:
            raise ValueError("Tile pixelsize does not match TiledImage pixelsize.")
        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_region = TileSlice.from_tile(tile)

        if add_translation:
            # This logic is a bit hacky to be improved
            roi_extra = tile_region.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_region)

    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 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."""

        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_slice = region.roi.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 TileRegion.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
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
def add_tile(self, tile: Tile, add_translation: bool = False) -> None:
    """Add a Tile to the TiledImage as a TileRegion."""
    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.pixelsize != tile.acquisition_details.pixelsize:
        raise ValueError("Tile pixelsize does not match TiledImage pixelsize.")
    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_region = TileSlice.from_tile(tile)

    if add_translation:
        # This logic is a bit hacky to be improved
        roi_extra = tile_region.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_region)
group_by_fov() -> list[TileFOVGroup[ImageLoaderInterfaceType]]

Group TileSlices by field of view name.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
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,
        )
        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 
284
285
286
287
288
289
290
291
292
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 
294
295
296
297
298
299
300
301
302
303
304
305
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
    )
roi() -> Roi

Get the global ROI covering all TileSlices in the TiledImage.

Source code in ome_zarr_converters_tools/core/_tile_region.py 
257
258
259
260
261
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 
249
250
251
252
253
254
255
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,
    )

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 TiledImages belonging to an HCS 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.

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
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 TiledImages belonging to an HCS 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.
    """
    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 multiple path components to a base URL.

This is used instead of os.path.join or pathlib.Path to ensure support for both local and S3 URLs.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def join_url_paths(base_url: str, *paths: str) -> str:
    """Join multiple path components to a base URL.

    This is used instead of os.path.join or pathlib.Path to ensure
    support for both local and S3 URLs.
    """
    # Ensure base_url does not end with a slash
    base_url = base_url.rstrip("/")
    # Iterate for all but the last path to avoid adding a trailing slash
    for path in paths:
        # Strip leading slashes from path components
        path = str(path).lstrip("/")
        base_url = f"{base_url}/{path}"
    return base_url

local_url_to_path(url: str) -> Path

Convert a local URL to a Path object.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
24
25
26
27
28
29
def local_url_to_path(url: str) -> Path:
    """Convert a local URL to a Path object."""
    path = Path(url)
    path = path.resolve().absolute()
    path.parent.mkdir(parents=True, exist_ok=True)
    return path

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

Build a list of TiledImages belonging to an HCS acquisition.

Parameters:

  • tiles_table (DataFrame) –

    DataFrame containing the tiles table.

  • acquisition_details (AcquisitionDetails) –

    AcquisitionDetails model for the acquisition.

Source code in ome_zarr_converters_tools/core/_table.py 
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 single_images_from_dataframe(
    *,
    tiles_table: pd.DataFrame,
    acquisition_details: AcquisitionDetails,
) -> list[Tile]:
    """Build a list of TiledImages belonging to an HCS acquisition.

    Args:
        tiles_table: DataFrame containing the tiles table.
        acquisition_details: AcquisitionDetails model for the acquisition.
    """
    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], converter_options: ConverterOptions, resource: Any | None = None) -> list[TiledImage]

Create a TiledImage from a dictionary.

Parameters:

  • tiles (list[Tile]) –

    List of Tile models to build the TiledImage from.

  • converter_options (ConverterOptions) –

    ConverterOptions model for the conversion.

  • 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 
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
58
def tiled_image_from_tiles(
    *,
    tiles: list[Tile],
    converter_options: ConverterOptions,
    resource: Any | None = None,
) -> list[TiledImage]:
    """Create a TiledImage from a dictionary.

    Args:
        tiles: List of Tile models to build the TiledImage from.
        converter_options: ConverterOptions model for the conversion.
        resource: Optional resource to assist in processing.

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

    """
    split_tiles = converter_options.tiling_mode == TilingMode.NO_TILING
    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 not split_tiles:
            suffix = ""
            add_translation = False
        else:
            suffix = f"_{tile.fov_name}"
            add_translation = True
        tile.collection._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,
                pixelsize=acquisition_details.pixelsize,
                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, TilingMode, WriterMode, OverwriteMode, AlignmentCorrections, 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 
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
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 system for start and length values
    start_x_coo: COO_SYSTEM_TYPE = "world"
    start_y_coo: COO_SYSTEM_TYPE = "world"
    start_z_coo: COO_SYSTEM_TYPE = "world"
    start_t_coo: COO_SYSTEM_TYPE = "world"
    length_x_coo: COO_SYSTEM_TYPE = "pixel"
    length_y_coo: COO_SYSTEM_TYPE = "pixel"
    length_z_coo: COO_SYSTEM_TYPE = "pixel"
    length_t_coo: COO_SYSTEM_TYPE = "pixel"
    # Spacing information
    pixelsize: 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 | None = None

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

    # Stage orientation corrections
    stage_corrections: 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 
150
151
152
153
154
155
156
157
@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

ChannelInfo

Bases: BaseModel

Channel information.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class ChannelInfo(BaseModel):
    """Channel information."""

    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.
    """
    colors: DefaultColors = DefaultColors.blue
    """The color associated with the channel, e.g. for visualization purposes."""
channel_label: str instance-attribute

Label of the channel.

colors: DefaultColors = DefaultColors.blue class-attribute instance-attribute

The color associated with the channel, 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.

ConverterOptions

Bases: BaseModel

Options for the OME-Zarr conversion process.

Source code in ome_zarr_converters_tools/models/_converter_options.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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class ConverterOptions(BaseModel):
    """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. This may 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.
    """
    tiling_mode: TilingMode = Field(default=TilingMode.AUTO, title="Tiling Mode")
    """
    Tiling mode to use during conversion.

    - Auto: Automatically determine if Snap to Grid is possible, otherwise use Snap to
      Corners.
    - Snap to Grid: Tile images to fit a regular grid. This is only possible if image
      positions align to a grid (potentially with overlap).
    - 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.
    - No Tiling: Each field of view is written as a single OME-Zarr.
    """
    tiling_tolerance: float = Field(
        default=0, ge=0, title="Tiling Tolerance (in pixels)"
    )
    """
    Tolerance in pixels for determining if Snap to Grid is possible.
    This accounts for minor jitter in microscope stage positions when determining if
    Snap to Grid tiling can be applied.
    """
    alignment_correction: StagePositionCorrections = Field(
        default_factory=StagePositionCorrections,
        title="Alignment Corrections",
    )
    """Alignment correction options."""
    omezarr_options: OmeZarrOptions = Field(
        default_factory=OmeZarrOptions, title="OME-Zarr Options"
    )
    """Options specific to OME-Zarr writing."""
    temp_json_options: TempJsonOptions = Field(
        default_factory=TempJsonOptions, title="Temporary JSON Options"
    )
    """Options for temporary JSON storage."""
    model_config = ConfigDict(extra="forbid")
alignment_correction: StagePositionCorrections = Field(default_factory=StagePositionCorrections, title='Alignment Corrections') class-attribute instance-attribute

Alignment correction options.

omezarr_options: OmeZarrOptions = Field(default_factory=OmeZarrOptions, title='OME-Zarr Options') class-attribute instance-attribute

Options specific to OME-Zarr writing.

temp_json_options: TempJsonOptions = Field(default_factory=TempJsonOptions, title='Temporary JSON Options') class-attribute instance-attribute

Options for temporary JSON storage.

tiling_mode: TilingMode = Field(default=(TilingMode.AUTO), title='Tiling Mode') class-attribute instance-attribute

Tiling mode to use during conversion.

  • Auto: Automatically determine if Snap to Grid is possible, otherwise use Snap to Corners.
  • Snap to Grid: Tile images to fit a regular grid. This is only possible if image positions align to a grid (potentially with overlap).
  • 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.
  • No Tiling: Each field of view is written as a single OME-Zarr.
tiling_tolerance: float = Field(default=0, ge=0, title='Tiling Tolerance (in pixels)') class-attribute instance-attribute

Tolerance in pixels for determining if Snap to Grid is possible. This accounts for minor jitter in microscope stage positions when determining if Snap to Grid tiling can be applied.

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. This may 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

Data type enumeration.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
19
20
21
22
23
24
class DataTypeEnum(StrEnum):
    """Data type enumeration."""

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

DefaultImageLoader

Bases: ImageLoaderInterface

Source code in ome_zarr_converters_tools/models/_loader.py 
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class DefaultImageLoader(ImageLoaderInterface):
    file_path: str

    def load_data(self, resource: Any = None) -> np.ndarray:
        """Load the image data as a NumPy array."""
        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):
            path = join_url_paths(resource, self.file_path)
        else:
            path = self.file_path

        suffix = path.split("/")[-1].split(".")[-1].lower()
        if suffix in ["tiff", "tif"]:
            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:
            raise ValueError(
                f"DefaultImageLoader cannot handle file type {suffix}, "
                "supported types are .tiff, .tif, .png, .jpg, .jpeg, .bmp, .npy"
            )

    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)
load_data(resource: Any = None) -> np.ndarray

Load the image data as a NumPy array.

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

    suffix = path.split("/")[-1].split(".")[-1].lower()
    if suffix in ["tiff", "tif"]:
        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:
        raise ValueError(
            f"DefaultImageLoader cannot handle file type {suffix}, "
            "supported types are .tiff, .tif, .png, .jpg, .jpeg, .bmp, .npy"
        )

FixedSizeChunking

Bases: BaseModel

Chunking strategy with fixed chunk sizes.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class FixedSizeChunking(BaseModel):
    """Chunking strategy with fixed chunk sizes."""

    mode: Literal["Fixed Size"] = "Fixed Size"
    """Fixed size chunking."""
    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_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

Fixed size chunking.

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: BaseModel

Chunking strategy that matches the field of view.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class FovBasedChunking(BaseModel):
    """Chunking strategy that matches the field of view."""

    mode: Literal["Same as FOV"] = "Same as FOV"
    """Chunking based on FOV size."""
    xy_scaling: Scalings = Field(default=Scalings.ONE, title="XY Scaling Factor")
    """
    Scaling factor for XY chunk size. If set to 1, chunk size matches FOV size.
    If set to 0.5, chunk size is half the FOV size (smaller chunks, more files).
    If set to 2, chunk size is double the FOV size (larger chunks, less 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

Chunking based on FOV size.

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 XY chunk size. If set to 1, chunk size matches FOV size. If set to 0.5, chunk size is half the FOV size (smaller chunks, more files). If set to 2, chunk size is double the FOV size (larger chunks, less files).

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

Chunk size for Z dimension.

OmeZarrOptions

Bases: BaseModel

Options specific to OME-Zarr writing.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class OmeZarrOptions(BaseModel):
    """Options specific to OME-Zarr writing."""

    num_levels: int = Field(default=5, ge=1)
    """Number of resolution levels to create."""
    chunks: ChunkingStrategy = Field(
        default_factory=FovBasedChunking, title="Chunking Strategy"
    )
    """Chunking strategy to use."""
    ngff_version: NgffVersions = DefaultNgffVersion
    """Version of the OME-NGFF specification to target."""
    table_backend: BackendType = Field(
        default=BackendType.ANNDATA, title="Table Backend"
    )
    """Backend type for storing tables."""
    model_config = ConfigDict(extra="forbid")
chunks: ChunkingStrategy = Field(default_factory=FovBasedChunking, title='Chunking Strategy') class-attribute instance-attribute

Chunking strategy to use.

ngff_version: NgffVersions = DefaultNgffVersion class-attribute instance-attribute

Version of the OME-NGFF specification to target.

num_levels: int = Field(default=5, ge=1) class-attribute instance-attribute

Number of resolution levels to create.

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

Backend type for storing tables.

StageOrientation

Bases: BaseModel

Stage orientation corrections.

Source code in ome_zarr_converters_tools/models/_acquisition.py 
 98
 99
100
101
102
103
104
105
106
107
class StageOrientation(BaseModel):
    """Stage orientation corrections."""

    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."""
    model_config = ConfigDict(extra="forbid")
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: BaseModel

Alignment correction for stage positions.

Source code in ome_zarr_converters_tools/models/_converter_options.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class StagePositionCorrections(BaseModel):
    """Alignment correction for stage positions."""

    align_xy: bool = Field(default=False, title="Align XY")
    """
    Whether to align the positions in the XY plane by FOV.
    This addresses minor imprecision that often occurs during image acquisition.
    """
    align_z: bool = Field(default=False, title="Align Z")
    """
    Whether to align the positions in the Z axis by FOV.
    This addresses minor imprecision that often occurs during image acquisition.
    """
    align_t: bool = Field(default=False, title="Align T")
    """
    Whether to align the positions in the T axis by FOV.
    This addresses minor imprecision that often occurs during image acquisition.
    """
    model_config = ConfigDict(extra="forbid")
align_t: bool = Field(default=False, title='Align T') class-attribute instance-attribute

Whether to align the positions in the T axis by FOV. This addresses minor imprecision that often occurs during image acquisition.

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

Whether to align the positions in the XY plane by FOV. This addresses minor imprecision that often occurs during image acquisition.

align_z: bool = Field(default=False, title='Align Z') class-attribute instance-attribute

Whether to align the positions in the Z axis by FOV. This addresses minor imprecision that often occurs during image acquisition.

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 
27
28
29
30
31
32
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"]

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

Join multiple path components to a base URL.

This is used instead of os.path.join or pathlib.Path to ensure support for both local and S3 URLs.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def join_url_paths(base_url: str, *paths: str) -> str:
    """Join multiple path components to a base URL.

    This is used instead of os.path.join or pathlib.Path to ensure
    support for both local and S3 URLs.
    """
    # Ensure base_url does not end with a slash
    base_url = base_url.rstrip("/")
    # Iterate for all but the last path to avoid adding a trailing slash
    for path in paths:
        # Strip leading slashes from path components
        path = str(path).lstrip("/")
        base_url = f"{base_url}/{path}"
    return base_url

local_url_to_path(url: str) -> Path

Convert a local URL to a Path object.

Source code in ome_zarr_converters_tools/models/_url_utils.py 
24
25
26
27
28
29
def local_url_to_path(url: str) -> Path:
    """Convert a local URL to a Path object."""
    path = Path(url)
    path = path.resolve().absolute()
    path.parent.mkdir(parents=True, exist_ok=True)
    return path

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.

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'.

  • 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 
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
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.

    Args:
        collection_type: Name of the collection setup handler. By convention,
            the name of the CollectionInterfaceType, e.g., 'SingleImage'
            or 'ImageInPlate'.
        function: Function that performs the collection setup step.
        overwrite: Whether to overwrite an existing collection setup step
            with the same name.
    """
    if collection_type is None:
        collection_type = function.__name__
    if not overwrite and collection_type in _collection_setup_registry:
        raise ValueError(
            f"Collection setup handler '{collection_type}' is already registered."
        )
    _collection_setup_registry[collection_type] = function

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

Register a new filter.

Parameters:

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

    Name of the registration step.

  • function (FilterFunctionProtocol) –

    Function that performs the registration step.

  • overwrite (bool, default: False ) –

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

Source code in ome_zarr_converters_tools/pipelines/_filters.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def add_filter(
    *,
    function: FilterFunctionProtocol,
    name: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new filter.

    Args:
        name: Name of the registration step.
        function: Function that performs the registration step.
        overwrite: Whether to overwrite an existing registration step
            with the same name.
    """
    if name is None:
        name = function.__name__
    if not overwrite and name in _filter_registry:
        raise ValueError(f"Filter step '{name}' is already registered.")
    _filter_registry[name] = function

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

Register a new registration step function.

Parameters:

  • name (str) –

    Name of the registration step.

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

    Function that performs the registration step.

  • overwrite (bool, default: False ) –

    Whether to overwrite an existing registration step.

Source code in ome_zarr_converters_tools/pipelines/_registration_pipeline.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def add_registration_func(
    function: Callable[..., TiledImage],
    name: str,
    overwrite: bool = False,
) -> None:
    """Register a new registration step function.

    Args:
        name: Name of the registration step.
        function: Function that performs the registration step.
        overwrite: Whether to overwrite an existing registration step.
    """
    if not overwrite and name in _registration_registry:
        raise ValueError(f"Registration step '{name}' is already registered.")
    _registration_registry[name] = function

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

Register a new validator function.

Parameters:

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

    Name of the registration step.

  • function (ValidatorFunctionProtocol) –

    Function that performs the registration step.

  • overwrite (bool, default: False ) –

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

Source code in ome_zarr_converters_tools/pipelines/_validators.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def add_validator(
    function: ValidatorFunctionProtocol,
    name: str | None = None,
    overwrite: bool = False,
) -> None:
    """Register a new validator function.

    Args:
        name: Name of the registration step.
        function: Function that performs the registration step.
        overwrite: Whether to overwrite an existing registration step
            with the same name.
    """
    if name is None:
        name = function.__name__
    if not overwrite and name in _validator_registry:
        raise ValueError(f"Validator step '{name}' is already registered.")
    _validator_registry[name] = function

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.

Returns:

  • None –

    The list of TiledImage after applying the collection setup handler.

Source code in ome_zarr_converters_tools/pipelines/_collection_setup.py 
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
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.

    Returns:
        The list of TiledImage after applying the collection setup handler.
    """
    collection_type = collection_type
    setup_function = _collection_setup_registry.get(collection_type)
    if setup_function is None:
        raise ValueError(
            f"Collection setup handler '{collection_type}' is not registered."
        )
    return setup_function(
        tiled_images=tiled_images,
        zarr_dir=zarr_dir,
        ngff_version=ngff_version,
        overwrite_mode=overwrite_mode,
    )

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
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."""
    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[ValidatorStep] | 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[ValidatorStep] | 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[ValidatorStep] | 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,
        converter_options=converter_options,
        resource=resource,
    )
    if validators is not None:
        tiled_images = apply_validator_pipeline(
            tiled_images, validators_config=validators
        )
    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 ( OmeZarrContainer ) –

    The written OME-Zarr container.

Source code in ome_zarr_converters_tools/pipelines/_write_ome_zarr.py 
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
212
213
214
215
216
217
218
219
220
221
222
223
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:
        OmeZarrContainer: 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
    tiled_image.regions = _region_to_pixel_coordinates(
        tiled_image.regions,
        tiled_image.pixel_size,
    )
    base_group = zarr.open_group(store=zarr_url, mode=mode, zarr_format=zarr_format)
    omezarr_options = converter_options.omezarr_options
    try:
        # This can only succeed in "extend" mode if the group already exists
        ome_zarr = open_ome_zarr_container(base_group, cache=True)
        return ome_zarr

    except Exception:
        channels_meta = build_channels_meta(tiled_image)
        ome_zarr = create_empty_ome_zarr(
            store=base_group,
            axes_names=tiled_image.axes,
            shape=tiled_image.shape(),
            chunks=_compute_chunk_size(tiled_image, omezarr_options),
            pixelsize=tiled_image.pixelsize,
            z_spacing=tiled_image.z_spacing,
            time_spacing=tiled_image.t_spacing,
            levels=omezarr_options.num_levels,
            channels_meta=channels_meta,
            translation=tiled_image.translation,
            overwrite=True,
            ngff_version=omezarr_options.ngff_version,
        )
    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 tiled_image.group_by_fov():
            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.

ome_zarr_converters_tools.fractal

API for building OME-Zarr converters tasks for Fractal.

AcquisitionOptions

Bases: BaseModel

Acquisition options for conversion.

These are option that can be specified per acquisition. by the user at conversion time. This is not to be confused with AcquisitionDetails, this model is used in fractal tasks to override/update details from AcquisitionDetails model.

Source code in ome_zarr_converters_tools/fractal/_models.py 
 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
class AcquisitionOptions(BaseModel):
    """Acquisition options for conversion.

    These are option that can be specified per acquisition.
    by the user at conversion time.
    This is not to be confused with AcquisitionDetails,
    this model is used in fractal tasks to override/update
    details from AcquisitionDetails model.
    """

    channels: list[ChannelInfo] | None = None
    """List of channel information."""
    pixel_info: PixelSizeModel | None = Field(
        default=None, title="Pixel Size Information"
    )
    """Pixel size information."""
    condition_table_path: str | None = None
    """Optional path to a condition table CSV file."""
    axes: str | None = None
    """Axes to use for the image data, e.g. "czyx"."""
    data_type: DataTypeEnum | None = Field(default=None, title="Data Type")
    """Data type of the image data."""
    stage_corrections: StageOrientation = Field(
        default_factory=StageOrientation, title="Stage Corrections"
    )
    """Stage orientation corrections."""
    filters: list[ImplementedFilters] = Field(default_factory=list)
    """List of filters to apply."""

    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

    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_details.channels = self.channels
        if self.pixel_info is not None:
            updated_details.pixelsize = self.pixel_info.pixelsize
            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 is not None:
            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 to use for the image data, e.g. "czyx".

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

List of channel information.

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

Optional path to a condition table CSV file.

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

Data type of the image data.

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

List of filters to apply.

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

Pixel size information.

stage_corrections: StageOrientation = Field(default_factory=StageOrientation, title='Stage Corrections') class-attribute instance-attribute

Stage orientation corrections.

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 
74
75
76
77
78
79
80
81
82
83
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
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 
 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
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_details.channels = self.channels
    if self.pixel_info is not None:
        updated_details.pixelsize = self.pixel_info.pixelsize
        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 is not None:
        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

ConvertParallelInitArgs

Bases: BaseModel

Arguments for the compute task.

Source code in ome_zarr_converters_tools/fractal/_models.py 
20
21
22
23
24
25
class ConvertParallelInitArgs(BaseModel):
    """Arguments for the compute task."""

    tiled_image_json_dump_url: str
    converter_options: ConverterOptions
    overwrite_mode: OverwriteMode = OverwriteMode.NO_OVERWRITE

PixelSizeModel

Bases: BaseModel

Pixel size model.

Source code in ome_zarr_converters_tools/fractal/_models.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class PixelSizeModel(BaseModel):
    """Pixel size model."""

    pixelsize: float
    """
    Pixel size in micrometers.
    """
    z_spacing: float
    """
    Z spacing in micrometers.
    """
    t_spacing: float
    """
    Time spacing in seconds.
    """
pixelsize: float instance-attribute

Pixel size in micrometers.

t_spacing: float instance-attribute

Time spacing in seconds.

z_spacing: float instance-attribute

Z spacing in micrometers.

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 
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 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 (str): 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 
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
181
182
183
184
185
186
187
188
189
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",
            "WellExcludeFilter",
        ),
        (
            base,
            "pipelines/_filters.py",
            "WellIncludeFilter",
        ),
        (
            base,
            "pipelines/_filters.py",
            "RegexIncludeFilter",
        ),
        (
            base,
            "pipelines/_filters.py",
            "RegexExcludeFilter",
        ),
        (
            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/_converter_options.py",
            "TempJsonOptions",
        ),
        (
            base,
            "models/_converter_options.py",
            "FovBasedChunking",
        ),
        (
            base,
            "models/_converter_options.py",
            "FixedSizeChunking",
        ),
        (
            base,
            "models/_acquisition.py",
            "ChannelInfo",
        ),
    ]

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

Create a pickle file for the tiled image.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def dump_to_json(temp_json_url: str, tiled_image: TiledImage) -> str:
    """Create a pickle file for the tiled image."""
    json_data = tiled_image.model_dump_json()
    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_data)
    logger.debug(f"JSON file created: {tile_json_name}")

    return tile_json_name

generic_compute_task(*, zarr_url: str, init_args: ConvertParallelInitArgs, collection_type: type[CollectionInterfaceType], image_loader_type: type[ImageLoaderInterfaceType], resource: Any = 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) –

    Arguments from the initialization task.

  • 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, default: None ) –

    The resource to associate with the context model.

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

    Args:
        zarr_url (str): URL to the OME-Zarr file.
        init_args (ConvertParallelInitArgs): Arguments from the initialization task.
        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): The resource to associate with the context model.
    """
    logger.info(f"Starting conversion for Zarr URL: {zarr_url}")
    for t in range(3):  # Retry up to 3 times
        try:
            tiled_image_loaded = tiled_image_from_json(
                tiled_image_json_dump_url=init_args.tiled_image_json_dump_url,
                collection_type=collection_type,
                image_loader_type=image_loader_type,
            )
            logger.info(
                f"Successfully loaded JSON file: {init_args.tiled_image_json_dump_url}"
            )
            break  # Exit loop if successful
        except FileNotFoundError:
            logger.error(
                f"JSON file does not exist: "
                f"{init_args.tiled_image_json_dump_url}, retrying..."
            )
            sleep_time = 2 ** (t + 1)
            time.sleep(sleep_time)
    else:
        raise FileNotFoundError(
            f"JSON file does not exist after 3 retries: "
            f"{init_args.tiled_image_json_dump_url}"
        )
    registration_pipeline = build_default_registration_pipeline(
        alignment_corrections=init_args.converter_options.alignment_correction,
        tiling_mode=init_args.converter_options.tiling_mode,
        tolerance=init_args.converter_options.tiling_tolerance,
    )
    ome_zarr = tiled_image_creation_pipeline(
        zarr_url=zarr_url,
        tiled_image=tiled_image_loaded,
        registration_pipeline=registration_pipeline,
        converter_options=init_args.converter_options,
        writer_mode=init_args.converter_options.writer_mode,
        overwrite_mode=init_args.overwrite_mode,
        resource=resource,
    )
    remove_json(init_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 
 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
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 (str): 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 = tiled_image_json_dump_url.rsplit(fs.sep, 1)[0]
        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 TiledImageWithContext models 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 
 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
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 TiledImageWithContext models 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.
    """
    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:

  • TiledImage ( TiledImage ) –

    The loaded TiledImage object.

Source code in ome_zarr_converters_tools/fractal/_json_utils.py 
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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 (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:
        TiledImage: 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:
            logger.error(
                f"JSON file does not exist: {tiled_image_json_dump_url}, retrying..."
            )
            sleep_time = 2 ** (t + 1)
            time.sleep(sleep_time)

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