Updated raw ramp file, fixed error in init_steps

liger_iris_pipeline/pipeline/base_pipeline.py

@@ -72,7 +72,6 @@ def __init__(self, config_file : str | None = None, **kwargs):
             if key not in ("class", "steps", "config_file"):
                 _val = self.parse_config_kwarg(key, val, pipeline_spec)
                 setattr(self, key, _val)
-
 
     def init_steps(self, config : config_parser.ConfigObj):
         if self.config_file is not None:
@@ -91,9 +90,9 @@ def init_steps(self, config : config_parser.ConfigObj):
             if step_config_file is not None:
                 step_config = config_parser.load_config_file(config_file=step_config_file)
                 step_kwargs.update(step_kwargs)
-                for key, val in step_config.items():
-                    if key not in ('class', 'config_file'):
-                        step_kwargs[key] = val
+            for key, val in step_config.items():
+                if key not in ('class', 'config_file'):
+                    step_kwargs[key] = val
             new_step = _class(
                 config_file=step_config_file,
                 **step_kwargs
@@ -103,7 +102,6 @@ def init_steps(self, config : config_parser.ConfigObj):
     def input_to_asn(self, input):
         """
         Convert input to an association.
-        TODO: Add case for LigerIRISDataModel
 
         Parameters:
             input (str | Path | LigerIRISAssociation): The input file.

liger_iris_pipeline/pipeline/create_flatfield.py

@@ -68,13 +68,13 @@ def process_exposure_product(self, exp_product : dict):
             data=input_model.data, err=input_model.err, dq=input_model.dq
         )
         _meta = copy.deepcopy(input_model.meta.instance)
-        _meta.update(flat_model.meta.instance)
+        _meta.update(flat_model.meta.instance) # TODO: Check if this is the right way to merge the meta data
         flat_model.meta = _meta
         flat_model.meta.reftype = "FLAT"
         flat_model.meta.pedigree = None
         flat_model.meta.version = '0.0.1'
         flat_model.meta.filename = None
-        #flat_model.meta.data_type = model_result.__class.__name__
+        flat_model.meta.data_type = flat_model.__class__.__name__
 
         self.log.info(f"Finished processing {members_by_type["flat"][0]}")
 

liger_iris_pipeline/pipeline/stage1.py

@@ -40,9 +40,7 @@ def process(self, input):
         # Process each exposure.
         results = []
         for product in self.asn.products:
-            self.log.info("Processing product {}".format(product["name"]))
-            if self.save_results:
-                self.output_file = product["name"]
+            self.log.info(f"Processing product {input}")
             result = self.process_exposure_product(product)
 
             # Save result
@@ -51,8 +49,6 @@ def process(self, input):
 
         self.log.info("Stage1Pipeline completed")
 
-        self.output_use_model = True
-
         return results
 
     # Process each exposure
@@ -64,7 +60,7 @@ def process_exposure_product(self, exp_product):
         """
         members_by_type = self.asn_product_by_types(exp_product)
 
-        science = members_by_type["science"][0]
+        science = members_by_type["sci"][0]
         self.log.info(f"Processing input {science} ...")
         with datamodels.RampModel(science) as input_model:
             input_model = self.nonlinear_correction.run(input_model)

liger_iris_pipeline/readout/fit_ramp_numba.py

@@ -1,8 +1,10 @@
 from numba import jit, njit, prange
 import numpy as np
 
+# TODO: Implement jump detection, look into more sophisticated methods
 
-def fit_ramp_utr_loopdev(times, ramp):
+
+def _fit_ramp_utr_loop_experimental(times, ramp):
     n_reads = len(times)
     for i in range(n_reads):
         xymult = times[i] * ramp[i]
@@ -26,7 +28,6 @@ def fit_ramp_utr(times, ramp):
     slope = (numerator / denominator)
     return slope
 
-
 @njit
 def fit_ramps_utr(times, ramps):
     ny, nx, n_groups, _ = ramps.shape

liger_iris_pipeline/readout/fit_ramp_step.py

@@ -13,8 +13,8 @@
 class FitRampStep(LigerIRISStep):
 
     spec = """
-        method = string(default='utr')  # Ramp fit method. Options are 'utr' and 'mcds'.
-        num_coadd = integer(default=3)
+        method = string(default='utr')  # Ramp fit method. Options are 'utr' and 'mcds'. For 'cds', use 'mcds' and set num_coadd to 1.
+        num_coadd = integer(default=3) # The number of coadds for the 'mcds' method.
     """
 
     class_alias = "ramp_fit"
@@ -43,7 +43,7 @@ def process(self, input):
 
         # TODO: Generalize the conversion from RampModel -> ImagerModel/IFUImageModel
         _meta = copy.deepcopy(input_model.meta.instance)
-        _meta.update(input_model.meta.instance)
+        _meta.update(input_model.meta.instance) # TODO: Check if this is the right way to merge the meta data
         model_result.meta = _meta
         model_result.meta.filename = None
         model_result.meta.data_level = 1

liger_iris_pipeline/tests/test_create_flat.py

@@ -44,16 +44,6 @@ def test_create_flat(tmp_path):
     raw_flat_filename = str(tmp_path / "2024B-P123-008_IRIS_IMG1_FLAT-Y_LVL1_0001-00.fits")
     raw_flat_model.save(raw_flat_filename)
 
-    # ASN
-    asn = L1Association.from_product({
-        "members": [
-            {
-                "expname": raw_flat_filename,
-                "exptype": "flat",
-            },
-        ],
-    })
-
     # Create a temporary config file
     conf = create_config()
     config_file = str(tmp_path / "test_config.cfg")
@@ -62,7 +52,7 @@ def test_create_flat(tmp_path):
 
     # Initialize flatfield pipeline
     pipeline = liger_iris_pipeline.CreateFlatfield(config_file=config_file)
-    flat_model = pipeline.run(asn, output_dir=str(tmp_path))
+    flat_model = pipeline.run(raw_flat_filename, output_dir=str(tmp_path))
 
     # Open dark
     dark_model = datamodels.open(pipeline.dark_sub.dark_filename)

liger_iris_pipeline/tests/test_stage1.py

@@ -6,15 +6,14 @@
 
 def create_config():
     conf = """
-    name = "ImagerStage1Pipeline"
-    class = "liger_iris_pipeline.ImagerStage1Pipeline"
+    class = "liger_iris_pipeline.Stage1Pipeline"
     save_results = True
 
     [steps]
         [[nonlinear_correction]]
             skip = False
         [[ramp_fit]]
-            method = "mcds"
+            method = "utr"
     """
     return conf
 
@@ -43,42 +42,35 @@ def test_imager_stage1(tmp_path):
     source = np.full((10, 10), 1000)
 
     # Create a ramp model
-    ramp_model = create_ramp(source, meta, readtime=1, n_reads_per_group=10, n_groups=5, nonlin_coeffs = None)
+    ramp_model = create_ramp(source, meta, readtime=1, n_reads_per_group=10, n_groups=5, nonlin_coeffs = None, noise=False)
     ramp_filename = str(tmp_path / "2024B-P123-008_IRIS_IMG1_SCI-Y_LVL0_0001-00.fits")
     ramp_model.save(ramp_filename)
-
-    # Save the ramp model
-    asn = L0Association.from_product({
-        "name": "Test",
-        "members": [
-            {
-                "expname": ramp_filename,
-                "exptype": "science",
-            },
-        ]
-    })
 
     # Create a temporary config file
     conf = create_config()
     config_file = str(tmp_path / "test_config.cfg")
     with open(config_file, "w") as f:
         f.write(conf)
 
-    # Create and call the pipeline object
+    # Create the pipeline
     pipeline = liger_iris_pipeline.Stage1Pipeline(config_file=config_file, output_dir=str(tmp_path))
-    results = pipeline.run(asn)
-    model_result = results[0]
 
-    # Test MCDS
+    # Test UTR
+    pipeline.ramp_fit.method = "utr"
+    results = pipeline.run(ramp_filename)
+    model_result = results[0]
     np.testing.assert_allclose(model_result.data, source, rtol=1e-6)
-
-    conf = conf.replace('mcds', 'utr')
-    with open(config_file, "w") as f:
-        f.write(conf)
 
-    pipeline = liger_iris_pipeline.Stage1Pipeline(config_file=config_file, output_dir=str(tmp_path))
-    results = pipeline.run(asn)
+    # Test MCDS
+    pipeline.ramp_fit.method = "mcds"
+    pipeline.ramp_fit.num_coadd = 3
+    results = pipeline.run(ramp_filename)
     model_result = results[0]
+    np.testing.assert_allclose(model_result.data, source, rtol=1e-6)
 
-    # Test UTR
+    # Test CDS
+    pipeline.ramp_fit.method = "mcds"
+    pipeline.ramp_fit.num_coadd = 1
+    results = pipeline.run(ramp_filename)
+    model_result = results[0]
     np.testing.assert_allclose(model_result.data, source, rtol=1e-6)

liger_iris_pipeline/tests/test_utils.py

@@ -175,7 +175,9 @@ def create_ramp(
         meta : dict,
         readtime : float, # e- RMS
         n_reads_per_group : int, n_groups : int,
+        read_noise : float = 0, kTC_noise : float = 0,
         nonlin_coeffs : np.ndarray | None = None,
+        noise : bool = False,
     ):
     times = np.zeros(shape=(n_groups, n_reads_per_group), dtype=float)
     data = np.zeros(shape=(source.shape[0], source.shape[1], n_groups, n_reads_per_group), dtype=np.int16)
@@ -186,6 +188,11 @@ def create_ramp(
             image = data[:, :, i, j-1] + source * readtime
             if nonlin_coeffs is not None:
                 image *= np.round(np.polyval(nonlin_coeffs[::-1], t).astype(np.int16))
+            if noise:
+                if j == 0:
+                    image += np.random.normal(loc=0, scale=np.sqrt(source * t + read_noise**2 + kTC_noise**2), size=source.shape)
+                else:
+                    image += np.random.normal(loc=0, scale=np.sqrt(source * t + read_noise**2), size=source.shape)
             times[i, j] = t
             data[:, :, i, j] = image.copy()
     np.clip(data, 0, np.iinfo(np.int16).max)