WIP: Add concept for composable UMM generators

Author: reweeden

mandible/umm_generator/__init__.py

@@ -0,0 +1,92 @@
+import inspect
+from typing import Any, Dict, Type
+
+
+class MISSING:
+    __slots__ = ()
+
+
+class Umm:
+    _attributes = {}
+
+    def __init_subclass__(cls, **kwargs):
+        super().__init_subclass__(**kwargs)
+
+        # TODO(reweeden): Make this work with multiple inheritance?
+        parent_cls = super(cls, cls)
+        attributes = {**parent_cls._attributes}
+
+        for name, typ in get_annotations(cls).items():
+            # TODO(reweeden): What if we're overwriting an attribute from the
+            # parent and the types don't match?
+            attributes[name] = (typ, getattr(cls, name, MISSING))
+
+        # Update attributes with unannotated default values
+        for name, value in inspect.getmembers(cls):
+            if name.startswith("_") or inspect.isfunction(value):
+                continue
+
+            if name not in attributes:
+                attributes[name] = (Any, value)
+            else:
+                typ, _ = attributes[name]
+                attributes[name] = (typ, value)
+
+        cls._attributes = attributes
+
+    def __init__(self, metadata: Dict[str, Any]):
+        for name, (typ, default) in self._attributes.items():
+            if inspect.isclass(typ) and issubclass(typ, Umm):
+                if type(self) is typ:
+                    # TODO(reweeden): Error type?
+                    raise RuntimeError(
+                        f"Self-reference detected for attribute '{name}'",
+                    )
+
+                setattr(self, name, typ(metadata))
+            else:
+                value = default
+                if value is MISSING:
+                    # TODO(reweeden): Ability to set handler function manually?
+                    # For example:
+                    # class Foo(Umm):
+                    #     Attribute: str = Attr()
+                    #
+                    #     @Attribute.getter
+                    #     def get_attribute(self, metadata):
+                    #         ...
+                    handler_name = f"get_{name}"
+                    handler = getattr(self, handler_name, None)
+                    if handler:
+                        value = handler(metadata)
+
+                if value is MISSING:
+                    # TODO(reweeden): Error type?
+                    raise RuntimeError(
+                        f"Missing value for '{name}'. "
+                        f"Try implementing a 'get_{name}' method",
+                    )
+                setattr(self, name, value)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return _to_dict(self)
+
+
+def get_annotations(cls) -> Dict[str, Type[Any]]:
+    if hasattr(inspect, "get_annotations"):
+        return inspect.get_annotations(cls, eval_str=True)
+
+    # TODO(reweeden): String evaluation
+    return dict(cls.__annotations__)
+
+
+def _to_dict(obj: Any) -> Any:
+    if isinstance(obj, Umm):
+        return {
+            name: _to_dict(
+                getattr(obj, name),
+            )
+            for name in obj._attributes
+        }
+
+    return obj

mandible/umm_generator/base.py

@@ -0,0 +1,47 @@
+from typing import Any, Dict, Sequence
+
+from .base import Umm
+
+
+class AdditionalAttribute(Umm):
+    Name: str
+    Values: Sequence[str]
+
+
+class CollectionReference(Umm):
+    ShortName: str
+    Version: str
+
+
+# class DataGranule:
+#     "ArchiveAndDistributionInformation": self.get_archive_and_distribution_information(),
+#     "DayNightFlag": "Unspecified",
+#     "Identifiers": self.get_identifiers(),
+#     "ProductionDateTime": to_umm_str(self.get_product_creation_time()),
+
+
+class MetadataSpecification(Umm):
+    Name: str = "UMM-G"
+    URL: str = "https://cdn.earthdata.nasa.gov/umm/granule/v1.6.5"
+    Version: str = "1.6.5"
+
+
+class UmmG(Umm):
+    # Sorted?
+    AdditionalAttributes: Sequence[AdditionalAttribute]
+    CollectionReference: CollectionReference
+    # DataGranule: DataGranule
+    GranuleUR: str
+    MetadataSpecification: MetadataSpecification
+    # OrbitCalculatedSpatialDomains: self.get_orbit_calculated_spatial_domains(),
+    # PGEVersionClass: self.get_pge_version_class(),
+    # Platforms: self.get_platforms(),
+    # Projects: self.get_projects(),
+    # ProviderDates: self.get_provider_dates(),
+    # RelatedUrls: self.get_related_urls(),
+    # SpatialExtent: self.get_spatial_extent(),
+    # TemporalExtent: self.get_temporal_extent(),
+    # InputGranules: self.get_input_granules(),
+
+    def get_GranuleUR(self, metadata: Dict[str, Any]) -> str:
+        return metadata["granule"]["granuleId"]

mandible/umm_generator/umm_g.py

@@ -0,0 +1,79 @@
+import pytest
+
+from mandible.umm_generator.base import Umm
+from mandible.umm_generator.umm_g import CollectionReference, UmmG
+
+
+def test_custom_umm():
+    class TestComponent(Umm):
+        Field1: str
+        Field2: int
+
+        def get_Field1(self, metadata) -> str:
+            return metadata["field_1"]
+
+        def get_Field2(self, metadata) -> int:
+            return metadata["field_2"]
+
+    class TestMain(Umm):
+        Name: str
+        Component: TestComponent
+
+        def get_Name(self, metadata) -> str:
+            return metadata["name"]
+
+    metadata = {
+        "field_1": "Value 1",
+        "field_2": "Value 2",
+        "name": "Test Name",
+    }
+    item = TestMain(metadata)
+
+    assert item.Name == "Test Name"
+    assert item.to_dict() == {
+        "Name": "Test Name",
+        "Component": {
+            "Field1": "Value 1",
+            "Field2": "Value 2",
+        }
+    }
+
+
+def test_umm_g_abstract():
+    with pytest.raises(Exception):
+        _ = UmmG({})
+
+
+def test_umm_g():
+    class CustomCollectionReference(CollectionReference):
+        ShortName: str = "FOOBAR"
+        Version: str = "10"
+
+    class BasicUmmG(UmmG):
+        AdditionalAttributes = []
+        CollectionReference: CustomCollectionReference
+
+    metadata = {
+        "granule": {
+            "granuleId": "SomeGranuleId",
+        },
+    }
+    umm_g = BasicUmmG(metadata)
+
+    assert umm_g.AdditionalAttributes == []
+    assert umm_g.CollectionReference.ShortName == "FOOBAR"
+    assert umm_g.CollectionReference.Version == "10"
+
+    assert umm_g.to_dict() == {
+        "AdditionalAttributes": [],
+        "CollectionReference": {
+            "ShortName": "FOOBAR",
+            "Version": "10",
+        },
+        "GranuleUR": "SomeGranuleId",
+        "MetadataSpecification": {
+            "Name": "UMM-G",
+            "URL": "https://cdn.earthdata.nasa.gov/umm/granule/v1.6.5",
+            "Version": "1.6.5",
+        },
+    }