feat(extraction): support flexible channel names

src/aliby/tile/tiler.py

@@ -557,18 +557,12 @@ class Tiler(ProcessABC):
         item: string
             The channel
         """
-
-        # pattern = re.compile(f"*{item}*", re.IGNORECASE)
-        for i, ch in enumerate(self.channels):
-            found = re.match(item, ch, re.IGNORECASE)
-            if found:
-                if len(found.string) - (found.endpos - found.start()):
-                    print(f"WARNING: channel {item} matched {ch} using regex")
-                return i
-
-        raise Warning(
-            f"Reference channel {item} not in the available channels: {self.channels}"
-        )
+        channel = find_channel_index(self.channels, item)
+        if channel is None:
+            raise Warning(
+                f"Reference channel {channel} not in the available channels: {self.channels}"
+            )
+        return channel
 
     @staticmethod
     def ifoob_pad(full, slices):
@@ -610,3 +604,24 @@ class Tiler(ProcessABC):
                 # pad tile with median value of trap image
                 trap = np.pad(trap, [[0, 0]] + padding.tolist(), "median")
         return trap
+
+
+def find_channel_index(image_channels: t.List[str], channel: str):
+    """
+    Access
+    """
+    for i, ch in enumerate(image_channels):
+        found = re.match(channel, ch, re.IGNORECASE)
+        if found:
+            if len(found.string) - (found.endpos - found.start()):
+                print(f"WARNING: channel {channel} matched {ch} using regex")
+            return i
+
+
+def find_channel_name(image_channels: t.List[str], channel: str):
+    """
+    Find the name of the channel according to a given channel regex.
+    """
+    index = find_channel_index(image_channels, channel)
+    if index is not None:
+        return image_channels[index]

src/extraction/core/extractor.py

@@ -579,7 +579,7 @@ class Extractor(ProcessABC):
         """
         if channels is None:
             channels = (*self.params.tree,)
-        if channel in channels:
+        if channel in channels:  # TODO start here to fetch channel using regex
             return traps[:, channels.index(channel), 0]
         elif channel in self.img_bgsub:
             return self.img_bgsub[channel]

src/extraction/core/functions/defaults.py

 # File with defaults for ease of use
+import re
 import typing as t
 from pathlib import PosixPath
 
@@ -9,7 +10,8 @@ def exparams_from_meta(
     meta: t.Union[dict, PosixPath, str], extras: t.Collection[str] = ["ph"]
 ):
     """
-    Obtain parameters from metadata of hdf5 file
+    Obtain parameters from metadata of hdf5 file.
+    It compares a list of candidate channels using case-inspecific REGEX to identify valid channels.
     """
     meta = meta if isinstance(meta, dict) else load_attributes(meta)
     base = {
@@ -23,6 +25,7 @@ def exparams_from_meta(
         "GFPFast",
         "mCherry",
         "pHluorin405",
+        "pHluorin488",
         "Flavin",
         "Cy5",
         "mKO2",
@@ -35,32 +38,61 @@ def exparams_from_meta(
         "std",
         "imBackground",
         "max5px",
-        "nuc_est_conv",
+        # "nuc_est_conv",
     }
 
-    default_rm = {r: default_metrics for r in default_reductions}
+    # Defined ratiometric combinations that can be used as ratio
+    # key is numerator and value is denominator; add more to support additional channel names
+    ratiometric_combinations = {"phluorin405": ("phluorin488", "gfpfast")}
+
+    default_reduction_metrics = {
+        r: default_metrics for r in default_reductions
+    }
     # default_rm["None"] = ["nuc_conv_3d"] # Uncomment this to add nuc_conv_3d (slow)
 
-    av_flch = candidate_channels.intersection(
-        meta["channels/channel"]
-    ).difference({"Brightfield", "DIC", "BrightfieldGFP"})
+    from aliby.tile.tiler import find_channel_name
+
+    extant_fluorescence_ch = []
+    for av_channel in candidate_channels:
+        # Find channels in metadata whose names match
+        found_channel = find_channel_name(meta["channels/channel"], av_channel)
+        if found_channel is not None:
+            extant_fluorescence_ch.append(found_channel)
 
-    for ch in av_flch:
-        base["tree"][ch] = default_rm
+    for ch in extant_fluorescence_ch:
+        base["tree"][ch] = default_reduction_metrics
 
-    base["sub_bg"] = av_flch
+    base["sub_bg"] = extant_fluorescence_ch
 
     # Additional extraction defaults when channels available
     if "ph" in extras:
-        if {"pHluorin405", "GFPFast"}.issubset(av_flch):
+        # SWAINLAB-specific names
+        # find first valid combination of ratiometric fluorescence channels
+        numerator_channel, denominator_channel = (None, None)
+        for ch1, chs2 in ratiometric_combinations.items():
+            found_channel1 = find_channel_name(extant_fluorescence_ch, ch1)
+            if found_channel1 is not None:
+                numerator_channel = found_channel1
+                for ch2 in chs2:
+                    found_channel2 = find_channel_name(
+                        extant_fluorescence_ch, ch2
+                    )
+                    if found_channel2:
+                        denominator_channel = found_channel2
+                        break
 
+        # If two compatible pHluorin channels are available
+        if numerator_channel is not None and denominator_channel is not None:
             sets = {
                 b + a: (x, y)
                 for a, x in zip(
                     ["", "_bgsub"],
                     (
-                        ["GFPFast", "pHluorin405"],
-                        ["GFPFast_bgsub", "pHluorin405_bgsub"],
+                        [numerator_channel, denominator_channel],
+                        [
+                            f"{numerator_channel}_bgsub",
+                            f"{denominator_channel}_bgsub",
+                        ],
                     ),
                 )
                 for b, y in zip(["em_ratio", "gsum"], ["div0", "add"])
@@ -68,7 +100,7 @@ def exparams_from_meta(
             for i, v in sets.items():
                 base["multichannel_ops"][i] = [
                     *v,
-                    default_rm,
+                    default_reduction_metrics,
                 ]
 
     return base