Refactor: refactor stac creation & create items in standardise_valida…

…te.py TDE-444 (#167)

* refactor: create stac

* feat: create items during standardise_validate.py

* fix: format

scripts/create_stac.py

@@ -0,0 +1,85 @@
+import argparse
+import json
+import os
+from typing import Any, Dict, Optional
+
+from linz_logger import get_log
+
+from scripts.cli.cli_helper import format_date, format_source, valid_date
+from scripts.files.files_helper import get_file_name_from_path, is_tiff
+from scripts.files.fs import write
+from scripts.files.geotiff import get_extents
+from scripts.gdal.gdalinfo import gdal_info
+from scripts.stac.imagery.collection import ImageryCollection
+from scripts.stac.imagery.item import ImageryItem
+
+
+def main() -> None:
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source", dest="source", nargs="+", required=True)
+    parser.add_argument("--collection_id", dest="collection_id", help="Unique id for collection", required=False)
+    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(
+        "--start_datetime", dest="start_datetime", help="start datetime in format YYYY-MM-DD", type=valid_date, required=True
+    )
+    parser.add_argument(
+        "--end_datetime", dest="end_datetime", help="end datetime in format YYYY-MM-DD", type=valid_date, required=True
+    )
+
+    arguments = parser.parse_args()
+
+    source = format_source(arguments.source)
+
+    if arguments.collection_id:
+        collection = ImageryCollection(title=arguments.title, description=arguments.description, collection_id=arguments.ulid)
+    else:
+        collection = ImageryCollection(title=arguments.title, description=arguments.description)
+
+    for file in source:
+        if not is_tiff(file):
+            get_log().trace("file_not_tiff_skipped", file=file)
+            continue
+        gdalinfo_result = gdal_info(file)
+        item = create_item(
+            file,
+            format_date(arguments.start_datetime),
+            format_date(arguments.end_datetime),
+            arguments.collection_id,
+            gdalinfo_result,
+        )
+        tmp_file_path = os.path.join("/tmp/", f"{item.stac['id']}.json")
+        write(tmp_file_path, json.dumps(item.stac).encode("utf-8"))
+        get_log().info("imagery_stac_item_created", file=file)
+
+        collection.add_item(item)
+
+        tmp_file_path = os.path.join("/tmp/", "collection.json")
+        write(tmp_file_path, json.dumps(collection.stac).encode("utf-8"))
+
+
+def create_item(
+    file: str,
+    start_datetime: str,
+    end_datetime: str,
+    collection_id: str,
+    gdalinfo_result: Optional[Dict[Any, Any]] = None,
+) -> ImageryItem:
+    id_ = get_file_name_from_path(file)
+
+    if not gdalinfo_result:
+        gdalinfo_result = gdal_info(file)
+
+    geometry, bbox = get_extents(gdalinfo_result)
+
+    item = ImageryItem(id_, file)
+    item.update_datetime(start_datetime, end_datetime)
+    item.update_spatial(geometry, bbox)
+    item.add_collection(collection_id)
+
+    return item
+
+
+if __name__ == "__main__":
+    main()

scripts/stac/util/geotiff.py → scripts/files/geotiff.py

@@ -1,10 +1,8 @@
-from typing import List, Tuple
+from typing import Any, Dict, List, Tuple
 
-from scripts.gdal.gdalinfo import gdal_info
 
-
-def get_extents(path: str) -> Tuple[List[List[float]], List[float]]:
-    corner_coordinates = gdal_info(path)["cornerCoordinates"]
+def get_extents(gdalinfo_result: Dict[Any, Any]) -> Tuple[List[List[float]], List[float]]:
+    corner_coordinates = gdalinfo_result["cornerCoordinates"]
 
     upper_left = [corner_coordinates["upperLeft"][0], corner_coordinates["upperLeft"][1]]
     upper_right = [corner_coordinates["upperRight"][0], corner_coordinates["upperRight"][1]]

scripts/stac/imagery/collection.py

@@ -1,31 +1,37 @@
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 import ulid
 
 from scripts.stac.util.STAC_VERSION import STAC_VERSION
 
+if TYPE_CHECKING:
+    from scripts.stac.imagery.item import ImageryItem
+
 
 class ImageryCollection:
     stac: Dict[str, Any]
 
-    def __init__(
-        self, title: Optional[str] = None, description: Optional[str] = None, stac: Optional[Dict[str, Any]] = None
-    ) -> None:
-        if stac:
-            self.stac = stac
-        elif title and description:
-            self.stac = {
-                "type": "Collection",
-                "stac_version": STAC_VERSION,
-                "id": str(ulid.ULID()),
-                "title": title,
-                "description": description,
-                "license": "CC-BY-4.0",
-                "links": [{"rel": "self", "href": "./collection.json", "type": "application/json"}],
-            }
-        else:
-            raise Exception("incorrect initialising parameters must have 'stac' or 'title and description'")
+    def __init__(self, title: str, description: str, collection_id: Optional[str] = None) -> None:
+        if not collection_id:
+            collection_id = str(ulid.ULID())
+
+        self.stac = {
+            "type": "Collection",
+            "stac_version": STAC_VERSION,
+            "id": collection_id,
+            "title": title,
+            "description": description,
+            "license": "CC-BY-4.0",
+            "links": [{"rel": "self", "href": "./collection.json", "type": "application/json"}],
+        }
+
+    def add_item(self, item: "ImageryItem") -> None:
+        item_self_link = next((feat for feat in item.stac["links"] if feat["rel"] == "self"), None)
+        if item_self_link:
+            self.add_link(href=item_self_link["href"])
+            self.update_temporal_extent(item.stac["properties"]["start_datetime"], item.stac["properties"]["end_datetime"])
+            self.update_spatial_extent(item.stac["bbox"])
 
     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})
@@ -34,11 +40,12 @@ def update_spatial_extent(self, item_bbox: List[float]) -> None:
         if "extent" not in self.stac:
             self.update_extent(bbox=item_bbox)
             return
-        if not self.stac["extent"]["spatial"]["bbox"]:
+        if self.stac["extent"]["spatial"]["bbox"] == [None]:
             self.update_extent(bbox=item_bbox)
             return
 
-        bbox = self.stac["extent"]["spatial"]["bbox"]
+        bbox = self.stac["extent"]["spatial"]["bbox"][0]
+
         min_x = min(bbox[0], bbox[2])
         max_x = max(bbox[0], bbox[2])
         min_y = min(bbox[1], bbox[3])
@@ -64,11 +71,11 @@ def update_temporal_extent(self, item_start_datetime: str, item_end_datetime: st
         if "extent" not in self.stac:
             self.update_extent(interval=[item_start_datetime, item_end_datetime])
             return
-        if not self.stac["extent"]["temporal"]["interval"]:
+        if self.stac["extent"]["temporal"]["interval"] == [None]:
             self.update_extent(interval=[item_start_datetime, item_end_datetime])
             return
 
-        interval = self.stac["extent"]["temporal"]["interval"]
+        interval = self.stac["extent"]["temporal"]["interval"][0]
 
         item_start = datetime.strptime(item_start_datetime, "%Y-%m-%dT%H:%M:%SZ")
         item_end = datetime.strptime(item_end_datetime, "%Y-%m-%dT%H:%M:%SZ")
@@ -96,12 +103,12 @@ def update_extent(self, bbox: Optional[List[float]] = None, interval: Optional[L
         if "extent" not in self.stac:
             self.stac["extent"] = {
                 "spatial": {
-                    "bbox": bbox,
+                    "bbox": [bbox],
                 },
-                "temporal": {"interval": interval},
+                "temporal": {"interval": [interval]},
             }
             return
         if bbox:
-            self.stac["extent"]["spatial"]["bbox"] = bbox
+            self.stac["extent"]["spatial"]["bbox"] = [bbox]
         if interval:
-            self.stac["extent"]["temporal"]["interval"] = interval
+            self.stac["extent"]["temporal"]["interval"] = [interval]

scripts/stac/imagery/item.py

@@ -1,6 +1,6 @@
-from typing import Any, Dict, List, Optional
+import os
+from typing import Any, Dict, List
 
-from scripts.stac.imagery.collection import ImageryCollection
 from scripts.stac.util import checksum
 from scripts.stac.util.STAC_VERSION import STAC_VERSION
 from scripts.stac.util.stac_extensions import StacExtensions
@@ -9,28 +9,23 @@
 class ImageryItem:
     stac: Dict[str, Any]
 
-    def __init__(self, id_: Optional[str] = None, path: Optional[str] = None, stac: Optional[Dict[str, Any]] = None) -> None:
-        if stac:
-            self.stac = stac
-        elif id_ and path:
-            self.stac = {
-                "type": "Feature",
-                "stac_version": STAC_VERSION,
-                "id": id_,
-                "links": [
-                    {"rel": "self", "href": f"./{id_}.json", "type": "application/json"},
-                ],
-                "assets": {
-                    "visual": {
-                        "href": path,
-                        "type": "image/tiff; application:geotiff; profile:cloud-optimized",
-                        "file:checksum": checksum.multihash_as_hex(path),
-                    }
-                },
-                "stac_extensions": [StacExtensions.file.value],
-            }
-        else:
-            raise Exception("incorrect initialising parameters must have 'stac' or 'id_ and path'")
+    def __init__(self, id_: str, file: str) -> None:
+        self.stac = {
+            "type": "Feature",
+            "stac_version": STAC_VERSION,
+            "id": id_,
+            "links": [
+                {"rel": "self", "href": f"./{id_}.json", "type": "application/json"},
+            ],
+            "assets": {
+                "visual": {
+                    "href": os.path.join(".", os.path.basename(file)),
+                    "type": "image/tiff; application:geotiff; profile:cloud-optimized",
+                    "file:checksum": checksum.multihash_as_hex(file),
+                }
+            },
+            "stac_extensions": [StacExtensions.file.value],
+        }
 
     def update_datetime(self, start_datetime: str, end_datetime: str) -> None:
         self.stac["properties"] = {
@@ -43,10 +38,10 @@ def update_spatial(self, geometry: List[List[float]], bbox: List[float]) -> None
         self.stac["geometry"] = {"type": "Polygon", "coordinates": [geometry]}
         self.stac["bbox"] = bbox
 
-    def add_collection(self, collection: ImageryCollection, path: str) -> None:
-        self.stac["collection"] = collection.stac["title"]
-        self.add_link(rel="collection", href=path)
-        self.add_link(rel="parent", href=path)
+    def add_collection(self, collection_id: str) -> None:
+        self.stac["collection"] = collection_id
+        self.add_link(rel="collection")
+        self.add_link(rel="parent")
 
-    def add_link(self, rel: str, href: str, file_type: str = "application/json") -> None:
+    def add_link(self, rel: str, href: str = "./collection.json", file_type: str = "application/json") -> None:
         self.stac["links"].append({"rel": rel, "href": href, "type": file_type})