feat: preserve Leica confocal channel identity (dye, excitation, emission) in OME-Zarr

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Leica `.lif` **confocal** channel identity is now preserved in the OME-Zarr
+  output: each channel's fluorophore (dye), excitation/emission wavelengths, and
+  detector are read from the scene metadata and written as OME-XML `<Channel>`
+  elements and omero channel labels (e.g. `ALEXA 594 (590 nm)`) instead of
+  display-LUT color names (`Blue`, `Gray`, …).
 - `size_on_disk(path)` and `format_bytes(n)` helpers in `zarrmony._storage`.
   Handle single files, local directory trees (recursive), and remote fsspec
   URIs; render byte counts as `2.3 GB` style using powers of 1024.

src/zarrmony/api.py

@@ -29,6 +29,11 @@
     ZarrmonyError,
 )
 from zarrmony.metadata.channel_colors import colors_for_channels
+from zarrmony.metadata.lif_channels import (
+    channels_to_ome_channels,
+    channels_to_omero,
+    extract_channels,
+)
 from zarrmony.metadata.model import UserMetadata
 from zarrmony.naming import resolve_scene_dirnames
 from zarrmony.readers.default import derive_bioio_distribution
@@ -191,6 +196,137 @@ def _channels_for_scene(
     ]
 
 
+def _scene_channel_count(reader: Any) -> int:
+    """The current scene's C size, from the reader's public xarray surface.
+
+    Mirrors how :func:`inspect` reads dims/sizes (no coupling to writer
+    internals). A scene with no ``C`` dim has one implicit channel.
+    """
+    xarr = reader.xarray_dask_data
+    return int(xarr.sizes["C"]) if "C" in xarr.dims else 1
+
+
+def _lif_scene_root_fast(reader: Any) -> ET.Element | None:
+    """The current scene's settings XML via bioio-lif's ``scene_root`` fast path.
+
+    ``scene_root`` is bioio-lif's *plate* row/column locator: it returns the
+    well's ``<Element>`` node for plate scenes but RAISES ``ValueError`` ("Row or
+    column value is missing…") for ordinary non-plate confocal scenes, and may be
+    absent on non-LIF readers. We read it inside a guard so either outcome —
+    raise or ``None`` — is a clean miss, not a crash, letting the caller fall
+    back to the document-order ``<Image>`` locator.
+    """
+    try:
+        return getattr(reader, "scene_root", None)
+    except Exception:  # noqa: BLE001 — a raise here just means "not this path"
+        return None
+
+
+def _lif_scene_image(reader: Any) -> ET.Element | None:
+    """The current scene's ``<Image>`` element from the full LIF XML, fail-safe.
+
+    This is the confocal locator from bioio-lif PR #52: ``reader.metadata`` is
+    the whole-document LIF ``ElementTree``; ``.//Image`` returns the per-scene
+    ``<Image>`` elements in scene order, and ``reader.current_scene_index``
+    selects the one being converted. Each such element carries that scene's
+    ``ChannelDescription`` + ``LDM_Block_Sequential_List`` — exactly what
+    :func:`extract_channels` consumes.
+
+    Every reader-surface access is guarded so no partially-readable reader can
+    raise out of here: missing/``None`` metadata, a metadata object without
+    ``findall``, an empty ``<Image>`` list, and an absent or out-of-range
+    ``current_scene_index`` all return ``None`` (a clean miss). Returns the
+    located element or ``None``; never raises.
+    """
+    metadata = getattr(reader, "metadata", None)
+    if metadata is None or not hasattr(metadata, "findall"):
+        return None
+    images = metadata.findall(".//Image")
+    if not images:
+        return None
+    index = getattr(reader, "current_scene_index", None)
+    if not isinstance(index, int) or not (0 <= index < len(images)):
+        return None
+    return images[index]
+
+
+def _lif_scene_channels(reader: Any) -> tuple[list[dict] | None, list[Channel] | None]:
+    """The LIF-vs-other decision, in one place.
+
+    Returns ``(extracted, omero_channels)`` when ``reader`` is a LIF reader whose
+    current scene yields channel identities AND that identity count matches the
+    scene's C size; otherwise ``(None, None)``. ``extracted`` is the raw
+    per-channel identity dicts (handed to :func:`_lif_ome_image` once scene sizes
+    are known); ``omero_channels`` is the display label/color list for
+    :func:`write_scene`. Both projections come from the same vetted extraction,
+    so omero and the OME-XML never disagree.
+
+    Locating the scene XML is two-tier (see the two helpers above):
+
+    1. ``scene_root`` — bioio-lif's plate row/column locator. It works for plate
+       wells but RAISES for ordinary non-plate **confocal** scenes, so it is only
+       a *fast path* that preserves existing plate behavior.
+    2. the document-order ``<Image>`` locator (``.//Image[current_scene_index]``,
+       per bioio-lif PR #52) — the fallback that makes confocal scenes work.
+
+    The count check matters: the omero block and OME-XML ``Pixels/@SizeC`` must
+    describe the channels the array actually has. If extraction and data
+    disagree in EITHER direction — too few or too many identities (corruption,
+    partial metadata) — we decline the projection rather than mislabel the image.
+
+    Fail-safe: any failure — not a LIF reader, no/garbage scene XML, an empty or
+    count-mismatched extraction, or any unexpected reader-surface error — returns
+    ``(None, None)`` so callers fall back cleanly to the existing name-based
+    path. Metadata never crashes a conversion.
+    """
+    try:
+        # First-try fast path (plate wells), then the confocal fallback. Each
+        # locator is internally guarded, so this never raises; an unlocatable
+        # scene simply yields ``None`` here and a clean decline below. Test
+        # ``is None`` explicitly rather than ``a or b`` — an ``Element`` with no
+        # children is falsy, so ``or`` would wrongly skip a valid childless
+        # ``scene_root`` and is deprecated besides.
+        scene_element = _lif_scene_root_fast(reader)
+        if scene_element is None:
+            scene_element = _lif_scene_image(reader)
+        if scene_element is None:
+            return None, None
+        scene_xml = ET.tostring(scene_element, encoding="unicode")
+        extracted = extract_channels(scene_xml)
+        if not extracted or len(extracted) != _scene_channel_count(reader):
+            return None, None
+        omero_channels = [
+            Channel(label=o["label"], color=o["color"])
+            for o in channels_to_omero(extracted)
+        ]
+        return extracted, omero_channels
+    except Exception:  # noqa: BLE001 — never break a conversion over metadata
+        return None, None
+
+
+def _lif_ome_image(
+    extracted: list[dict], scene_index: int, name: str, scene_record: dict
+) -> Image | None:
+    """A real per-scene OME ``Image`` carrying the extracted channel identities.
+
+    Sizes come from ``scene_record`` (exactly as :func:`_stub_image` does); the
+    canonical ``<Channel>`` elements come from :func:`channels_to_ome_channels`.
+    ``extracted`` was already count-checked against the scene's C size in
+    :func:`_lif_scene_channels`; the ``SizeC`` assertion here is belt-and-
+    suspenders. Returns ``None`` on any surprise so the caller falls back to the
+    stub Image (with name-based omero, the existing behavior).
+    """
+    try:
+        image = _stub_image(scene_index, name, scene_record)
+        ome_channels = channels_to_ome_channels(extracted)
+        if len(ome_channels) != image.pixels.size_c:
+            return None
+        image.pixels.channels = ome_channels
+        return image
+    except Exception:  # noqa: BLE001 — never break a conversion over metadata
+        return None
+
+
 def _source_xml_filename(input_path: str | Path) -> str:
     ext = Path(str(input_path)).suffix.lstrip(".").lower()
     return f"raw.{ext}.xml" if ext else "raw.xml"
@@ -432,7 +568,16 @@ def _convert_per_scene(
             )
             continue
 
-        channels = _channels_for_scene(reader, channel_colors)
+        # LIF-vs-other decision in one place: for a LIF scene with extractable
+        # channel identities, drive both omero (label/color) and the canonical
+        # OME-XML <Channel>s from that identity; otherwise fall through to the
+        # existing name-based path. (None, None) for everything non-LIF.
+        lif_extracted, lif_omero_channels = _lif_scene_channels(reader)
+        channels = (
+            lif_omero_channels
+            if lif_omero_channels is not None
+            else _channels_for_scene(reader, channel_colors)
+        )
 
         scene_record = write_scene(
             reader,
@@ -451,16 +596,23 @@ def _convert_per_scene(
             scene_user_md = per_scene_user_metadata[scene_name]
             scene_record["user_metadata"] = scene_user_md
 
-        ome_image, warning = _try_get_ome_image(reader, scene_index)
         metadata_warnings: list[dict] = []
-        if warning is not None:
-            metadata_warnings.append(warning)
-            warnings.warn(
-                f"scene {scene_index} ({scene_name}): {warning['error']}",
-                ExtractorWarning,
-                stacklevel=2,
+        ome_image: Image | None = None
+        if lif_extracted is not None:
+            ome_image = _lif_ome_image(
+                lif_extracted, scene_index, scene_name, scene_record
             )
-            ome_image = _stub_image(scene_index, scene_name, scene_record)
+        if ome_image is None:
+            # Non-LIF, or the LIF Image couldn't be built — existing behavior.
+            ome_image, warning = _try_get_ome_image(reader, scene_index)
+            if warning is not None:
+                metadata_warnings.append(warning)
+                warnings.warn(
+                    f"scene {scene_index} ({scene_name}): {warning['error']}",
+                    ExtractorWarning,
+                    stacklevel=2,
+                )
+                ome_image = _stub_image(scene_index, scene_name, scene_record)
 
         ome_xml = build_ome_xml_for_scene(ome_image)
         write_per_scene_metadata(