Converted tags to regexes

metisp/pymetis/src/pymetis/base/impl.py

@@ -48,6 +48,7 @@ def __init__(self, recipe: 'MetisRecipe') -> None:
         self.inputset = None
         self.frameset = None
         self.header = None
+        self.products: Dict[str, PipelineProduct] = {}
         self.product_frames = cpl.ui.FrameSet()
 
     def run(self, frameset: cpl.ui.FrameSet, settings: Dict[str, Any]) -> cpl.ui.FrameSet:

metisp/pymetis/src/pymetis/inputs/base.py

@@ -18,6 +18,8 @@
 """
 
 from abc import abstractmethod
+import re
+from typing import Pattern
 
 import cpl
 
@@ -27,7 +29,7 @@
 class PipelineInput:
     _title: str = None                      # No universal title makes sense
     _required: bool = True                  # By default, inputs are required to be present
-    _tags: [str] = None                     # No universal tags are provided
+    _tags: Pattern = None                   # No universal tags are provided
     _group: str = None                      # No sensible default, must be provided explicitly
 
     @property
@@ -49,7 +51,7 @@ def group(self):
     def __init__(self,
                  *,
                  title: str = None,
-                 tags: [str] = None,
+                 tags: Pattern = None,
                  required: bool = None,
                  **kwargs):
         # First override the title, if provided in the constructor
@@ -66,30 +68,21 @@ def __init__(self,
             self._tags = tags
             Msg.debug(self.__class__.__qualname__, f"Overriding `tags` to {self.tags}")
 
-        # Now expand the tags with provided context from **kwargs
-        try:
-            self._tags = [tag.format(**kwargs) for tag in self._tags]
-        except KeyError as e:
-            Msg.error(self.__class__.__qualname__, f"Could not substitute tag placeholders: {e}")
-            raise e
-
         # Check if tags are defined...
         if not self.tags:
-            raise NotImplementedError(f"Pipeline input {self.__class__.__qualname__} has no defined tags")
+            raise NotImplementedError(f"Pipeline input {self.__class__.__qualname__} has no defined tag pattern")
 
-        # ...and that they are a list of strings (not a single string -- this leads to nasty errors)
-        if not isinstance(self.tags, list):
-            raise TypeError(f"Tags must be a list of string templates, got '{self.tags}'")
-        for tag in self.tags:
-            if not isinstance(tag, str):
-                raise TypeError(f"Tags must be a list of string templates, got '{type(tag)}'")
+        # ...and that they are a re pattern
+        if not isinstance(self.tags, re.Pattern):
+            raise TypeError(f"Tags must be a `re` pattern, got '{self.tags}'")
 
         # Override `required` if requested
         if required is not None:
             self._required = required
             Msg.debug(self.__class__.__qualname__, f"Overriding `required` to {self.required}")
 
-        # Check is frame_group is defined (if not, this gives rise to strange errors deep within CPL)
+        # Check is frame_group is defined (if not, this gives rise to strange errors deep within CPL
+        # that you really do not want to deal with)
         if not self.group:
             raise NotImplementedError(f"Pipeline input {self.__class__.__qualname__} has no defined group!")
 
@@ -121,7 +114,7 @@ def __init__(self,
         super().__init__(tags=tags, required=required, **kwargs)
 
         for frame in frameset:
-            if frame.tag in self.tags:
+            if self.tags.fullmatch(frame.tag):
                 if self.frame is None:
                     Msg.debug(self.__class__.__qualname__,
                               f"Found a {self.title} frame: {frame.file}.")
@@ -130,8 +123,14 @@ def __init__(self,
                                 f"Found another {self.title} frame: {frame.file}! "
                                 f"Discarding previously loaded {self.frame.file}.")
                 self.frame = frame
+            else:
+                Msg.debug(self.__class__.__qualname__,
+                          f"Ignoring {frame.file}: tag {frame.tag} does not match.")
 
     def verify(self):
+        """
+        Run all the required instantiation time checks
+        """
         self._verify_frame_present(self.frame)
 
     def _verify_frame_present(self,
@@ -159,21 +158,21 @@ class MultiplePipelineInput(PipelineInput):
     def __init__(self,
                  frameset: cpl.ui.FrameSet,
                  *,
-                 tags: [str] = None,
+                 tags: Pattern = None,
                  required: bool = None,
                  **kwargs):                     # Any other args
         self.frameset: cpl.ui.FrameSet | None = cpl.ui.FrameSet()
         super().__init__(tags=tags, required=required, **kwargs)
 
         for frame in frameset:
-            if frame.tag in self.tags:
+            if self.tags.fullmatch(frame.tag):
                 frame.group = self.group
                 self.frameset.append(frame)
                 Msg.debug(self.__class__.__qualname__,
                           f"Found a {self.title} frame: {frame.file}.")
             else:
                 Msg.debug(self.__class__.__qualname__,
-                          f"Ignoring {frame.file}: tag {frame.tag} not in {self.tags}.")
+                          f"Ignoring {frame.file}: tag {frame.tag} does not match.")
 
 
     def verify(self):
@@ -189,7 +188,7 @@ def _verify_frameset_not_empty(self) -> None:
             if self.required:
                 raise cpl.core.DataNotFoundError(f"No {self.title} frames found in the frameset.")
             else:
-                Msg.debug(f"No {self.title} frames found but not required.")
+                Msg.debug(self.__class__.__qualname__, f"No {self.title} frames found but not required.")
         else:
             Msg.debug(self.__class__.__qualname__, f"OK: {count} frames found")
 
@@ -222,7 +221,7 @@ def _verify_same_detector(self) -> None:
                                      }[det])
                 except KeyError as e:
                     raise KeyError(f"Invalid detector name! In {frame.file}, ESO DPR TECH is '{det}'") from e
-            except KeyError as e:
+            except KeyError:
                 Msg.warning(self.__class__.__qualname__, f"No detector (ESO DPR TECH) set!")
 
 

metisp/pymetis/src/pymetis/inputs/common.py

@@ -17,8 +17,9 @@
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 """
 
+import re
 import cpl
-from cpl.core import Msg
+from typing import Pattern
 
 from .base import MultiplePipelineInput, SinglePipelineInput
 
@@ -52,36 +53,36 @@ class RawInput(MultiplePipelineInput):
 
 class MasterDarkInput(SinglePipelineInput):
     _title: str = "master dark"
-    _tags: [str] = ["MASTER_DARK_{det}"]
+    _tags: Pattern = re.compile(r"MASTER_DARK_(?P<detector>2RG|GEO|IFU)")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB
 
 
 class MasterFlatInput(SinglePipelineInput):
     _title: str = "master flat"
-    _tags: [str] = ["MASTER_IMG_FLAT_LAMP_{det}"]
+    _tags: Pattern = re.compile(r"MASTER_IMG_FLAT_LAMP_(?P<detector>2RG|GEO|IFU)")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB
 
 
 class LinearityInput(SinglePipelineInput):
     _title: str = "linearity map"
-    _tags: [str] = ["LINEARITY_{det}"]
+    _tags: Pattern = re.compile(r"LINEARITY_(?P<detector>2RG|GEO|IFU)")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB
 
 
 class BadpixMapInput(SinglePipelineInput):
     _title: str = "bad pixel map"
-    _tags: [str] = ["BADPIX_MAP_{det}"]
+    _tags: Pattern = re.compile(r"BADPIX_MAP_(?P<detector>2RG|GEO|IFU)")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB
 
 
 class PersistenceMapInput(SinglePipelineInput):
     _title: str = "persistence map"
-    _tags: [str] = ["PERSISTENCE_MAP"]
+    _tags: Pattern = re.compile(r"PERSISTENCE_MAP")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB
     _required: bool = False     # Persistence maps are usually optional (can be overridden)
 
 
 class GainMapInput(SinglePipelineInput):
     _title: str = "gain map"
-    _tags = ["GAIN_MAP_{det}"]
+    _tags: Pattern = re.compile(r"GAIN_MAP_(?P<detector>2RG|GEO|IFU)")
     _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.CALIB

metisp/pymetis/src/pymetis/prefab/flat.py

@@ -17,6 +17,7 @@
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 """
 
+import re
 from abc import ABC
 from typing import Dict
 
@@ -39,7 +40,7 @@ class RawFlatInput(RawInput):
             """
             A subclass of RawInput that is handling the flat image raws.
             """
-            _tags = ["{band}_FLAT_LAMP_RAW", "{band}_FLAT_TWILIGHT_RAW"]
+            _tags = re.compile(r"(?P<band>(LM|N))_FLAT_(?P<target>LAMP|TWILIGHT)_RAW")
 
         class DarkFlatInput(MasterDarkInput):
             """
@@ -48,8 +49,8 @@ class DarkFlatInput(MasterDarkInput):
             pass
 
         def __init__(self, frameset):
-            self.raw = self.RawFlatInput(frameset, band=self.band)
-            self.master_dark = MasterDarkInput(frameset, det=self.detector)
+            self.raw = self.RawFlatInput(frameset)
+            self.master_dark = MasterDarkInput(frameset)
 
             self.inputs = [self.raw, self.master_dark]
 

metisp/pymetis/src/pymetis/prefab/rawimage.py

@@ -38,10 +38,7 @@ class InputSet(PipelineInputSet):
         detector: str = None
 
         def __init__(self, frameset: cpl.ui.FrameSet):
-            if self.detector is None:
-                raise NotImplementedError(f"{self.__class__.__qualname__} must define the detector")
-
-            self.raw = self.RawInput(frameset, det=self.detector)
+            self.raw = self.RawInput(frameset)
             self.inputs += [self.raw]
             super().__init__(frameset)
 

metisp/pymetis/src/pymetis/recipes/ifu/metis_ifu_reduce.py

@@ -17,8 +17,9 @@
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 """
 
+import re
+
 import cpl
-from cpl.core import Msg
 from typing import Any, Dict, Literal
 
 from pymetis.base import MetisRecipe, MetisRecipeImpl
@@ -43,29 +44,29 @@ class InputSet(DarkImageProcessor.InputSet):
         detector = "IFU"
 
         class RawInput(RawInput):
-            _tags = ["IFU_SCI_RAW", "IFU_STD_RAW"]
+            _tags = re.compile(r"IFU_(?P<target>SCI|STD)_RAW")
 
         class MasterDarkInput(DetectorIfuMixin, MasterDarkInput):
             _group: cpl.ui.Frame.FrameGroup = cpl.ui.Frame.FrameGroup.RAW
 
         class WavecalInput(SinglePipelineInput):
-            _tags = ["IFU_WAVECAL"]
+            _tags = re.compile(r"IFU_WAVECAL")
             _group = cpl.ui.Frame.FrameGroup.CALIB
             _title = "Wavelength calibration"
 
         class DistortionTableInput(SinglePipelineInput):
-            _tags = ["IFU_DISTORTION_TABLE"]
+            _tags = re.compile(r"IFU_DISTORTION_TABLE")
             _group = cpl.ui.Frame.FrameGroup.CALIB
             _title = "Distortion table"
 
         def __init__(self, frameset: cpl.ui.FrameSet):
             """
                 Here we also define all input frames specific for this recipe, except those handled by mixins.
             """
-            self.raw = self.RawInput(frameset, det=self.detector)
-            self.linearity_map = LinearityInput(frameset, det=self.detector)
+            self.raw = self.RawInput(frameset)
+            self.linearity_map = LinearityInput(frameset)
             self.persistence_map = PersistenceMapInput(frameset)
-            self.master_dark = self.MasterDarkInput(frameset, det="IFU")
+            self.master_dark = self.MasterDarkInput(frameset)
             self.ifu_wavecal = self.WavecalInput(frameset)
             self.ifu_distortion_table = self.DistortionTableInput(frameset)
             self.inputs += [self.linearity_map, self.persistence_map, self.master_dark, self.ifu_wavecal, self.ifu_distortion_table]

metisp/pymetis/src/pymetis/recipes/ifu/metis_ifu_telluric.py

@@ -31,27 +31,6 @@ class MetisIfuTelluricImpl(MetisRecipeImpl):
     def detector_name(self) -> str | None:
         return "2RG"
 
-    class Input(RecipeInput):
-        tags_combined = ["IFU_SCI_COMBINED", "IFU_STD_COMBINED"]
-        detector_name = '2RG'
-
-        def __init__(self, frameset: cpl.ui.FrameSet):
-            self.combined: cpl.ui.Frame | None = None
-            super().__init__(frameset)
-
-        def categorize_frame(self, frame: cpl.ui.Frame) -> None:
-            if frame.tag in self.tags_combined:
-                frame.group = cpl.ui.Frame.FrameGroup.RAW # TODO What group is this really?
-                self.combined = self._override_with_warning(self.combined, frame,
-                                                            origin=self.__class__.__qualname__,
-                                                            title="combined")
-                Msg.debug(self.__class__.__qualname__, f"Got IFU science combined frame: {frame.file}.")
-            else:
-                super().categorize_frame(frame)
-
-        def verify(self) -> None:
-            pass
-
     class ProductSciReduced1D(PipelineProduct):
         category = rf"IFU_SCI_REDUCED_1D"
 

metisp/pymetis/src/pymetis/recipes/img/metis_lm_img_background.py

@@ -17,6 +17,7 @@
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 """
 
+import re
 from typing import Dict
 
 import cpl
@@ -32,7 +33,7 @@
 class MetisLmImgBackgroundImpl(MetisRecipeImpl):
     class InputSet(PipelineInputSet):
         class LmBasicReducedInput(SinglePipelineInput):
-            _tags: [str] = ["LM_{target}_BASIC_REDUCED"]
+            _tags: re.Pattern = re.compile(r"LM_(?P<target>SCI|STD)_BASIC_REDUCED")
 
         def __init__(self, frameset: cpl.ui.FrameSet):
             super().__init__(frameset)

metisp/pymetis/src/pymetis/recipes/img/metis_lm_img_basic_reduce.py

@@ -17,6 +17,7 @@
 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 """
 
+import re
 from typing import Dict
 
 import cpl
@@ -57,31 +58,26 @@ class InputSet(DarkImageProcessor.InputSet):
         # RawImageProcessor.InputSet. It already knows that it wants a RawInput and MasterDarkInput class,
         # but does not know about the tags yet. So here we define tags for the raw input
         class Raw(RawInput):
-            _tags = ["LM_IMAGE_SCI_RAW", "LM_IMAGE_STD_RAW"]
+            _tags = re.compile(r"LM_IMAGE_(?P<target>SCI|STD)_RAW")
 
         # Also one master flat is required. We use a prefabricated class
         class MasterFlat(MasterFlatInput):
-            _tags = ["MASTER_IMG_FLAT_LAMP_LM", "MASTER_IMG_FLAT_TWILIGHT_LM"]
+            _tags = re.compile(r"MASTER_IMG_FLAT_(?P<source>LAMP|TWILIGHT)_(?P<band>LM)")
 
         # We could define the master dark explicitly too, but we can use a prefabricated class instead.
         # That already has its tags defined (for master darks it's always "MASTER_DARK_{det}"), so we just define
         # the detector and band. Those are now available for all Input classes here.
         # Of course, we could be more explicit and define them directly.
 
-        detector: str = '2RG'
-        band: str = 'LM'
-
         RawInput = Raw
         MasterDarkInput = MasterDarkInput
 
         def __init__(self, frameset: cpl.ui.FrameSet):
             super().__init__(frameset)
-            self.master_flat = self.MasterFlat(frameset,
-                                               tags=["MASTER_IMG_FLAT_LAMP_{band}", "MASTER_IMG_FLAT_TWILIGHT_{band}"],
-                                               band="LM", det=self.detector)
-            self.linearity = LinearityInput(frameset, det=self.detector)
+            self.master_flat = self.MasterFlat(frameset)
+            self.linearity = LinearityInput(frameset)
             self.persistence = PersistenceMapInput(frameset, required=False)
-            self.gain_map = GainMapInput(frameset, det=self.detector)
+            self.gain_map = GainMapInput(frameset)
 
             # We need to register the inputs (just to be able to do `for x in self.inputs:`)
             self.inputs += [self.master_flat, self.linearity, self.persistence, self.gain_map]