Use dqflags defined in roman_datamodels (#1099)

CHANGES.rst

@@ -1,6 +1,11 @@
 0.14.1 (unreleased)
 ==================
 
+general
+-------
+
+- Update the ``dqflags`` to use the ones stored in ``roman_datamodels`` [#1099]
+
 documentation
 -------------
 
@@ -34,7 +39,6 @@ jump detection
 
 - Added uneven ramp-jump detection docs. [#1035]
 
-
 documentation
 -------------
 
@@ -44,7 +48,6 @@ documentation
 
 - added user documentation for ``roman_static_preview`` script [#1046]
 
-
 ramp_fitting
 ------------
 
@@ -55,15 +58,13 @@ resample
 
 - Update resample step to handle the L3 meta data [#1057]
 
-
 general
 -------
 
 - Update elp steps to check for CRDS not returning a reference file [#1055]
 
 - Fix bug where ``compare_asdf`` failed to detect ``DataModel`` type differences. [#1066]
 
-
 0.13.0 (2023-11-28)
 ===================
 

romancal/conftest.py

@@ -17,6 +17,8 @@
 
 from romancal.assign_wcs import pointing
 
+collect_ignore = ["lib/dqflags.py"]
+
 
 @pytest.fixture
 def mk_tmp_dirs():

romancal/dq_init/dq_init_step.py

@@ -4,9 +4,9 @@
 import roman_datamodels as rdm
 from roman_datamodels import maker_utils
 from roman_datamodels.datamodels import RampModel
+from roman_datamodels.dqflags import pixel
 
 from romancal.dq_init import dq_initialization
-from romancal.lib import dqflags
 from romancal.stpipe import RomanStep
 
 __all__ = ["DQInitStep"]
@@ -75,9 +75,7 @@ def process(self, input):
         x_start = input_model.meta.guidestar.gw_window_xstart
         x_end = input_model.meta.guidestar.gw_window_xsize + x_start
         # set pixeldq array to GW_AFFECTED_DATA (2**4) for the given range
-        output_model.pixeldq[int(x_start) : int(x_end), :] = dqflags.pixel[
-            "GW_AFFECTED_DATA"
-        ]
+        output_model.pixeldq[int(x_start) : int(x_end), :] = pixel.GW_AFFECTED_DATA
         self.log.info(
             f"Flagging rows from: {x_start} to {x_end} as affected by guide window read"
         )

romancal/dq_init/tests/test_dq_init.py

@@ -3,10 +3,10 @@
 from astropy import units as u
 from roman_datamodels import maker_utils, stnode
 from roman_datamodels.datamodels import MaskRefModel, ScienceRawModel
+from roman_datamodels.dqflags import pixel
 
 from romancal.dq_init import DQInitStep
 from romancal.dq_init.dq_initialization import do_dqinit
-from romancal.lib import dqflags
 
 RNG = np.random.default_rng(83)
 
@@ -56,18 +56,16 @@ def test_dq_im(xstart, ystart, xsize, ysize, ngroups, instrument, exp_type):
 
     # assert that the pixels read back in match the mapping from ref data to
     # science data
-    assert dqdata[100, 100] == dqflags.pixel["SATURATED"]
-    assert dqdata[200, 100] == dqflags.pixel["JUMP_DET"]
-    assert dqdata[300, 100] == dqflags.pixel["DROPOUT"]
-    assert dqdata[400, 100] == dqflags.pixel["PERSISTENCE"]
-    assert dqdata[500, 100] == dqflags.pixel["DO_NOT_USE"]
-    assert dqdata[600, 100] == dqflags.pixel["GW_AFFECTED_DATA"]
-    assert dqdata[100, 200] == dqflags.pixel["SATURATED"] + dqflags.pixel["DO_NOT_USE"]
-    assert dqdata[200, 200] == dqflags.pixel["JUMP_DET"] + dqflags.pixel["DO_NOT_USE"]
-    assert dqdata[300, 200] == dqflags.pixel["DROPOUT"] + dqflags.pixel["DO_NOT_USE"]
-    assert (
-        dqdata[400, 200] == dqflags.pixel["PERSISTENCE"] + dqflags.pixel["DO_NOT_USE"]
-    )
+    assert dqdata[100, 100] == pixel.SATURATED
+    assert dqdata[200, 100] == pixel.JUMP_DET
+    assert dqdata[300, 100] == pixel.DROPOUT
+    assert dqdata[400, 100] == pixel.PERSISTENCE
+    assert dqdata[500, 100] == pixel.DO_NOT_USE
+    assert dqdata[600, 100] == pixel.GW_AFFECTED_DATA
+    assert dqdata[100, 200] == pixel.SATURATED + pixel.DO_NOT_USE
+    assert dqdata[200, 200] == pixel.JUMP_DET + pixel.DO_NOT_USE
+    assert dqdata[300, 200] == pixel.DROPOUT + pixel.DO_NOT_USE
+    assert dqdata[400, 200] == pixel.PERSISTENCE + pixel.DO_NOT_USE
 
 
 def test_groupdq():
@@ -172,15 +170,9 @@ def test_dq_add1_groupdq():
 
     # test if pixels in pixeldq were incremented in value by 1
     # check that previous dq flag is added to mask value
-    assert (
-        outfile.pixeldq[505, 505]
-        == dqflags.pixel["JUMP_DET"] + dqflags.pixel["DO_NOT_USE"]
-    )
+    assert outfile.pixeldq[505, 505] == pixel.JUMP_DET + pixel.DO_NOT_USE
     # check two flags propagate correctly
-    assert (
-        outfile.pixeldq[400, 500]
-        == dqflags.pixel["SATURATED"] + dqflags.pixel["DO_NOT_USE"]
-    )
+    assert outfile.pixeldq[400, 500] == pixel.SATURATED + pixel.DO_NOT_USE
 
 
 @pytest.mark.parametrize(
@@ -304,7 +296,7 @@ def test_dqinit_resultantdq(instrument, exptype):
     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.resultantdq[1, 12, 12] = pixel["DROPOUT"]
     wfi_sci_raw.data = u.Quantity(
         np.ones(shape, dtype=np.uint16), u.DN, dtype=np.uint16
     )
@@ -331,8 +323,8 @@ def test_dqinit_resultantdq(instrument, exptype):
     # 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"]
+    assert wfi_sci_raw_model.resultantdq[1, 12, 12] == pixel["DROPOUT"]
+    assert result.groupdq[1, 12, 12] == pixel["DROPOUT"]
 
 
 @pytest.mark.parametrize(

romancal/flatfield/flat_field.py

@@ -6,8 +6,7 @@
 
 import numpy as np
 from astropy import units as u
-
-from romancal.lib import dqflags
+from roman_datamodels.dqflags import pixel
 
 log = logging.getLogger(__name__)
 log.setLevel(logging.DEBUG)
@@ -91,17 +90,15 @@ def apply_flat_field(science, flat):
     # Find pixels in the flat that have a value of NaN and set
     # their DQ to NO_FLAT_FIELD
     flat_nan = np.isnan(flat_data)
-    flat_dq[flat_nan] = np.bitwise_or(flat_dq[flat_nan], dqflags.pixel["NO_FLAT_FIELD"])
+    flat_dq[flat_nan] = np.bitwise_or(flat_dq[flat_nan], pixel.NO_FLAT_FIELD)
 
     # Find pixels in the flat that have a value of zero, and set
     # their DQ to NO_FLAT_FIELD
     flat_zero = np.where(flat_data == 0.0)
-    flat_dq[flat_zero] = np.bitwise_or(
-        flat_dq[flat_zero], dqflags.pixel["NO_FLAT_FIELD"]
-    )
+    flat_dq[flat_zero] = np.bitwise_or(flat_dq[flat_zero], pixel.NO_FLAT_FIELD)
 
     # Find all pixels in the flat that have a DQ value of NO_FLAT_FIELD
-    flat_bad = np.bitwise_and(flat_dq, dqflags.pixel["NO_FLAT_FIELD"])
+    flat_bad = np.bitwise_and(flat_dq, pixel.NO_FLAT_FIELD)
 
     # Reset the flat value of all bad pixels to 1.0, so that no
     # correction is made

romancal/jump/jump_step.py

@@ -7,9 +7,9 @@
 
 import numpy as np
 from roman_datamodels import datamodels as rdd
+from roman_datamodels.dqflags import group, pixel
 from stcal.jump.jump import detect_jumps
 
-from romancal.lib import dqflags
 from romancal.stpipe import RomanStep
 
 log = logging.getLogger(__name__)
@@ -118,11 +118,11 @@ def process(self, input):
             #              separate PR
             dqflags_d = {}  # Dict of DQ flags
             dqflags_d = {
-                "GOOD": dqflags.group["GOOD"],
-                "DO_NOT_USE": dqflags.group["DO_NOT_USE"],
-                "SATURATED": dqflags.group["SATURATED"],
-                "JUMP_DET": dqflags.group["JUMP_DET"],
-                "NO_GAIN_VALUE": dqflags.pixel["NO_GAIN_VALUE"],
+                "GOOD": group.GOOD,
+                "DO_NOT_USE": group.DO_NOT_USE,
+                "SATURATED": group.SATURATED,
+                "JUMP_DET": group.JUMP_DET,
+                "NO_GAIN_VALUE": pixel.NO_GAIN_VALUE,
             }
 
             gdq, pdq, *_ = detect_jumps(

romancal/lib/basic_utils.py

@@ -2,11 +2,7 @@
 
 import numpy as np
 from roman_datamodels.datamodels import AssociationsModel
-
-from romancal.lib import dqflags
-
-SATURATEDPIX = dqflags.pixel["SATURATED"]
-SATURATEDGRP = dqflags.group["SATURATED"]
+from roman_datamodels.dqflags import group, pixel
 
 
 def bytes2human(n):
@@ -49,9 +45,9 @@ def is_fully_saturated(model):
     Check to see if all data pixels are flagged as saturated.
     """
 
-    if np.all(np.bitwise_and(model.groupdq, SATURATEDGRP) == SATURATEDGRP):
+    if np.all(np.bitwise_and(model.groupdq, group.SATURATED) == group.SATURATED):
         return True
-    elif np.all(np.bitwise_and(model.pixeldq, SATURATEDPIX) == SATURATEDPIX):
+    elif np.all(np.bitwise_and(model.pixeldq, pixel.SATURATED) == pixel.SATURATED):
         return True
 
     return False

romancal/lib/dqflags.py

@@ -1,69 +1,11 @@
-""" Roman Data Quality Flags
+import warnings
 
-The definitions are documented in the Roman RTD:
+from roman_datamodels.dqflags import group, pixel
 
-[NOTE: Documentation not yet implemented. Fix this URL when completed.]
+__all__ = ["pixel", "group"]
 
-https://roman-cal-pipeline.readthedocs.io/en/latest/roman/references_general/references_general.html#data-quality-flags
-
-
-Implementation
--------------
-
-The flags are implemented as "bit flags": Each flag is assigned a bit position
-in a byte, or multi-byte word, of memory. If that bit is set, the flag assigned
-to that bit is interpreted as being set or active.
-
-The data structure that stores bit flags is just the standard Python `int`,
-which provides 32 bits. Bits of an integer are most easily referred to using
-the formula `2**bit_number` where `bit_number` is the 0-index bit of interest.
-"""
-
-# for this file drop E241 (multiple spaces after :, for readability)
-# flake8: noqa: E241
-# Pixel-specific flags
-pixel = {
-    "GOOD": 0,  # No bits set, all is good
-    "DO_NOT_USE": 2**0,  # Bad pixel. Do not use.
-    "SATURATED": 2**1,  # Pixel saturated during exposure
-    "JUMP_DET": 2**2,  # Jump detected during exposure
-    "DROPOUT": 2**3,  # Data lost in transmission
-    "GW_AFFECTED_DATA": 2**4,  # Data affected by the GW read window
-    "PERSISTENCE": 2**5,  # High persistence (was RESERVED_2)
-    "AD_FLOOR": 2**6,  # Below A/D floor (0 DN, was RESERVED_3)
-    "OUTLIER": 2**7,  # Flagged by outlier detection (was RESERVED_4)
-    "UNRELIABLE_ERROR": 2**8,  # Uncertainty exceeds quoted error
-    "NON_SCIENCE": 2**9,  # Pixel not on science portion of detector
-    "DEAD": 2**10,  # Dead pixel
-    "HOT": 2**11,  # Hot pixel
-    "WARM": 2**12,  # Warm pixel
-    "LOW_QE": 2**13,  # Low quantum efficiency
-    "TELEGRAPH": 2**15,  # Telegraph pixel
-    "NONLINEAR": 2**16,  # Pixel highly nonlinear
-    "BAD_REF_PIXEL": 2**17,  # Reference pixel cannot be used
-    "NO_FLAT_FIELD": 2**18,  # Flat field cannot be measured
-    "NO_GAIN_VALUE": 2**19,  # Gain cannot be measured
-    "NO_LIN_CORR": 2**20,  # Linearity correction not available
-    "NO_SAT_CHECK": 2**21,  # Saturation check not available
-    "UNRELIABLE_BIAS": 2**22,  # Bias variance large
-    "UNRELIABLE_DARK": 2**23,  # Dark variance large
-    "UNRELIABLE_SLOPE": 2**24,  # Slope variance large (i.e., noisy pixel)
-    "UNRELIABLE_FLAT": 2**25,  # Flat variance large
-    "RESERVED_5": 2**26,  #
-    "RESERVED_6": 2**27,  #
-    "UNRELIABLE_RESET": 2**28,  # Sensitive to reset anomaly
-    "RESERVED_7": 2**29,  #
-    "OTHER_BAD_PIXEL": 2**30,  # A catch-all flag
-    "REFERENCE_PIXEL": 2**31,  # Pixel is a reference pixel
-}
-
-# Group-specific flags. Once groups are combined, these flags
-# are equivalent to the pixel-specific flags.
-group = {
-    "GOOD": pixel["GOOD"],
-    "DO_NOT_USE": pixel["DO_NOT_USE"],
-    "SATURATED": pixel["SATURATED"],
-    "JUMP_DET": pixel["JUMP_DET"],
-    "DROPOUT": pixel["DROPOUT"],
-    "AD_FLOOR": pixel["AD_FLOOR"],
-}
+warnings.warn(
+    "romancal.dqflags is deprecated. Please use roman_datamodels.dqflags instead.",
+    DeprecationWarning,
+    stacklevel=2,
+)

romancal/lib/psf.py

@@ -20,10 +20,9 @@
     SourceGrouper,
 )
 from roman_datamodels.datamodels import ImageModel
+from roman_datamodels.dqflags import pixel
 from webbpsf import conf, gridded_library, restart_logging
 
-from romancal.lib.dqflags import pixel as roman_dq_flag_map
-
 __all__ = [
     "create_gridded_psf_model",
     "fit_psf_to_image_model",
@@ -389,9 +388,7 @@ def dq_to_boolean_mask(image_model_or_dq, ignore_flags=0, flag_map_name="ROMAN_D
         dq = image_model_or_dq
 
     # add the Roman DQ flags to the astropy bitmask registry:
-    dq_flag_map = dict(roman_dq_flag_map)
-    dq_flag_map.pop("GOOD")
-
+    dq_flag_map = {dq.name: dq.value for dq in pixel if dq.name != "GOOD"}
     bitmask.extend_bit_flag_map(flag_map_name, **dq_flag_map)
 
     # convert the bitmask to a boolean mask:

romancal/lib/tests/test_suffix.py

@@ -14,6 +14,7 @@ def enable_logging():
         yield
 
 
+@pytest.mark.filterwarnings("ignore:romancal.dqflags is deprecated.*")
 def test_suffix_existence(enable_logging):
     """Generate current suffix list and compare"""
 

romancal/linearity/linearity_step.py

@@ -5,9 +5,9 @@
 import numpy as np
 from astropy import units as u
 from roman_datamodels import datamodels as rdd
+from roman_datamodels.dqflags import pixel
 from stcal.linearity.linearity import linearity_correction
 
-from romancal.lib import dqflags
 from romancal.stpipe import RomanStep
 
 __all__ = ["LinearityStep"]
@@ -53,7 +53,7 @@ def process(self, input):
             # The third return value is the procesed zero frame which
             # Roman does not use.
             new_data, new_pdq, _ = linearity_correction(
-                input_model.data.value, gdq, pdq, lin_coeffs, lin_dq, dqflags.pixel
+                input_model.data.value, gdq, pdq, lin_coeffs, lin_dq, pixel
             )
 
             input_model.data = u.Quantity(