write collection file as well

src/hazard/docs_store.py

@@ -1,7 +1,7 @@
 import json
 import os
 from pathlib import PurePosixPath
-from typing import Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 import s3fs  # type: ignore
 from fsspec import AbstractFileSystem  # type: ignore
@@ -10,7 +10,7 @@
 
 from hazard.sources.osc_zarr import default_dev_bucket
 
-from .inventory import HazardResource, HazardResources, resource_from_stac_item_dict
+from .inventory import HazardResource, HazardResources
 
 
 class DocStore:
@@ -86,14 +86,60 @@ def write_new_empty_inventory(self):
         with self._fs.open(path, "w") as f:
             f.write(json_str)
 
-    def update_inventory_stac(self, resources: Iterable[HazardResource]):
+    def write_inventory_stac(self, resources: Iterable[HazardResource]):
         """Write a hazard models inventory as STAC."""
 
-        path = self._full_path_inventory()
-        models = HazardResources(resources=resources).to_stac_items(path_root=self._root, items_as_dicts=True)
-        json_str = json.dumps(models, indent=4)
-        with self._fs.open(path, "w") as f:
+        items = HazardResources(resources=resources).to_stac_items(path_root=self._root, items_as_dicts=True)
+        for it in items:
+            with self._fs.open(self._full_path_stac_item(id=it["id"]), "w") as f:
+                f.write(json.dumps(it, indent=4))
+        catalog_path = self._full_path_stac_catalog()
+        catalog = self.stac_catalog(items=items)
+        with self._fs.open(catalog_path, "w") as f:
+            json_str = json.dumps(catalog, indent=4)
             f.write(json_str)
+        collection_path = self._full_path_stac_collection()
+        collection = self.stac_collection(items=items)
+        with self._fs.open(collection_path, "w") as f:
+            json_str = json.dumps(collection, indent=4)
+            f.write(json_str)
+
+    def stac_catalog(self, items: List[Dict[str, Any]]) -> Dict[str, Any]:
+
+        return {
+            "stac_version": "1.0.0",
+            "id": "osc-hazard-indicators-catalog",
+            "type": "catalog",
+            "description": "OS-C hazard indicators catalog",
+            "links": [
+                {"rel": "self", "href": "./catalog.json"},
+                {"rel": "root", "href": "./catalog.json"},
+                {"rel": "child", "href": "./collection.json"},
+            ]
+            + [{"rel": "item", "href": f"./{x['id']}.json"} for x in items],
+        }
+
+    def stac_collection(self, items: List[Dict[str, Any]]) -> Dict[str, Any]:
+
+        return {
+            "stac_version": "1.0.0",
+            "type": "Collection",
+            "stac_extensions": [],
+            "id": "osc-hazard-indicators-collection",
+            "title": "OS-C hazard indicators collection",
+            "description": "OS-C hazard indicators collection",
+            "license": "CC-BY-4.0",
+            "extent": {
+                "spatial": {"bbox": [[-180, -90, 180, 90]]},
+                "temporal": {"interval": [["1950-01-01T00:00:00Z", "2100-12-31T23:59:59Z"]]},
+            },
+            "providers": [{"name": "UKRI", "roles": ["producer"], "url": "https://www.ukri.org/"}],
+            "links": [
+                {"rel": "self", "type": "application/json", "href": "./collection.json"},
+                {"rel": "root", "type": "application/json", "href": "./catalog.json"},
+            ]
+            + [{"rel": "item", "href": f"./{x['id']}.json"} for x in items],
+        }
 
     def update_inventory(self, resources: Iterable[HazardResource], remove_existing: bool = False):
         """Add the hazard models provided to the inventory. If a model with the same key
@@ -126,3 +172,12 @@ def _full_path_doc(self, path: str):
 
     def _full_path_inventory(self):
         return str(PurePosixPath(self._root, "inventory.json"))
+
+    def _full_path_stac_item(self, id: str):
+        return str(PurePosixPath(self._root, f"{id}.json"))
+
+    def _full_path_stac_catalog(self):
+        return str(PurePosixPath(self._root, "catalog.json"))
+
+    def _full_path_stac_collection(self):
+        return str(PurePosixPath(self._root, "collection.json"))

src/hazard/inventory.py

@@ -186,32 +186,15 @@ def to_stac_item(
 class HazardResources(BaseModel):
     resources: List[HazardResource]
 
-    def to_stac_items(self, path_root: str, items_as_dicts: bool = False) -> Dict[str, Union[str, pystac.Item, Dict]]:
+    def to_stac_items(self, path_root: str, items_as_dicts: bool = False) -> List[Dict[str, Any]]:
         """
         converts hazard resources to a list of STAC items.
         """
         stac_items_lists = [
             resource.to_stac_items(path_root=path_root, items_as_dicts=items_as_dicts) for resource in self.resources
         ]
         stac_items_flat = list(itertools.chain(*stac_items_lists))
-        return {
-            "type": "FeatureCollection",
-            "features": stac_items_flat,
-        }
-
-
-def resource_from_stac_item_dict(stac_item: Dict) -> HazardResource:
-    """
-    converts STAC item to HazardResource
-    """
-
-    return HazardResource(
-        **{
-            k.replace("osc-hazard:", ""): stac_item["properties"][k]
-            for k in stac_item["properties"]
-            if "osc-hazard:" in k
-        }
-    )
+        return stac_items_flat
 
 
 def expand(item: str, key: str, param: str):