Arch finalization proposal

STACpopulator/input.py

@@ -2,6 +2,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Iterator, MutableMapping, Optional, Tuple
 
+import numpy as np
 import requests
 import siphon
 import xncml
@@ -37,10 +38,6 @@ def reset(self):
         pass
 
 
-
-
-
-
 class THREDDSLoader(GenericLoader):
     def __init__(self, thredds_catalog_url: str, depth: Optional[int] = None) -> None:
         """Constructor
@@ -89,6 +86,21 @@ def extract_metadata(self, ds: siphon.catalog.Dataset) -> MutableMapping[str, An
         # Convert NcML to CF-compliant dictionary
         attrs = xncml.Dataset.from_text(r.content).to_cf_dict()
 
+        # Converting numpy datatypes to python standard datatypes
+        for key, value in attrs["attributes"].items():
+            if isinstance(value, list):
+                newlist = []
+                for item in value:
+                    if issubclass(type(item), np.integer):
+                        newlist.append(int(item))
+                    elif issubclass(type(item), np.floating):
+                        newlist.append(float(item))
+                    else:
+                        newlist.append(item)
+                attrs["attributes"][key] = newlist
+            elif isinstance(type(value), np.integer):
+                attrs["attributes"][key] = int(value)
+
         attrs["access_urls"] = ds.access_urls
 
         return attrs

STACpopulator/models.py

@@ -0,0 +1,87 @@
+import datetime as dt
+from typing import Any, Dict, List, Optional, Union
+
+from pydantic import (
+    AnyHttpUrl,
+    AnyUrl,
+    BaseModel,
+    Field,
+    SerializeAsAny,
+    field_validator,
+)
+from typing_extensions import TypedDict
+
+
+class Geometry(TypedDict):
+    type: str
+    coordinates: List[List[List[float]]]
+
+
+class Asset(BaseModel):
+    href: AnyHttpUrl
+    media_type: Optional[str] = None
+    title: Optional[str] = None
+    description: Optional[str] = None
+    roles: Optional[List[str]] = None
+
+
+class STACItemProperties(BaseModel):
+    """Base STAC Item properties data model. In concrete implementations, users would want to define a new
+    data model that inherits from this base model and extends it with properties tailored to the data they are
+    ingesting."""
+
+    start_datetime: Optional[dt.datetime] = None
+    end_datetime: Optional[dt.datetime] = None
+    datetime: Optional[dt.datetime] = None
+
+    @field_validator("datetime", mode="before")
+    @classmethod
+    def validate_datetime(cls, v: Union[dt.datetime, str], values: Dict[str, Any]) -> dt:
+        if v == "null":
+            if not values["start_datetime"] and not values["end_datetime"]:
+                raise ValueError("start_datetime and end_datetime must be specified when datetime is null")
+
+
+# class Link(BaseModel):
+#     """
+#     https://github.com/radiantearth/stac-spec/blob/v1.0.0/collection-spec/collection-spec.md#link-object
+#     """
+
+#     href: str = Field(..., alias="href", min_length=1)
+#     rel: str = Field(..., alias="rel", min_length=1)
+#     type: Optional[str] = None
+#     title: Optional[str] = None
+#     # Label extension
+#     label: Optional[str] = Field(None, alias="label:assets")
+#     model_config = ConfigDict(use_enum_values=True)
+
+#     def resolve(self, base_url: str) -> None:
+#         """resolve a link to the given base URL"""
+#         self.href = urljoin(base_url, self.href)
+
+
+# class PaginationLink(Link):
+#     """
+#     https://github.com/radiantearth/stac-api-spec/blob/master/api-spec.md#paging-extension
+#     """
+
+#     rel: Literal["next", "previous"]
+#     method: Literal["GET", "POST"]
+#     body: Optional[Dict[Any, Any]] = None
+#     merge: bool = False
+
+
+# Links = RootModel[List[Union[PaginationLink, Link]]]
+
+
+class STACItem(BaseModel):
+    """STAC Item data model."""
+
+    id: str = Field(..., alias="id", min_length=1)
+    geometry: Optional[Geometry] = None
+    bbox: Optional[List[float]] = None
+    properties: Optional[SerializeAsAny[STACItemProperties]] = None
+    assets: Dict[str, Asset] = None
+    stac_extensions: Optional[List[AnyUrl]] = []
+    collection: Optional[str] = None
+    datetime: Optional[dt.datetime] = None  # Not in the spec, but needed by pystac.Item.

STACpopulator/populator_base.py

@@ -58,7 +58,7 @@ def __init__(
         self._ingest_pipeline = data_loader
         self._stac_host = self.validate_host(stac_host)
 
-        #self._collection_id = hashlib.md5(self.collection_name.encode("utf-8")).hexdigest()
+        # self._collection_id = hashlib.md5(self.collection_name.encode("utf-8")).hexdigest()
         self._collection_id = self.collection_name
         LOGGER.info("Initialization complete")
         LOGGER.info(f"Collection {self.collection_name} is assigned id {self._collection_id}")
@@ -76,6 +76,13 @@ def stac_host(self) -> str:
     def collection_id(self) -> str:
         return self._collection_id
 
+    @property
+    @abstractmethod
+    def item_properties_model(self):
+        """In derived classes, this property should be defined as a pydantic data model that derives from
+        models.STACItemProperties."""
+        pass
+
     def validate_host(self, stac_host: str) -> str:
         if not url_validate(stac_host):
             raise ValueError("stac_host URL is not appropriately formatted")
@@ -115,12 +122,12 @@ def ingest(self) -> None:
         for item_name, item_data in self._ingest_pipeline:
             LOGGER.info(f"Creating STAC representation for {item_name}")
             stac_item = self.create_stac_item(item_name, item_data)
-            post_stac_item(self.stac_host, self.collection_id, item_name, stac_item)
-            try:
-                pass
-            except Exception:
-                LOGGER.error(f"Failed adding STAC item {item_name}")
-                self.handle_ingestion_error("Posting Error", item_name, item_data)
+            # post_stac_item(self.stac_host, self.collection_id, item_name, stac_item)
+            # try:
+            #     pass
+            # except Exception:
+            #     LOGGER.error(f"Failed adding STAC item {item_name}")
+            #     self.handle_ingestion_error("Posting Error", item_name, item_data)
 
     @abstractmethod
     def handle_ingestion_error(self, error: str, item_name: str, item_data: MutableMapping[str, Any]):