RCAL-362 Fix Tvac Filename Node Type

CHANGES.rst

@@ -7,6 +7,8 @@
 
 - Enable asdf "lazy_tree" mode for all roman datamodels files [#358]
 
+- Fix to preserve Tvac & FPS tagged scalar node types. [#368]
+
 0.20.0 (2024-05-15)
 ===================
 

src/roman_datamodels/datamodels/_datamodels.py

@@ -18,6 +18,27 @@
 
 __all__ = []
 
+TVAC_SCALAR_NODES = [
+    stnode.TvacCalibrationSoftwareVersion,
+    stnode.TvacSdfSoftwareVersion,
+    stnode.TvacFilename,
+    stnode.TvacFileDate,
+    stnode.TvacModelType,
+    stnode.TvacOrigin,
+    stnode.TvacPrdSoftwareVersion,
+    stnode.TvacTelescope,
+]
+FPS_SCALAR_NODES = [
+    stnode.FpsCalibrationSoftwareVersion,
+    stnode.FpsSdfSoftwareVersion,
+    stnode.FpsFilename,
+    stnode.FpsFileDate,
+    stnode.FpsModelType,
+    stnode.FpsOrigin,
+    stnode.FpsPrdSoftwareVersion,
+    stnode.FpsTelescope,
+]
+
 
 class _DataModel(DataModel):
     """
@@ -189,6 +210,11 @@ def node_update(self, other):
                     if isinstance(self[key], np.ndarray):
                         self[key] = other.__getattr__(key).astype(self[key].dtype)
                         continue
+                    if type(other[key]) in (TVAC_SCALAR_NODES + FPS_SCALAR_NODES):
+                        from roman_datamodels.maker_utils import mk_basic_meta
+
+                        self[key] = mk_basic_meta(**{key: other[key]})[key]
+                        continue
                 self[key] = other.__getattr__(key)
 
         node_update(ramp, model)

src/roman_datamodels/stnode/_node.py

@@ -28,6 +28,27 @@
 validator_callbacks = HashableDict(asdfschema.YAML_VALIDATORS)
 validator_callbacks.update({"type": _check_type})
 
+TVAC_SCALAR_NODES = [
+    "TvacCalibrationSoftwareVersion",
+    "TvacSdfSoftwareVersion",
+    "TvacFilename",
+    "TvacFileDate",
+    "TvacModelType",
+    "TvacOrigin",
+    "TvacPrdSoftwareVersion",
+    "TvacTelescope",
+]
+FPS_SCALAR_NODES = [
+    "FpsCalibrationSoftwareVersion",
+    "FpsSdfSoftwareVersion",
+    "FpsFilename",
+    "FpsFileDate",
+    "FpsModelType",
+    "FpsOrigin",
+    "FpsPrdSoftwareVersion",
+    "FpsTelescope",
+]
+
 
 def _value_change(path, value, schema, pass_invalid_values, strict_validation, ctx):
     """
@@ -206,8 +227,19 @@ def __getattr__(self, key):
 
         # If the key is in the schema, then we can return the value
         if key in self._data:
+            if (".Tvac" in str(type(self._parent))) or (
+                str(type(self._data[key])).split(".")[-1].split("'")[0] in TVAC_SCALAR_NODES
+            ):
+                scaled_key = "tvac_" + key
+            elif (".Fps" in str(type(self._parent))) or (
+                str(type(self._data[key])).split(".")[-1].split("'")[0] in FPS_SCALAR_NODES
+            ):
+                scaled_key = "fps_" + key
+            else:
+                scaled_key = key
+
             # Cast the value into the appropriate tagged scalar class
-            value = self._convert_to_scalar(key, self._data[key])
+            value = self._convert_to_scalar(scaled_key, self._data[key])
 
             # Return objects as node classes, if applicable
             if isinstance(value, (dict, asdf.lazy_nodes.AsdfDictNode)):
@@ -229,9 +261,13 @@ def __setattr__(self, key, value):
 
         # Private keys should just be in the normal __dict__
         if key[0] != "_":
-
             # Wrap things in the tagged scalar classes if necessary
-            value = self._convert_to_scalar(key, value)
+            if (self._tag and "/tvac" in self._tag) or (".Tvac" in str(type(self._parent))):
+                value = self._convert_to_scalar("tvac_" + key, value)
+            elif (self._tag and "/fps" in self._tag) or (".Fps" in str(type(self._parent))):
+                value = self._convert_to_scalar("fps_" + key, value)
+            else:
+                value = self._convert_to_scalar(key, value)
 
             if key in self._data or key in self._schema_attributes:
                 # Perform validation if enabled
@@ -307,18 +343,32 @@ def __len__(self):
 
     def __getitem__(self, key):
         """Dictionary style access data"""
+
         if key in self._data:
-            return self._data[key]
+            if (".Tvac" in str(type(self._parent))) or (
+                str(type(self._data[key])).split(".")[-1].split("'")[0] in TVAC_SCALAR_NODES
+            ):
+                scaled_key = "tvac_" + key
+            elif (".Fps" in str(type(self._parent))) or (
+                str(type(self._data[key])).split(".")[-1].split("'")[0] in FPS_SCALAR_NODES
+            ):
+                scaled_key = "fps_" + key
+            else:
+                scaled_key = key
+
+            # Cast the value into the appropriate tagged scalar class
+            value = self._convert_to_scalar(scaled_key, self._data[key])
+            return value
 
         raise KeyError(f"No such key ({key}) found in node")
 
     def __setitem__(self, key, value):
         """Dictionary style access set data"""
 
         # Convert the value to a tagged scalar if necessary
-        if self._tag and "/tvac" in self._tag:
+        if (self._tag and "/tvac" in self._tag) or (".Tvac" in str(type(self._parent))):
             value = self._convert_to_scalar("tvac_" + key, value)
-        elif self._tag and "/fps" in self._tag:
+        elif (self._tag and "/fps" in self._tag) or (".Fps" in str(type(self._parent))):
             value = self._convert_to_scalar("fps_" + key, value)
         else:
             value = self._convert_to_scalar(key, value)

tests/test_models.py

@@ -980,7 +980,41 @@ def test_rampmodel_from_science_raw(model_class, expect_success):
 
         assert ramp.meta.calibration_software_version == model.meta.calibration_software_version
         assert ramp.meta.exposure.read_pattern == model.meta.exposure.read_pattern
+        assert ramp.validate() is None
 
     else:
         with pytest.raises(ValueError):
             datamodels.RampModel.from_science_raw(model)
+
+
+@pytest.mark.parametrize(
+    "model_class",
+    [datamodels.FpsModel, datamodels.RampModel, datamodels.ScienceRawModel, datamodels.TvacModel, datamodels.MosaicModel],
+)
+def test_model_assignment_access_types(model_class):
+    """Test assignment and access of model keyword value via keys and dot notation"""
+    # Test creation
+    model = utils.mk_datamodel(
+        model_class, meta={"calibration_software_version": "1.2.3", "exposure": {"read_pattern": [[1], [2], [3]]}}
+    )
+
+    assert model["meta"]["filename"] == model.meta["filename"]
+    assert model["meta"]["filename"] == model.meta.filename
+    assert model.meta.filename == model.meta["filename"]
+    assert type(model["meta"]["filename"]) == type(model.meta["filename"])
+    assert type(model["meta"]["filename"]) == type(model.meta.filename)
+    assert type(model.meta.filename) == type(model.meta["filename"])
+
+    # Test assignment
+    model2 = utils.mk_datamodel(model_class, meta={"calibration_software_version": "4.5.6"})
+
+    model.meta["filename"] = "Roman_keys_test.asdf"
+    model2.meta.filename = "Roman_dot_test.asdf"
+
+    assert model.validate() is None
+    assert model2.validate() is None
+
+    # Test assignment types
+    assert type(model["meta"]["filename"]) == type(model2.meta["filename"])
+    assert type(model["meta"]["filename"]) == type(model2.meta.filename)
+    assert type(model.meta.filename) == type(model2.meta["filename"])

tests/test_stnode.py

@@ -175,7 +175,7 @@ def test_set_pattern_properties():
     with pytest.raises(asdf.ValidationError):
         mdl.phot_table.F062.pixelareasr = 3.14
 
-    # This is invalid be cause it is not a scalar
+    # This is invalid because it is not a scalar
     with pytest.raises(asdf.ValidationError):
         mdl.phot_table.F062.photmjsr = [37.0] * (u.MJy / u.sr)
     with pytest.raises(asdf.ValidationError):