Rcal-622 Add the ability to copy resultantdq from SDF to a rampmodel

CHANGES.rst

@@ -1,6 +1,11 @@
 0.13.1 (unreleased)
 ===================
 
+dq_init
+-------
+
+- Add the ability to copy resultantdq from a SDF science raw model to the new rampmodel created by dq_init [#1085]
+
 outlier_detection
 -----------------
 

pyproject.toml

@@ -51,7 +51,7 @@ docs = [
 ]
 test = [
     'ci-watson >=0.5.0',
-    'pytest >=4.6.0',
+    'pytest >=4.6.0, <8.0.0',
     'pytest-astropy',
     'metrics_logger >= 0.1.0',
     'deepdiff',

romancal/dq_init/dq_init_step.py

@@ -45,20 +45,26 @@
         if not isinstance(input_model, RampModel):
             # Create base ramp node with dummy values (for validation)
             input_ramp = maker_utils.mk_ramp(shape=input_model.shape)
+            # check if the input model has a resultantdq from SDF
+            if hasattr(input_model, "resultantdq"):
+                input_ramp.groupdq = input_model.resultantdq.copy()
 
             # Copy input_model contents into RampModel
             for key in input_model.keys():
-                # If a dictionary (like meta), overwrite entires (but keep
-                # required dummy entries that may not be in input_model)
-                if isinstance(input_ramp[key], dict):
-                    input_ramp[key].update(input_model.__getattr__(key))
-                elif isinstance(input_ramp[key], np.ndarray):
-                    # Cast input ndarray as RampModel dtype
-                    input_ramp[key] = input_model.__getattr__(key).astype(
-                        input_ramp[key].dtype
-                    )
-                else:
-                    input_ramp[key] = input_model.__getattr__(key)
+                # check for resultantdq if present copy this to the emp
+                # it to the ramp model, we don't want to carry this around
+                if key != "resultantdq":
+                    # If a dictionary (like meta), overwrite entires (but keep
+                    # required dummy entries that may not be in input_model)
+                    if isinstance(input_ramp[key], dict):
+                        input_ramp[key].update(input_model.__getattr__(key))
+                    elif isinstance(input_ramp[key], np.ndarray):
+                        # Cast input ndarray as RampModel dtype
+                        input_ramp[key] = input_model.__getattr__(key).astype(
+                            input_ramp[key].dtype
+                        )
+                    else:
+                        input_ramp[key] = input_model.__getattr__(key)
 
             # Create model from node
             output_model = RampModel(input_ramp)

romancal/dq_init/tests/test_dq_init.py

@@ -284,6 +284,57 @@ def test_dqinit_refpix(instrument, exptype):
     assert result.dq_border_ref_pix_bottom.shape == (4, 20)
 
 
+@pytest.mark.parametrize(
+    "instrument, exptype",
+    [
+        ("WFI", "WFI_IMAGE"),
+    ],
+)
+def test_dqinit_resultantdq(instrument, exptype):
+    """Test that the basic inferface works for data requiring a DQ reffile"""
+
+    # Set test size
+    shape = (2, 20, 20)
+
+    # Create test science raw model
+    wfi_sci_raw = maker_utils.mk_level1_science_raw(shape=shape, dq=True)
+    wfi_sci_raw.meta.instrument.name = instrument
+    wfi_sci_raw.meta.instrument.detector = "WFI01"
+    wfi_sci_raw.meta.instrument.optical_element = "F158"
+    wfi_sci_raw.meta["guidestar"]["gw_window_xstart"] = 1012
+    wfi_sci_raw.meta["guidestar"]["gw_window_xsize"] = 16
+    wfi_sci_raw.meta.exposure.type = exptype
+    wfi_sci_raw.resultantdq[1, 12, 12] = dqflags.pixel["DROPOUT"]
+    wfi_sci_raw.data = u.Quantity(
+        np.ones(shape, dtype=np.uint16), u.DN, dtype=np.uint16
+    )
+    wfi_sci_raw_model = ScienceRawModel(wfi_sci_raw)
+
+    # Create mask model
+    maskref = stnode.MaskRef()
+    meta = maker_utils.mk_ref_common("MASK")
+    meta["instrument"]["optical_element"] = "F158"
+    meta["instrument"]["detector"] = "WFI01"
+    maskref["meta"] = meta
+    maskref["data"] = np.ones(shape[1:], dtype=np.float32)
+    maskref["dq"] = np.zeros(shape[1:], dtype=np.uint16)
+    maskref["err"] = (RNG.uniform(size=shape[1:]) * 0.05).astype(np.float32)
+    maskref_model = MaskRefModel(maskref)
+
+    # Perform Data Quality application step
+    result = DQInitStep.call(wfi_sci_raw_model, override_mask=maskref_model)
+
+    # check that the resultantdq is present in the raw model
+    assert hasattr(wfi_sci_raw_model, "resultantdq")
+    # check that the resultantdq is not copied to the result
+    assert not hasattr(result, "resultantdq")
+    # check to see the resultantdq is the correct shape
+    assert wfi_sci_raw_model.resultantdq.shape == shape
+    # check to see the resultantdq & groupdq have the correct value
+    assert wfi_sci_raw_model.resultantdq[1, 12, 12] == dqflags.pixel["DROPOUT"]
+    assert result.groupdq[1, 12, 12] == dqflags.pixel["DROPOUT"]
+
+
 @pytest.mark.parametrize(
     "instrument, exptype",
     [