Move the default tag system into the datamodel

changes/660.bugfix.rst

@@ -0,0 +1,2 @@
+Fix some small inconsistencies in the ``Builder``/``create_*`` API and add better
+handling of the node specific ``tag``` setting.

changes/663.feature.rst

@@ -0,0 +1,2 @@
+Switch the ``DataModel`` subclasses from directly pointing at the stnode dynamic class
+to instead listing the ``tag_pattern`` for the schema under pinning that dynamic class.

changes/664.feature.rst

@@ -0,0 +1,3 @@
+Completely remove all of the ``._stnode._mixin`` related structure replacing them
+with specific ``TaggedScalarNode`` classes: ``TaggedStrNode``, ``TaggedTimeNode``.
+Any other necessary functionality was moved into  the  wrapping data models themselves.

changes/669.feature.rst

@@ -0,0 +1,2 @@
+Remove the ``_default_tag`` and ``_latest_manifest`` ClassVars from the TaggedNode classes. These
+have been replaced with dynamic (cached) lookups to the ASDF configuration itself.

changes/671.feature.rst

@@ -0,0 +1,3 @@
+Add simple tag migration function to the ``DataModel`` class. This is a dumb
+function that just makes a new model from the old with a new tag. It does not
+attempt to migrate any of the data, so routines in ``romancal`` maybe needed.

changes/673.feature.rst

@@ -0,0 +1 @@
+Move the default tag methods from the nodes in STNode into the DataModels themselves.

docs/conf.py

@@ -42,6 +42,7 @@
     "astropy": ("https://docs.astropy.org/en/stable/", None),
     "asdf": ("https://asdf.readthedocs.io/en/latest/", None),
     "rad": ("https://rad.readthedocs.io/en/latest/", None),
+    "semantic_version": ("https://python-semanticversion.readthedocs.io/en/latest/", None),
 }
 
 # Add any Sphinx extension module names here, as strings. They can be

src/roman_datamodels/_stnode/__init__.py

@@ -6,8 +6,8 @@
 from __future__ import annotations
 
 from ._converters import *  # noqa: F403
-from ._mixins import *  # noqa: F403
 from ._node import *  # noqa: F403
 from ._schema import *  # noqa: F403
 from ._stnode import *  # noqa: F403
 from ._tagged import *  # noqa: F403
+from ._uri import *  # noqa: F403

src/roman_datamodels/_stnode/_converters.py

@@ -21,7 +21,7 @@
 )
 
 if TYPE_CHECKING:
-    from ._tagged import SerializationNode, TaggedListNode, TaggedObjectNode, TaggedScalarNode
+    from ._tagged import SerializationNode, TaggedNode, TaggedObjectNode
 
 __all__ = [
     "TaggedListNodeConverter",
@@ -61,15 +61,12 @@ def types(self) -> tuple[type[SerializationNode], ...]:
     def to_yaml_tree(self, obj: SerializationNode, tag, ctx):
         return obj.data
 
-    def from_yaml_tree(self, node, tag, ctx) -> TaggedObjectNode | TaggedListNode | TaggedScalarNode:
+    def from_yaml_tree(self, node, tag, ctx) -> TaggedNode:
         if "file_date" in tag:
             converter = ctx.extension_manager.get_converter_for_type(Time)
             node = converter.from_yaml_tree(node, tag, ctx)
 
-        # TODO: Add method for setting read_tag with some checks
-        obj = NODE_CLASSES_BY_TAG[tag](node)
-        obj._read_tag = tag
-        return obj
+        return NODE_CLASSES_BY_TAG[tag].from_tag(node=node, tag=tag)
 
 
 class _TaggedNodeConverter(_RomanConverter):
@@ -135,10 +132,17 @@ def types(self):
         return list(SCALAR_NODE_CLASSES_BY_PATTERN.values())
 
     def to_yaml_tree(self, obj, tag, ctx):
-        node = type(obj).__bases__[0](obj)
+        from ._tagged import TaggedStrNode, TaggedTimeNode
 
-        if "file_date" in obj.tag:
-            converter = ctx.extension_manager.get_converter_for_type(type(node))
-            node = converter.to_yaml_tree(node, tag, ctx)
+        match obj:
+            case TaggedStrNode():
+                node = str(obj)
+
+            case TaggedTimeNode():
+                converter = ctx.extension_manager.get_converter_for_type(Time)
+                node = converter.to_yaml_tree(Time(obj), tag, ctx)
+
+            case _:
+                raise TypeError(f"Unsupported type {type(obj)} for TaggedScalarNodeConverter")
 
         return super().to_yaml_tree(node, obj.tag, ctx)

src/roman_datamodels/_stnode/_factories.py

@@ -5,24 +5,20 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import Any
 
-from astropy.time import Time
+from ._tagged import (
+    TaggedListNode,
+    TaggedNode,
+    TaggedObjectNode,
+    TaggedScalarNode,
+    TaggedStrNode,
+    TaggedTimeNode,
+    class_name_from_tag_uri,
+)
 
-from . import _mixins
-from ._tagged import TaggedListNode, TaggedObjectNode, TaggedScalarNode, class_name_from_tag_uri
+__all__ = ("stnode_factory",)
 
-if TYPE_CHECKING:
-    from ._tagged import tagged_type
-
-__all__ = ["stnode_factory"]
-
-# Map of scalar types by pattern (str is default)
-_SCALAR_TYPE_BY_PATTERN = {
-    "asdf://stsci.edu/datamodels/roman/tags/file_date-*": Time,
-    "asdf://stsci.edu/datamodels/roman/tags/fps/file_date-*": Time,
-    "asdf://stsci.edu/datamodels/roman/tags/tvac/file_date-*": Time,
-}
 # Map of node types by pattern (TaggedObjectNode is default)
 _NODE_TYPE_BY_PATTERN = {
     "asdf://stsci.edu/datamodels/roman/tags/cal_logs-*": TaggedListNode,
@@ -48,7 +44,7 @@ def docstring_from_tag(tag_def: dict[str, Any]) -> str:
     return docstring + f"Class generated from tag '{tag_def['tag_uri']}'"
 
 
-def scalar_factory(pattern: str, latest_manifest: str, tag_def: dict[str, Any]) -> type[TaggedScalarNode]:
+def scalar_factory(pattern: str, tag_def: dict[str, Any]) -> type[TaggedScalarNode]:
     """
     Factory to create a TaggedScalarNode class from a tag
 
@@ -57,49 +53,25 @@ def scalar_factory(pattern: str, latest_manifest: str, tag_def: dict[str, Any])
     pattern: str
         A tag pattern/wildcard
 
-    latest_manifest: str
-        URI for the latest manifest
-
     tag_def: dict
         A tag entry from the RAD manifest
 
     Returns
     -------
     A dynamically generated TaggedScalarNode subclass
     """
-    class_name = class_name_from_tag_uri(pattern)
-
-    # TaggedScalarNode subclasses are really subclasses of the type of the scalar,
-    #   with the TaggedScalarNode as a mixin.  This is because the TaggedScalarNode
-    #   is supposed to be the scalar, but it needs to be serializable under a specific
-    #   ASDF tag.
-    # _SCALAR_TYPE_BY_PATTERN will need to be updated as new wrappers of scalar types are added
-    #   to the RAD manifest.
-    # assume everything is a string if not otherwise defined
-    class_type = _SCALAR_TYPE_BY_PATTERN.get(pattern, str)
-
-    # In special cases one may need to add additional features to a tagged node class.
-    #   This is done by creating a mixin class with the name <ClassName>Mixin in _mixins.py
-    #   Here we mixin the mixin class if it exists.
-    if hasattr(_mixins, mixin := f"{class_name}Mixin"):
-        class_type = (class_type, getattr(_mixins, mixin), TaggedScalarNode)
-    else:
-        class_type = (class_type, TaggedScalarNode)
 
     return type(
-        class_name,
-        class_type,
+        class_name_from_tag_uri(pattern),
+        (TaggedTimeNode,) if "file_date" in pattern else (TaggedStrNode,),
         {
-            "_pattern": pattern,
-            "_latest_manifest": latest_manifest,
-            "_default_tag": tag_def["tag_uri"],
             "__module__": "roman_datamodels._stnode",
             "__doc__": docstring_from_tag(tag_def),
         },
     )
 
 
-def node_factory(pattern: str, latest_manifest: str, tag_def: dict[str, Any]) -> type[TaggedObjectNode | TaggedListNode]:
+def node_factory(pattern: str, tag_def: dict[str, Any]) -> type[TaggedObjectNode | TaggedListNode]:
     """
     Factory to create a TaggedObjectNode or TaggedListNode class from a tag
 
@@ -108,46 +80,26 @@ def node_factory(pattern: str, latest_manifest: str, tag_def: dict[str, Any]) ->
     pattern: str
         A tag pattern/wildcard
 
-    latest_manifest: str
-        URI for the latest manifest
-
     tag_def: dict
         A tag entry from the RAD manifest
 
     Returns
     -------
     A dynamically generated TaggedObjectNode or TaggedListNode subclass
     """
-    class_name = class_name_from_tag_uri(pattern)
-
-    base_class_type = _NODE_TYPE_BY_PATTERN.get(pattern, TaggedObjectNode)
-
-    # In special cases one may need to add additional features to a tagged node class.
-    #   This is done by creating a mixin class with the name <ClassName>Mixin in _mixins.py
-    #   Here we mixin the mixin class if it exists.
-    class_type: tuple[Any, tagged_type] | tuple[tagged_type]
-    if hasattr(_mixins, mixin := f"{class_name}Mixin"):
-        class_type = (getattr(_mixins, mixin), base_class_type)
-    else:
-        class_type = (base_class_type,)
 
     return type(
-        class_name,
-        class_type,
+        class_name_from_tag_uri(pattern),
+        (TaggedListNode,) if "cal_logs" in pattern else (TaggedObjectNode,),
         {
-            "_pattern": pattern,
-            "_latest_manifest": latest_manifest,
-            "_default_tag": tag_def["tag_uri"],
             "__module__": "roman_datamodels._stnode",
             "__doc__": docstring_from_tag(tag_def),
             "__slots__": (),
         },
     )
 
 
-def stnode_factory(
-    pattern: str, latest_manifest: str, tag_def: dict[str, Any]
-) -> type[TaggedObjectNode | TaggedListNode | TaggedScalarNode]:
+def stnode_factory(pattern: str, tag_def: dict[str, Any]) -> type[TaggedNode]:
     """
     Construct a tagged STNode class from a tag
 
@@ -156,9 +108,6 @@ def stnode_factory(
     pattern: str
         A tag pattern/wildcard
 
-    latest_manifest: str
-        URI for the latest manifest
-
     tag_def: dict
         A tag entry from the RAD manifest
 
@@ -169,6 +118,6 @@ def stnode_factory(
     # TaggedScalarNodes are a special case because they are not a subclass of a
     #   _node class, but rather a subclass of the type of the scalar.
     if "tagged_scalar" in tag_def["schema_uri"]:
-        return scalar_factory(pattern, latest_manifest, tag_def)
+        return scalar_factory(pattern, tag_def)
     else:
-        return node_factory(pattern, latest_manifest, tag_def)
+        return node_factory(pattern, tag_def)