First commit for lib_channels (ref #211)

fractal_tasks_core/lib_channels.py

@@ -0,0 +1,55 @@
+from typing import Dict
+from typing import Sequence
+
+
+def get_channel(
+    *, channels: Sequence[Dict], label: str = None, wavelength_id: str = None
+):
+    """
+    Find matching channel in a list
+
+    Find the channel that has the required values of ``label`` and/or
+    ``wavelength_id``, and identify its positional index (which also
+    corresponds to its index in the zarr array).
+
+    :param channels: A list of channel dictionary, where each channel includes
+                     (at least) the ``label`` and ``wavelength_id`` keys
+    :param label: The label to look for in the list of channels.
+    :param wavelength_id: The wavelength_id to look for in the list of
+                          channels.
+
+    """
+    if label:
+        if wavelength_id:
+            matching_channels = [
+                c
+                for c in channels
+                if (
+                    c["label"] == label and c["wavelength_id"] == wavelength_id
+                )
+            ]
+        else:
+            matching_channels = [c for c in channels if c["label"] == label]
+    else:
+        if wavelength_id:
+            matching_channels = [
+                c for c in channels if c["wavelength_id"] == wavelength_id
+            ]
+        else:
+            raise ValueError(
+                "get_channel requires at least one in {label,wavelength_id} "
+                "arguments"
+            )
+
+    if len(matching_channels) > 1:
+        raise ValueError(f"Inconsistent set of channels: {channels}")
+    elif len(matching_channels) == 0:
+        raise ValueError(
+            f"No channel found in {channels} for {label=} and {wavelength_id=}"
+        )
+
+    channel = matching_channels[0]
+    label = channel["label"]
+    wavelength_id = channel["wavelength_id"]
+    array_index = channels.index(channel)
+    return label, wavelength_id, array_index

tests/data/omero/channels_list.json

@@ -0,0 +1,10 @@
+[
+    {
+        "label": "label_1",
+        "wavelength_id": "wavelength_id_1"
+    },
+    {
+        "label": "label_2",
+        "wavelength_id": "wavelength_id_2"
+    }
+]

tests/test_unit_channels_addressing.py

@@ -0,0 +1,30 @@
+import json
+from pathlib import Path
+
+from devtools import debug
+
+from fractal_tasks_core.lib_channels import get_channel
+
+
+def test_get_channel(testdata_path: Path):
+    with (testdata_path / "omero/channels_list.json").open("r") as f:
+        omero_channels = json.load(f)
+    debug(omero_channels)
+
+    label, wl_id, index = get_channel(channels=omero_channels, label="label_1")
+    assert wl_id == "wavelength_id_1"
+    assert index == 0
+    label, wl_id, index = get_channel(
+        channels=omero_channels, wavelength_id="wavelength_id_2"
+    )
+    assert label == "label_2"
+    assert index == 1
+
+    label, wl_id, index = get_channel(
+        channels=omero_channels,
+        label="label_2",
+        wavelength_id="wavelength_id_2",
+    )
+    assert label == "label_2"
+    assert wl_id == "wavelength_id_2"
+    assert index == 1