HCS API reference¶

def wells_paths(self) -> list[str]:
    """Return the wells paths in the plate."""
    return self.meta.wells_paths

@deprecated(replacement="images_paths()")
async def images_paths_async(self, acquisition: int | None = None) -> list[str]:
    """Return the images paths in the plate asynchronously.

    Deprecated: use `images_paths()`. Well metadata is read from the plate's
    own attributes, so there was never any IO here to parallelise.

    Args:
        acquisition: The acquisition id to filter the images.
    """
    return self.images_paths(acquisition=acquisition)

def images_paths(self, acquisition: int | None = None) -> list[str]:
    """Return the images paths in the plate.

    If acquisition is None, return all images paths in the plate.
    Else, return the images paths in the plate for the given acquisition.

    Args:
        acquisition (int | None): The acquisition id to filter the images.
    """
    wells = self.get_wells()
    images = []
    for well_path, well in wells.items():
        for img_path in well.paths(acquisition):
            images.append(f"{well_path}/{img_path}")
    return images

def well_images_paths(
    self, row: str, column: int | str, acquisition: int | None = None
) -> list[str]:
    """Return the images paths in a well.

    If acquisition is None, return all images paths in the well.
    Else, return the images paths in the well for the given acquisition.

    Args:
        row (str): The row of the well.
        column (int | str): The column of the well.
        acquisition (int | None): The acquisition id to filter the images.
    """
    images = []
    well = self.get_well(row=row, column=column)
    for path in well.paths(acquisition):
        images.append(self._image_path(row=row, column=column, path=path))
    return images

def get_image_acquisition_id(
    self, row: str, column: int | str, image_path: str
) -> int | None:
    """Get the acquisition id of an image in a well.

    Args:
        row (str): The row of the well.
        column (int | str): The column of the well.
        image_path (str): The path of the image.

    Returns:
        int | None: The acquisition id of the image.
    """
    well = self.get_well(row=row, column=column)
    return well.get_image_acquisition_id(image_path=image_path)

def get_well(self, row: str, column: int | str) -> OmeZarrWell:
    """Get a well from the plate.

    Args:
        row (str): The row of the well.
        column (int | str): The column of the well.

    Returns:
        OmeZarrWell: The well.
    """
    well_path = self._well_path(row=row, column=column)
    return self._get_well(well_path=well_path)

@deprecated(replacement="get_wells(max_workers=...)")
async def get_wells_async(
    self, max_workers: int | None = None
) -> dict[str, OmeZarrWell]:
    """Get all wells in the plate asynchronously.

    Deprecated: use `get_wells(max_workers=...)`.

    Args:
        max_workers: How many wells to open at a time. `None` leaves the
            fan-out to asyncio's default thread executor.

    Returns:
        A dictionary of wells, keyed by well path.
    """
    paths = self.wells_paths()
    factories = [(lambda p=p: asyncio.to_thread(self._get_well, p)) for p in paths]
    wells = await _gather_bounded(factories, max_workers=max_workers)
    return dict(zip(paths, wells, strict=True))

def get_wells(self, max_workers: int | None = None) -> dict[str, OmeZarrWell]:
    """Get all wells in the plate.

    Args:
        max_workers: How many wells to open concurrently. `None` (the
            default) opens them one at a time in the calling thread.

    Returns:
        A dictionary of wells, keyed by well path.
    """
    paths = self.wells_paths()
    wells = _map_workers(self._get_well, paths, max_workers=max_workers)
    return dict(zip(paths, wells, strict=True))

@deprecated(replacement="get_images(max_workers=...)")
async def get_images_async(
    self, acquisition: int | None = None, max_workers: int | None = None
) -> dict[str, OmeZarrContainer]:
    """Get all images in the plate asynchronously.

    Deprecated: use `get_images(max_workers=...)`.

    Args:
        acquisition: The acquisition id to filter the images.
        max_workers: How many images to open at a time. `None` leaves the
            fan-out to asyncio's default thread executor.

    Returns:
        A dictionary of images, keyed by image path.
    """
    paths = self.images_paths(acquisition=acquisition)
    factories = [(lambda p=p: asyncio.to_thread(self._get_image, p)) for p in paths]
    images = await _gather_bounded(factories, max_workers=max_workers)
    return dict(zip(paths, images, strict=True))

def get_images(
    self, acquisition: int | None = None, max_workers: int | None = None
) -> dict[str, OmeZarrContainer]:
    """Get all images in the plate.

    Args:
        acquisition: The acquisition id to filter the images.
        max_workers: How many images to open concurrently. `None` (the
            default) opens them one at a time in the calling thread.

    Returns:
        A dictionary of images, keyed by image path.
    """
    paths = self.images_paths(acquisition=acquisition)
    images = _map_workers(self._get_image, paths, max_workers=max_workers)
    return dict(zip(paths, images, strict=True))

def get_image(
    self, row: str, column: int | str, image_path: str
) -> OmeZarrContainer:
    """Get an image from the plate.

    Args:
        row (str): The row of the well.
        column (int | str): The column of the well.
        image_path (str): The path of the image.

    Returns:
        OmeZarrContainer: The image.
    """
    image_path = self._image_path(row=row, column=column, path=image_path)
    return self._get_image(image_path)

def get_image_store(
    self, row: str, column: int | str, image_path: str
) -> StoreOrGroup:
    """Get the image store from the plate.

    Args:
        row (str): The row of the well.
        column (int | str): The column of the well.
        image_path (str): The path of the image.
    """
    well = self.get_well(row=row, column=column)
    return well.get_image_store(image_path=image_path)

def get_well_images(
    self, row: str, column: str | int, acquisition: int | None = None
) -> dict[str, OmeZarrContainer]:
    """Get all images in a well.

    Args:
        row: The row of the well.
        column: The column of the well.
        acquisition: The acquisition id to filter the images.
    """
    images = {}
    for image_paths in self.well_images_paths(
        row=row, column=column, acquisition=acquisition
    ):
        group_handler = self._group_handler.get_handler(image_paths)
        images[image_paths] = OmeZarrContainer(group_handler)
    return images

def atomic_add_image(
    self,
    row: str,
    column: int | str,
    image_path: str,
    acquisition_id: int | None = None,
    acquisition_name: str | None = None,
) -> str:
    """Parallel safe version of `add_image`.

    Every worker adding to a plate contends on the same two metadata files,
    the plate's for the well list and the well's for the image list. This
    serialises both read-modify-writes behind a file lock, so concurrent
    workers cannot lose each other's updates. The lock is an OS file lock:
    it holds across threads and processes on one machine, and on a shared
    network filesystem only if the mount honours `flock`.

    Note:
        On Windows the lock is best-effort and warns: `filelock` can hand
        the same lock to two workers at once, so a single writer is safe but
        concurrent ones can still lose an update. Run those on Linux/macOS.

    Raises:
        NgioValueError: If the store is not local, or if the plate was
            opened with caching enabled — neither supports the lock.
    """
    if image_path is None:
        raise ValueError(
            "Image path cannot be None for atomic add_image. "
            "If your intent is to add a well, use add_well instead."
        )
    path = self._add_image(
        row=row,
        column=column,
        image_path=image_path,
        acquisition_id=acquisition_id,
        acquisition_name=acquisition_name,
        atomic=True,
    )
    return path

def add_image(
    self,
    row: str,
    column: int | str,
    image_path: str,
    acquisition_id: int | None = None,
    acquisition_name: str | None = None,
) -> str:
    """Add an image to an ome-zarr plate."""
    if image_path is None:
        raise ValueError(
            "Image path cannot be None for atomic add_image. "
            "If your intent is to add a well, use add_well instead."
        )
    path = self._add_image(
        row=row,
        column=column,
        image_path=image_path,
        acquisition_id=acquisition_id,
        acquisition_name=acquisition_name,
        atomic=False,
    )
    return path

def add_well(
    self,
    row: str,
    column: int | str,
) -> OmeZarrWell:
    """Add a well to an ome-zarr plate."""
    _ = self._add_image(
        row=row,
        column=column,
        image_path=None,
        acquisition_id=None,
        acquisition_name=None,
        atomic=False,
    )
    return self.get_well(row=row, column=column)

def add_column(
    self,
    column: int | str,
) -> "OmeZarrPlate":
    """Add a column to an ome-zarr plate."""
    meta, _ = self.meta.add_column(column)
    self.meta_handler.update_meta(meta)
    self.meta_handler._group_handler.clean_cache()
    return self

def add_row(
    self,
    row: str,
) -> "OmeZarrPlate":
    """Add a row to an ome-zarr plate."""
    meta, _ = self.meta.add_row(row)
    self.meta_handler.update_meta(meta)
    self.meta_handler._group_handler.clean_cache()
    return self

def add_acquisition(
    self,
    acquisition_id: int,
    acquisition_name: str,
) -> "OmeZarrPlate":
    """Add an acquisition to an ome-zarr plate.

    Be aware that this is not a parallel safe operation.

    Args:
        acquisition_id (int): The acquisition id.
        acquisition_name (str): The acquisition name.
    """
    meta = self.meta.add_acquisition(
        acquisition_id=acquisition_id, acquisition_name=acquisition_name
    )
    self.meta_handler.update_meta(meta)
    self.meta_handler._group_handler.clean_cache()
    return self

def atomic_remove_image(
    self,
    row: str,
    column: int | str,
    image_path: str,
):
    """Parallel safe version of `remove_image`.

    Serialises the read-modify-write of the plate and well metadata behind a
    file lock, so concurrent workers cannot lose each other's updates. The
    lock is an OS file lock: it holds across threads and processes on one
    machine, and on a shared network filesystem only if the mount honours
    `flock`.

    Note:
        On Windows the lock is best-effort and warns: `filelock` can hand
        the same lock to two workers at once, so a single writer is safe but
        concurrent ones can still lose an update. Run those on Linux/macOS.

    Raises:
        NgioValueError: If the store is not local, or if the plate was
            opened with caching enabled — neither supports the lock.
    """
    return self._remove_image(
        row=row,
        column=column,
        image_path=image_path,
        atomic=True,
    )

def remove_image(
    self,
    row: str,
    column: int | str,
    image_path: str,
):
    """Remove an image from an ome-zarr plate."""
    return self._remove_image(
        row=row,
        column=column,
        image_path=image_path,
        atomic=False,
    )

def derive_plate(
    self,
    store: StoreOrGroup,
    plate_name: str | None = None,
    ngff_version: NgffVersions | None = None,
    keep_acquisitions: bool = False,
    cache: bool = False,
    overwrite: bool = False,
) -> "OmeZarrPlate":
    """Derive a new OME-Zarr plate from an existing one.

    Args:
        store (StoreOrGroup): The Zarr store or group that stores the plate.
        plate_name (str | None): The name of the new plate.
        ngff_version (NgffVersion): The NGFF version to use for the new plate.
        keep_acquisitions (bool): Whether to keep the acquisitions in the new plate.
        cache (bool): Whether to use a cache for the zarr group metadata.
        overwrite (bool): Whether to overwrite the existing plate.
    """
    return derive_ome_zarr_plate(
        ome_zarr_plate=self,
        store=store,
        plate_name=plate_name,
        ngff_version=ngff_version,
        keep_acquisitions=keep_acquisitions,
        cache=cache,
        overwrite=overwrite,
    )

def list_tables(self, filter_types: TypedTable | str | None = None) -> list[str]:
    """List all tables in the plate."""
    tables_container = self._get_tables_container(create_mode=False)
    if tables_container is None:
        return []
    return tables_container.list(filter_types=filter_types)

def list_roi_tables(self) -> list[str]:
    """List all ROI tables in the plate.

    Returns `[]` when the plate has no tables, matching `list_tables`.
    """
    tables_container = self._get_tables_container(create_mode=False)
    if tables_container is None:
        return []

    roi = tables_container.list(filter_types="roi_table")
    masking_roi = tables_container.list(filter_types="masking_roi_table")
    return roi + masking_roi

def get_roi_table(self, name: str) -> RoiTable:
    """Get a ROI table from the image.

    Args:
        name (str): The name of the table.
    """
    table = self.tables_container.get(name=name, strict=True)
    if not isinstance(table, RoiTable):
        raise NgioValueError(f"Table {name} is not a ROI table. Got {type(table)}")
    return table

def get_masking_roi_table(self, name: str) -> MaskingRoiTable:
    """Get a masking ROI table from the image.

    Args:
        name (str): The name of the table.
    """
    table = self.tables_container.get(name=name, strict=True)
    if not isinstance(table, MaskingRoiTable):
        raise NgioValueError(
            f"Table {name} is not a masking ROI table. Got {type(table)}"
        )
    return table

def get_feature_table(self, name: str) -> FeatureTable:
    """Get a feature table from the image.

    Args:
        name (str): The name of the table.
    """
    table = self.tables_container.get(name=name, strict=True)
    if not isinstance(table, FeatureTable):
        raise NgioValueError(
            f"Table {name} is not a feature table. Got {type(table)}"
        )
    return table

def get_generic_roi_table(self, name: str) -> GenericRoiTable:
    """Get a generic ROI table from the image.

    Args:
        name (str): The name of the table.
    """
    table = self.tables_container.get(name=name, strict=True)
    if not isinstance(table, GenericRoiTable):
        raise NgioValueError(
            f"Table {name} is not a generic ROI table. Got {type(table)}"
        )
    return table

def get_condition_table(self, name: str) -> ConditionTable:
    """Get a condition table from the image.

    Args:
        name (str): The name of the table.
    """
    table = self.tables_container.get(name=name, strict=True)
    if not isinstance(table, ConditionTable):
        raise NgioValueError(
            f"Table {name} is not a condition table. Got {type(table)}"
        )
    return table

def get_table(self, name: str) -> Table:
    """Get a table from the image.

    Args:
        name (str): The name of the table.
    """
    return self.tables_container.get(name=name, strict=False)

def get_table_as(
    self,
    name: str,
    table_cls: type[TableType],
    backend: TableBackend | None = None,
) -> TableType:
    """Get a table from the image as a specific type.

    Args:
        name (str): The name of the table.
        table_cls (type[TableType]): The type of the table.
        backend (TableBackend | None): The backend to use. If None,
            the default backend is used.
    """
    return self.tables_container.get_as(
        name=name,
        table_cls=table_cls,
        backend=backend,
    )

def add_table(
    self,
    name: str,
    table: Table,
    backend: TableBackend | None = None,
    overwrite: bool = False,
) -> None:
    """Add a table to the plate.

    If `backend` is `None` (default), the table's own backend is preserved.
    """
    self.tables_container.add(
        name=name, table=table, backend=backend, overwrite=overwrite
    )

def delete_table(self, name: str, missing_ok: bool = False) -> None:
    """Delete a table from the group.

    Args:
        name (str): The name of the table to delete.
        missing_ok (bool): If True, do not raise an error if the table does not
            exist.

    """
    table_container = self._get_tables_container(create_mode=False)
    if table_container is None and missing_ok:
        return
    if table_container is None:
        raise NgioValueError(
            f"No tables found in the image, cannot delete {name}. "
            "Set missing_ok=True to ignore this error."
        )
    table_container.delete(name=name, missing_ok=missing_ok)

def list_image_tables(
    self,
    acquisition: int | None = None,
    filter_types: str | None = None,
    mode: Literal["common", "all"] = "common",
    max_workers: int | None = None,
) -> list[str]:
    """List all image tables in the plate.

    Args:
        acquisition: The acquisition id to filter the images.
        filter_types: The type of tables to filter. If None, return all
            tables.
        mode: Whether to return only tables common to every image
            (`"common"`) or the union across them (`"all"`).
        max_workers: How many images to read concurrently. `None` (the
            default) reads them one at a time in the calling thread.
    """
    images = tuple(
        self.get_images(acquisition=acquisition, max_workers=max_workers).values()
    )
    return list_image_tables(
        images=images,
        filter_types=filter_types,
        mode=mode,
        max_workers=max_workers,
    )

@deprecated(replacement="list_image_tables(max_workers=...)")
async def list_image_tables_async(
    self,
    acquisition: int | None = None,
    filter_types: str | None = None,
    mode: Literal["common", "all"] = "common",
    max_workers: int | None = None,
) -> list[str]:
    """List all image tables in the plate asynchronously.

    Deprecated: use `list_image_tables(max_workers=...)`.

    Args:
        acquisition: The acquisition id to filter the images.
        filter_types: The type of tables to filter. If None, return all
            tables.
        mode: Whether to return only tables common to every image
            (`"common"`) or the union across them (`"all"`).
        max_workers: How many images to read at a time. `None` leaves the
            fan-out to asyncio's default thread executor.
    """
    images = tuple(self.get_images(acquisition=acquisition).values())
    return await _list_image_tables_async(
        images=images,
        filter_types=filter_types,
        mode=mode,
        max_workers=max_workers,
    )

def concatenate_image_tables(
    self,
    name: str,
    acquisition: int | None = None,
    strict: bool = True,
    index_key: str | None = None,
    mode: Literal["eager", "lazy"] = "eager",
    max_workers: int | None = None,
) -> Table:
    """Concatenate tables from all images in the plate.

    Args:
        name: The name of the table to concatenate.
        index_key: The key to use for the index of the concatenated table.
        acquisition: The acquisition id to filter the images.
        strict: If True, raise an error if the table is not found in the image.
        index_key: If a string is provided, a new index column will be created
            new_index_pattern = {row}_{column}_{path_in_well}_{label}
        mode: The mode to use for concatenation. Can be 'eager' or 'lazy'.
            if 'eager', the table will be loaded into memory.
            if 'lazy', the table will be loaded as a lazy frame.
        max_workers: How many images to read concurrently. `None` (the
            default) reads them one at a time in the calling thread.
    """
    images = self.get_images(acquisition=acquisition, max_workers=max_workers)
    extras = _build_extras(tuple(images.keys()))
    return concatenate_image_tables(
        images=tuple(images.values()),
        extras=extras,
        name=name,
        index_key=index_key,
        strict=strict,
        mode=mode,
        max_workers=max_workers,
    )