feat!: stac providers TDE-623 (#348)

* feat: simple stac provider

* feat: add roles into default provider (linz) if selected as licensor (or producer)

* chore: remove debug

scripts/collection_from_items.py

@@ -1,13 +1,15 @@
 import argparse
 import json
 import os
+from typing import List
 
 from boto3 import client
 from linz_logger import get_log
 
 from scripts.files.fs_s3 import bucket_name_from_path, get_object_parallel_multithreading, list_json_in_uri
 from scripts.logging.time_helper import time_in_ms
 from scripts.stac.imagery.collection import ImageryCollection
+from scripts.stac.imagery.provider import Provider, ProviderRole
 
 
 def main() -> None:
@@ -16,15 +18,21 @@ def main() -> None:
     parser.add_argument("--collection-id", dest="collection_id", required=True)
     parser.add_argument("--title", dest="title", help="collection title", required=True)
     parser.add_argument("--description", dest="description", help="collection description", required=True)
+    parser.add_argument("--producer", dest="producer", help="imagery producer", required=True)
+    parser.add_argument("--licensor", dest="licensor", help="imagery licensor", required=True)
     parser.add_argument(
         "--concurrency", dest="concurrency", help="The number of files to limit concurrent reads", required=True, type=int
     )
 
     arguments = parser.parse_args()
     uri = arguments.uri
+    providers: List[Provider] = [
+        {"name": arguments.producer, "roles": [ProviderRole.PRODUCER]},
+        {"name": arguments.licensor, "roles": [ProviderRole.LICENSOR]},
+    ]
 
     collection = ImageryCollection(
-        title=arguments.title, description=arguments.description, collection_id=arguments.collection_id
+        title=arguments.title, description=arguments.description, collection_id=arguments.collection_id, providers=providers
     )
 
     if not uri.startswith("s3://"):

scripts/stac/imagery/collection.py

@@ -5,13 +5,16 @@
 import ulid
 
 from scripts.files.fs import write
+from scripts.stac.imagery.provider import Provider, ProviderRole
 from scripts.stac.util.STAC_VERSION import STAC_VERSION
 
 
 class ImageryCollection:
     stac: Dict[str, Any]
 
-    def __init__(self, title: str, description: str, collection_id: Optional[str] = None) -> None:
+    def __init__(
+        self, title: str, description: str, collection_id: Optional[str] = None, providers: Optional[List[Provider]] = None
+    ) -> None:
         if not collection_id:
             collection_id = str(ulid.ULID())
 
@@ -23,8 +26,26 @@ def __init__(self, title: str, description: str, collection_id: Optional[str] =
             "description": description,
             "license": "CC-BY-4.0",
             "links": [{"rel": "self", "href": "./collection.json", "type": "application/json"}],
+            "providers": [],
         }
 
+        # If the providers passed has already a LINZ provider: add its default roles to it
+        has_linz = False
+        if providers:
+            linz = next((p for p in providers if p["name"] == "Toitū Te Whenua Land Information New Zealand"), None)
+            if linz:
+                linz["roles"].extend([ProviderRole.HOST, ProviderRole.PROCESSOR])
+                has_linz = True
+        else:
+            providers = []
+
+        if not has_linz:
+            providers.append(
+                {"name": "Toitū Te Whenua Land Information New Zealand", "roles": [ProviderRole.HOST, ProviderRole.PROCESSOR]}
+            )
+
+        self.add_providers(providers)
+
     def add_item(self, item: Dict[Any, Any]) -> None:
         item_self_link = next((feat for feat in item["links"] if feat["rel"] == "self"), None)
         if item_self_link:
@@ -35,6 +56,10 @@ def add_item(self, item: Dict[Any, Any]) -> None:
     def add_link(self, href: str, rel: str = "item", file_type: str = "application/json") -> None:
         self.stac["links"].append({"rel": rel, "href": href, "type": file_type})
 
+    def add_providers(self, providers: List[Provider]) -> None:
+        for p in providers:
+            self.stac["providers"].append(p)
+
     def update_spatial_extent(self, item_bbox: List[float]) -> None:
         if "extent" not in self.stac:
             self.update_extent(bbox=item_bbox)

scripts/stac/imagery/provider.py

@@ -0,0 +1,16 @@
+from enum import Enum
+from typing import List, TypedDict
+
+
+class ProviderRole(str, Enum):
+    PRODUCER = "producer"
+    LICENSOR = "licensor"
+    PROCESSOR = "processor"
+    HOST = "host"
+
+
+class Provider(TypedDict):
+    name: str
+    """Organization name"""
+    roles: List[ProviderRole]
+    """Organization roles"""

scripts/stac/tests/collection_test.py

@@ -9,6 +9,7 @@
 from scripts.files.fs import read
 from scripts.stac.imagery.collection import ImageryCollection
 from scripts.stac.imagery.item import ImageryItem
+from scripts.stac.imagery.provider import Provider, ProviderRole
 
 
 @pytest.fixture(name="setup_collection", autouse=True)
@@ -119,3 +120,39 @@ def test_write_collection_special_chars(setup_collection: ImageryCollection) ->
     rmtree(target)
 
     assert collection["title"] == title
+
+
+def test_add_providers(setup_collection: ImageryCollection) -> None:
+    collection = setup_collection
+    producer: Provider = {"name": "Maxar", "roles": [ProviderRole.PRODUCER]}
+    collection.add_providers([producer])
+
+    assert {"name": "Maxar", "roles": ["producer"]} in collection.stac["providers"]
+
+
+def test_default_provider_present() -> None:
+    licensor: Provider = {"name": "Toitū Te Whenua Land Information New Zealand", "roles": [ProviderRole.LICENSOR]}
+    producer: Provider = {"name": "Maxar", "roles": [ProviderRole.PRODUCER]}
+    title = "Test Urban Imagery"
+    description = "Test Urban Imagery Description"
+    collection = ImageryCollection(title, description, providers=[producer, licensor])
+
+    assert {
+        "name": "Toitū Te Whenua Land Information New Zealand",
+        "roles": ["licensor", "host", "processor"],
+    } in collection.stac["providers"]
+    assert {"name": "Toitū Te Whenua Land Information New Zealand", "roles": ["host", "processor"]} not in collection.stac[
+        "providers"
+    ]
+
+
+def test_default_provider_missing() -> None:
+    producer: Provider = {"name": "Maxar", "roles": [ProviderRole.PRODUCER]}
+    title = "Test Urban Imagery"
+    description = "Test Urban Imagery Description"
+    collection = ImageryCollection(title, description, providers=[producer])
+
+    assert {"name": "Toitū Te Whenua Land Information New Zealand", "roles": ["host", "processor"]} in collection.stac[
+        "providers"
+    ]
+    assert {"name": "Maxar", "roles": ["producer"]} in collection.stac["providers"]