RCAL-833 Recursively convert all meta attributes during model casting

src/roman_datamodels/datamodels/_datamodels.py

@@ -6,6 +6,8 @@
     from the schema manifest defined by RAD.
 """
 
+from collections.abc import Mapping
+
 import asdf
 import numpy as np
 from astropy.table import QTable
@@ -135,37 +137,59 @@ class RampModel(_RomanDataModel):
     @classmethod
     def from_science_raw(cls, model):
         """
-        Construct a RampModel from a ScienceRawModel
+        Attempt to construct a RampModel from a DataModel
+
+        If the model has a `resultantdq` attribute, this is copied into
+        the `RampModel.groupdq` attribute.
 
         Parameters
         ----------
-        model : ScienceRawModel or RampModel
-            The input science raw model (a RampModel will also work)
+        model : ScienceRawModel, TvacModel
+            The input data model (a RampModel will also work).
+
+        Returns
+        -------
+        output_model : RampModel
+            The RampModel built from the input model. If the input is already
+            a RampModel, it is simply returned.
         """
-
         if isinstance(model, cls):
             return model
+        if not isinstance(model, (ScienceRawModel, TvacModel)):
+            raise ValueError("Input must be one of (RampModel, ScienceRawModel, TvacModel)")
+
+        # Create base ramp node with dummy values (for validation)
+        from roman_datamodels.maker_utils import mk_ramp
+
+        input_ramp = mk_ramp(shape=model.shape)
+
+        # check if the input model has a resultantdq from SDF
+        if hasattr(model, "resultantdq"):
+            input_ramp.groupdq = model.resultantdq.copy()
+
+        # Define how to recursively copy all attributes.
+        def node_update(self, other, orig=False):
+            """Implement update to directly access each value"""
+            for key in other.keys():
+                if key == "resultantdq":
+                    continue
+                if key in self:
+                    if isinstance(self[key], Mapping):
+                        node_update(self[key], other.__getattr__(key))
+                        continue
+                    if isinstance(self[key], list):
+                        self[key] = other.__getattr__(key).data
+                        continue
+                    if isinstance(self[key], np.ndarray):
+                        self[key] = other.__getattr__(key).astype(self[key].dtype)
+                        continue
+                self[key] = other.__getattr__(key)
 
-        if isinstance(model, ScienceRawModel):
-            from roman_datamodels.maker_utils import mk_ramp
-
-            instance = mk_ramp(shape=model.shape)
-
-            # Copy input_model contents into RampModel
-            for key in model:
-                # If a dictionary (like meta), overwrite entries (but keep
-                # required dummy entries that may not be in input_model)
-                if isinstance(instance[key], dict):
-                    instance[key].update(getattr(model, key))
-                elif isinstance(instance[key], np.ndarray):
-                    # Cast input ndarray as RampModel dtype
-                    instance[key] = getattr(model, key).astype(instance[key].dtype)
-                else:
-                    instance[key] = getattr(model, key)
-
-            return cls(instance)
+        node_update(input_ramp, model)
 
-        raise ValueError("Input model must be a ScienceRawModel or RampModel")
+        # Create model from node
+        output_model = RampModel(input_ramp)
+        return output_model
 
 
 class RampFitOutputModel(_RomanDataModel):

tests/test_models.py

@@ -958,3 +958,10 @@ def test_datamodel_save_filename(tmp_path):
 
     with datamodels.open(filename) as new_ramp:
         assert new_ramp.meta.filename == filename.name
+
+
+@pytest.mark.parametrize("model_class", [datamodels.RampModel, datamodels.ScienceRawModel, datamodels.TvacModel])
+def test_rampmodel_from_science_raw(model_class):
+    """Test creation of RampModel from raw science/tvac"""
+    model = utils.mk_datamodel(model_class)
+    ramp = datamodels.RampModel.from_science_raw(model)