HCS API reference¶

def refresh(self) -> None:
    """Re-read metadata that `cache=True` holds for the object's lifetime.

    The answer to "another worker wrote to this plate and I want to see it".
    A no-op when the plate was opened with `cache=False`, where nothing is
    held. The atomic operations refresh themselves, so this is only needed
    for plain reads.
    """
    self._group_handler.clean_cache()
    self._wells_cache.clear()
    self._images_cache.clear()
    # Rebuilt on next access against the refreshed handler — it carries a
    # per-table type memo that would otherwise survive the refresh.
    self._tables_container = None
    self._tables_absent = False

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

def images_paths(
    self, acquisition: int | None = None, max_workers: MaxWorkers = 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.

    Image paths live in each well's own metadata, not the plate's, so this
    reads one document per well and there is no shortcut. What there is, on
    a remote store, is concurrency: those reads are round-trip bound and
    independent.

    Args:
        acquisition: The acquisition id to filter the images.
        max_workers: How many wells to read concurrently. `None` (the
            default) reads them one at a time; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep reading serially, or
            `"auto"` for a pool sized for round-trip-bound work.
    """
    wells = self.get_wells(max_workers=max_workers)
    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.
    """
    # Resolved once, not once per image: a per-image `_image_path` call
    # re-opens the well and re-reads the plate document, so the cost would
    # grow with the square of the well's image
    # count, to re-derive a prefix already in hand. The membership check
    # `_image_path` performs is vacuous here: these paths come from
    # `well.paths()` itself.
    well_path = self._well_path(row=row, column=column)
    well = self._get_well(well_path=well_path)
    return [f"{well_path}/{path}" for path in well.paths(acquisition)]

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)

def get_wells(self, max_workers: MaxWorkers = 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; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep it serial, or
            `"auto"` for a pool sized for round-trip-bound work.

    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))

def get_images(
    self, acquisition: int | None = None, max_workers: MaxWorkers = 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; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep it serial, or
            `"auto"` for a pool sized for round-trip-bound work.

    Returns:
        A dictionary of images, keyed by image path.
    """
    paths = self.images_paths(acquisition=acquisition, max_workers=max_workers)
    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.
    """
    # `_get_image`, not a fresh container per call: this duplicated it minus
    # the `_images_cache` lookup, so under `cache=True` it rebuilt the same
    # container every time and handed back a different object than
    # `get_images` did for the same path.
    return {
        image_path: self._get_image(image_path)
        for image_path in self.well_images_paths(
            row=row, column=column, acquisition=acquisition
        )
    }

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. The lock is an OS
            file lock, so there is nothing to take on a remote store.
    """
    if image_path is None:
        raise NgioValueError(
            "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 NgioValueError(
            "Image path cannot be None for 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.invalidate_meta()
    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.invalidate_meta()
    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.invalidate_meta()
    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. The lock is an OS
            file lock, so there is nothing to take on a remote store.
    """
    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 []

    # One pass, not one per type: each `list(filter_types=...)` opens every
    # table to read its type, so asking twice read every document twice to
    # sort names the first pass had already sorted.
    types = tables_container.table_types()
    return [
        name for name, table_type in types.items() if table_type in ROI_TABLE_TYPES
    ]

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: MaxWorkers = 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; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep it serial, or
            `"auto"` for a pool sized for round-trip-bound work.
    """
    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,
    )

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: MaxWorkers = 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; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep it serial, or
            `"auto"` for a pool sized for round-trip-bound work.
    """
    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,
    )

def concatenate_image_tables_as(
    self,
    name: str,
    table_cls: type[TableType],
    acquisition: int | None = None,
    index_key: str | None = None,
    strict: bool = True,
    mode: Literal["eager", "lazy"] = "eager",
    max_workers: MaxWorkers = None,
) -> TableType:
    """Concatenate tables from all images in the plate as a specific type.

    Args:
        name: The name of the table to concatenate.
        table_cls: The type 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.
        index_key: If a string is provided, a new index column will be created
            new_index_pattern = {row}_{column}_{path_in_well}_{label}
        strict: If True, raise an error if the table is not found in the image.
        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; that default becomes
            `"auto"` in ngio=1.2. Pass `1` to keep it serial, or
            `"auto"` for a pool sized for round-trip-bound work.
    """
    images = self.get_images(acquisition=acquisition, max_workers=max_workers)
    extras = _build_extras(tuple(images.keys()))
    return concatenate_image_tables_as(
        images=tuple(images.values()),
        extras=extras,
        name=name,
        table_cls=table_cls,
        index_key=index_key,
        strict=strict,
        mode=mode,
        max_workers=max_workers,
    )