Merge pull request spacetelescope#69 from rgcosentino/B21_integral_nonlin_update_rgc

B21 integral nonlin update rgc

Author: rgcosentino

src/wfi_reference_pipeline/reference_types/integral_non_linearity/integral_non_linearity.py

@@ -1,5 +1,3 @@
-
-
 import numpy as np
 import roman_datamodels.stnode as rds
 
@@ -17,30 +15,24 @@ class IntegralNonLinearity(ReferenceType):
     method creates the asdf reference file.
 
     Example creation
-    from wfi_reference_pipeline.reference_types.integral_non_linearity.integral_non_linearity import simulate_inl_correction_array
     from wfi_reference_pipeline.resources.make_dev_meta import MakeDevMeta
     from wfi_reference_pipeline.reference_types.integral_non_linearity.integral_non_linearity import IntegralNonLinearity
 
-    for det in range(1, 19):
-        arr = simulate_inl_correction_array()
-        # Create meta data object for INL ref file
-        tmp = MakeDevMeta(ref_type='INTEGRALNONLINEARITY')
-        # Update meta per detector to get the right values from the form and update description
-        tmp.meta_integral_non_linearity.instrument_detector = f"WFI{det:02d}"
-        tmp.meta_integral_non_linearity.description = (
-            "To support new integral non-linearity correction development for B21. "
-            "The integral non-linearity correction is applied to each channel, "
-            "numbered left to right from 1 to 32, with each channel containing 128 "
-            "pixels in length."
-            )
-        # Update the file name to match the detector
-        fl_name = 'new_roman_inl_' + tmp.meta_integral_non_linearity.instrument_detector
-        # Instantiate an object and write the file out
-        rfp_inl = IntegralNonLinearity(meta_data=tmp.meta_integral_non_linearity,
-                                    ref_type_data=arr,
-                                    clobber=True,
-                                    outfile=fl_name+'.asdf')
-        rfp_inl.generate_outfile()
+    arr - assumed here to be a numpy array that is derived in the instrument coordinate reference frame and properly
+    transformed into the science coordinate reference frame for ingestion into the class.
+
+    tmp = MakeDevMeta(ref_type='INTEGRALNONLINEARITY')
+    rfp_inl = IntegralNonLinearity(meta_data=tmp.meta_integral_non_linearity,
+                                ref_type_data=arr)
+    rfp_inl.generate_outfile()
+
+    Documentation and important notes. This effect comes from the Analog to Digital converters that read out the detector
+    array. The dimensions of the detector are 4096x4096 with 32 A/D converters and amplifiers reading out every 128 pixels
+    together at the same time. The correction implemented adjust the measured value to the actual value which varies across
+    all possible values in UNIT16 from 0 to 65535. This module expects that the transformation from instrument to science 
+    coordinate reference system has already taken place when the input data is passed into the class as ref_type_data.
+    
+    See https://roman-docs.stsci.edu/data-handbook/wfi-data-levels-and-products/coordinate-systems
     """
 
     def __init__(
@@ -96,10 +88,52 @@ def __init__(
         if len(self.meta_data.description) == 0:
             self.meta_data.description = "Roman WFI integral non linearity reference file."
 
-        self.inl_correction = ref_type_data
-        _, num_values = np.shape(ref_type_data)
-        self.value_array = np.linspace(0, 65535, num_values, dtype=np.uint16)
-        # TODO look at references for channel id number - https://roman-docs.stsci.edu/data-handbook-home/wfi-data-format/coordinate-systems
+        if ref_type_data is None and file_list is None:
+            msg = "RFP is simulating the INL correction"
+            print(msg)
+
+            ref_type_data = np.asarray(simulate_inl_correction_array())
+
+        # If INL correction arrays are provded as input into class then check everything needed
+        # for the array to be valid.
+        elif ref_type_data is not None:
+            # Convert to numpy array and enforce dtype in one go
+            ref_type_data = np.asarray(ref_type_data, dtype=np.float64)
+
+            # Print message only for detectors that need LR flip
+            flip_lr_detectors = {
+                "WFI03", "WFI06", "WFI09",
+                "WFI12", "WFI15", "WFI18"
+            }
+            if self.meta_data.instrument_detector in flip_lr_detectors:
+                msg = (
+                    "RFP using input ref_type_data assumed to be transformed into the "
+                    "proper science coordinate reference frame."
+                )
+                print(msg)
+
+            # Validate array shape
+            if ref_type_data.ndim != 2:
+                raise ValueError(
+                    "IntegralNonLinearity expects ref_type_data to be a 2D array "
+                    "with shape (32, 65536)."
+                )
+
+            n_chan, n_val = ref_type_data.shape
+            if n_chan != 32 or n_val != 65536:
+                raise ValueError(
+                    "Invalid INL correction array shape. "
+                    f"Expected (32, 65536), got ({n_chan}, {n_val})."
+                )
+
+            # Set attributes
+            self.inl_correction = ref_type_data
+            self.value_array = np.linspace(0, 65535, n_val, dtype=np.uint16)
+
+        elif file_list is not None:
+            raise ValueError(
+                    "Module currently not capable to support file list input."
+                    )
 
         self.outfile = outfile
 
@@ -117,44 +151,41 @@ def update_data_quality_array(self):
 
     def _make_inl_table(self):
         """
-        Construct the integral non linearity correction table to populate the data model.
-        """
-        inl_table = {}
-
-        # Assuming self.inl_correction is shaped (32, N)
-        # Each row corresponds to a science chunk (0–31)
-        for sci_chunk in range(32):
-            # Reverse channel mapping: 0→32, 1→31, ..., 31→1 for a detector that is
-            # needing to be flipped left right from detector to science coordinates
-            # NOTE: other detectors
-            channel = 32 - sci_chunk
-
-            inl_table[str(sci_chunk)] = {
-                'channel': channel,
-                'correction': self.inl_correction[sci_chunk]
-            }
+        Populate the INL table following the Roman INL reference schema.
 
-        return inl_table
+        This method is explicit to map instrument channel number and index in the
+        instrument coordinate reference frame to the science channel number and index
+        in the science coorcinate reference frame. 
 
+        Science channels are always numbered from bottom left to right for each detector
+        from 1-32. The science coordinate reference frame is how the pixels on the detector
+        look at the sky - after the hardware has been placed and some rotated to account
+        for the position of detector electronics. 
+
+        Instrument channels are indexed from 0-31 with the origin including detector electronics
+        always in the lower left. The transformation from instrument to science coordinates is
+        illustrated in https://roman-docs.stsci.edu/data-handbook/wfi-data-levels-and-products/coordinate-systems
+
+        """
+        table = {}
+        for science_chan in range(32):
+            key = f"science_channel_{science_chan+1:02d}"
+            table[key] = {
+                "instrument_channel": science_chan,
+                "correction": self.inl_correction[science_chan],
+            }
+        return table
 
     def populate_datamodel_tree(self):
         """
         Build the Roman datamodel tree for the integral non-linearity reference.
         """
-        try:
-            # Placeholder until official datamodel exists
-            inl_ref = rds.IntegralNonLinearity()
-        except AttributeError:
-            inl_ref = {"meta": {}, 
-                       "inl_table": {},
-                       "value": {}
-                       }
-
-        inl_ref["meta"] = self.meta_data.export_asdf_meta()
-        inl_ref["inl_table"] = self._make_inl_table()
-        inl_ref["value"] = self.value_array
-
-        return inl_ref
+        inl_datamodel = rds.IntegralnonlinearityRef()
+        inl_datamodel["meta"] = self.meta_data.export_asdf_meta()
+        inl_datamodel["inl_table"] = self._make_inl_table()
+        inl_datamodel["value"] = self.value_array
+
+        return inl_datamodel
 
 
 def simulate_inl_correction_array():

src/wfi_reference_pipeline/reference_types/tests/test_integralnonlinearity.py

@@ -0,0 +1,102 @@
+import numpy as np
+import pytest
+
+from wfi_reference_pipeline.constants import REF_TYPE_INTEGRALNONLINEARITY
+from wfi_reference_pipeline.reference_types.integral_non_linearity.integral_non_linearity import (
+    IntegralNonLinearity,
+)
+from wfi_reference_pipeline.resources.make_test_meta import MakeTestMeta
+
+
+@pytest.fixture
+def valid_meta_data():
+    """Fixture for generating valid WFIMetaIntegralNonLinearity metadata."""
+    test_meta = MakeTestMeta(ref_type=REF_TYPE_INTEGRALNONLINEARITY)
+    return test_meta.meta_integral_non_linearity
+
+
+@pytest.fixture
+def inl_array():
+    """Return a valid 2D INL correction array shape (32, 65536)."""
+    return np.zeros((32, 65536), dtype=np.float64)
+
+
+@pytest.fixture
+def integral_non_linearity_object(valid_meta_data, inl_array):
+    """Fixture for initializing an IntegralNonLinearity object with valid metadata."""
+    return IntegralNonLinearity(meta_data=valid_meta_data, ref_type_data=inl_array)
+
+
+class TestIntegralNonLinearity:
+
+    def test_integral_non_linearity_instantiation_with_valid_metadata(
+        self, integral_non_linearity_object
+    ):
+        """
+        Test that IntegralNonLinearity object is created successfully with valid metadata.
+        """
+        assert isinstance(integral_non_linearity_object, IntegralNonLinearity)
+
+    def test_default_description_is_set_if_empty(self, valid_meta_data, inl_array):
+        """
+        Test that a default description is populated if metadata description is empty.
+        """
+        valid_meta_data.description = ""
+        inl_obj = IntegralNonLinearity(meta_data=valid_meta_data, ref_type_data=inl_array)
+
+        assert inl_obj.meta_data.description == "Roman WFI integral non linearity reference file."
+
+    def test_populate_datamodel_tree(self, integral_non_linearity_object):
+        """
+        Test that the data model tree is correctly populated.
+        """
+        data_model_tree = integral_non_linearity_object.populate_datamodel_tree()
+
+        assert "meta" in data_model_tree
+        assert "inl_table" in data_model_tree
+        assert "value" in data_model_tree
+
+        inl_table = data_model_tree["inl_table"]
+        assert isinstance(inl_table, dict)
+        assert len(inl_table) == 32
+
+        # Ensure all keys exist and values are correct
+        # There are 32 amplifiers to read 128 pixels at a time. 
+        # https://roman-docs.stsci.edu/data-handbook/wfi-data-levels-and-products/coordinate-systems
+        for i in range(32):
+            key = f"science_channel_{i+1:02d}"
+            assert key in inl_table
+
+            table_entry = inl_table[key]
+            assert "instrument_channel" in table_entry
+            assert "correction" in table_entry
+
+            assert table_entry["instrument_channel"] == i
+            assert isinstance(table_entry["correction"], np.ndarray)
+            # The Analog to Digital conversion is in UINT16.
+            assert table_entry["correction"].shape == (65536,)
+
+        value_array = data_model_tree["value"]
+        assert isinstance(value_array, np.ndarray)
+        assert value_array.shape == (65536,)
+
+    def test_integral_non_linearity_outfile_default(self, integral_non_linearity_object):
+        """
+        Test that the default outfile name is correct.
+        """
+        assert integral_non_linearity_object.outfile == "roman_inl.asdf"
+
+    def test_invalid_ref_type_data_shape_raises_value_error(self, valid_meta_data):
+        """
+        Test that invalid INL array shapes raise a ValueError.
+        """
+        bad_array = np.zeros((10, 10))
+        with pytest.raises(ValueError):
+            IntegralNonLinearity(meta_data=valid_meta_data, ref_type_data=bad_array)
+
+    def test_invalid_meta_type_raises_type_error(self, inl_array):
+        """
+        Test that invalid meta type raises a TypeError.
+        """
+        with pytest.raises(TypeError):
+            IntegralNonLinearity(meta_data=object(), ref_type_data=inl_array)

src/wfi_reference_pipeline/resources/make_dev_meta.py

@@ -89,8 +89,10 @@ def _create_dev_meta_gain(self, meta_data):
         self.meta_gain = WFIMetaGain(*meta_data)
 
     def _create_dev_meta_integral_non_linearity(self, meta_data):
-        n_channels = '32'
-        n_pixels_per_channel = '128'
+        # There are 32 amplifiers to read 128 pixels at a time. 
+        # https://roman-docs.stsci.edu/data-handbook/wfi-data-levels-and-products/coordinate-systems
+        n_channels = 32
+        n_pixels_per_channel = 128
 
         meta_integral_non_linearity = [n_channels, n_pixels_per_channel]
 

src/wfi_reference_pipeline/resources/make_test_meta.py

@@ -87,16 +87,16 @@ def _create_test_meta_flat(self, meta_data):
     def _create_test_meta_gain(self, meta_data):
         self.meta_gain = WFIMetaGain(*meta_data)
 
-    def _create_test_meta_intengral_non_linearity(self, meta_data):
-        n_channels = '32'
-        n_pixels_per_channel = '128'
+    def _create_test_meta_integral_non_linearity(self, meta_data):
+        # There are 32 amplifiers to read 128 pixels at a time. 
+        # https://roman-docs.stsci.edu/data-handbook/wfi-data-levels-and-products/coordinate-systems
+        n_channels = 32
+        n_pixels_per_channel = 128
 
         meta_integral_non_linearity = [n_channels, n_pixels_per_channel]
         self.meta_integral_non_linearity = WFIMetaIntegralNonLinearity(*meta_data,
                                                                        *meta_integral_non_linearity)
 
-        self._create_test_meta_intengral_non_linearity = WFIMetaIntegralNonLinearity(*meta_data)
-
     def _create_test_meta_interpixelcapacitance(self, meta_data):
         ref_optical_element = "F158"
 
@@ -192,7 +192,7 @@ def __init__(self, ref_type):
             self._create_test_meta_gain(meta_data_params)
 
         if ref_type == REF_TYPE_INTEGRALNONLINEARITY:
-            self._create_test_meta_intengral_non_linearity(meta_data_params)
+            self._create_test_meta_integral_non_linearity(meta_data_params)
 
         if ref_type == REF_TYPE_INVERSELINEARITY:
             self._create_test_meta_inverselinearity(meta_data_params)

src/wfi_reference_pipeline/resources/wfi_meta_integral_non_linearity.py

@@ -12,8 +12,8 @@ class WFIMetaIntegralNonLinearity(WFIMetadata):
     All Fields are required and positional with base class fields first
     """
     # These are required reftype specific
-    n_channels: str
-    n_pixels_per_channel: str
+    n_channels: int
+    n_pixels_per_channel: int
 
     def __post_init__(self):
         super().__post_init__()