Feat: update stac collection TDE-453 (#134)

* feat: initialise collection object and stac

* test: collection test

* fix: appease mypy

* fix: move stac version to its own file for reuse

* feat: create items

* fix: add dependencies

* fix: minor code tidy and add test

* feat: implement update collection

* fix: appease pylint will be reverted in next PR

* test: add more tests

* fix: formatting

* fix: all formatting

* fix: remove additional file

* fix: typo

Co-authored-by: Alice Fage <[email protected]>

scripts/files/files_helper.py

@@ -8,3 +8,7 @@ def get_file_name_from_path(path: str) -> str:
 
 def is_tiff(path: str) -> bool:
     return path.lower().endswith((".tiff", ".tif"))
+
+
+def is_json(path: str) -> bool:
+    return path.lower().endswith(".json")

scripts/stac/imagery/collection.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Any, Dict, List, Optional
 
 import ulid
@@ -27,17 +28,80 @@ def __init__(
             raise Exception("incorrect initialising parameters must have 'stac' or 'title and description'")
 
     def add_link(self, href: str, rel: str = "item", file_type: str = "application/json") -> None:
-        # Will be implemented in Future PR
-        pass
+        self.stac["links"].append({"rel": rel, "href": href, "type": file_type})
 
     def update_spatial_extent(self, item_bbox: List[float]) -> None:
-        # Will be implemented in Future PR
-        pass
+        if "extent" not in self.stac:
+            self.update_extent(bbox=item_bbox)
+            return
+        if not self.stac["extent"]["spatial"]["bbox"]:
+            self.update_extent(bbox=item_bbox)
+            return
+
+        bbox = self.stac["extent"]["spatial"]["bbox"]
+        min_x = min(bbox[0], bbox[2])
+        max_x = max(bbox[0], bbox[2])
+        min_y = min(bbox[1], bbox[3])
+        max_y = max(bbox[1], bbox[3])
+
+        item_min_x = min(item_bbox[0], item_bbox[2])
+        item_max_x = max(item_bbox[0], item_bbox[2])
+        item_min_y = min(item_bbox[1], item_bbox[3])
+        item_max_y = max(item_bbox[1], item_bbox[3])
+
+        if item_min_x < min_x:
+            min_x = item_min_x
+        if item_min_y < min_y:
+            min_y = item_min_y
+        if item_max_x > max_x:
+            max_x = item_max_x
+        if item_max_y > max_y:
+            max_y = item_max_y
+
+        self.update_extent(bbox=[min_x, min_y, max_x, max_y])
 
     def update_temporal_extent(self, item_start_datetime: str, item_end_datetime: str) -> None:
-        # Will be implemented in Future PR
-        pass
+        if "extent" not in self.stac:
+            self.update_extent(interval=[item_start_datetime, item_end_datetime])
+            return
+        if not self.stac["extent"]["temporal"]["interval"]:
+            self.update_extent(interval=[item_start_datetime, item_end_datetime])
+            return
+
+        interval = self.stac["extent"]["temporal"]["interval"]
+
+        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")
+
+        collection_datetimes = []
+        for date in interval:
+            collection_datetimes.append(datetime.strptime(date, "%Y-%m-%dT%H:%M:%SZ"))
+
+        start_datetime = min(collection_datetimes[0], collection_datetimes[1])
+        end_datetime = max(collection_datetimes[0], collection_datetimes[1])
+
+        if item_start < start_datetime:
+            start_datetime = item_start
+        if item_end > end_datetime:
+            end_datetime = item_end
+
+        self.update_extent(
+            interval=[
+                start_datetime.strftime("%Y-%m-%dT%H:%M:%SZ"),
+                end_datetime.strftime("%Y-%m-%dT%H:%M:%SZ"),
+            ]
+        )
 
     def update_extent(self, bbox: Optional[List[float]] = None, interval: Optional[List[str]] = None) -> None:
-        # Will be implemented in Future PR
-        pass
+        if "extent" not in self.stac:
+            self.stac["extent"] = {
+                "spatial": {
+                    "bbox": bbox,
+                },
+                "temporal": {"interval": interval},
+            }
+            return
+        if bbox:
+            self.stac["extent"]["spatial"]["bbox"] = bbox
+        if interval:
+            self.stac["extent"]["temporal"]["interval"] = interval

scripts/stac/tests/collection_test.py

@@ -8,3 +8,36 @@ def test_imagery_stac_collection_initialise() -> None:
 
     assert collection.stac["title"] == title
     assert collection.stac["description"] == description
+
+
+def test_imagery_stac_collection_update() -> None:
+    title = "Test Urban Imagery"
+    description = "Test Urban Imagery Description"
+    bbox = [1799667.5, 5815977.0, 1800422.5, 5814986.0]
+    start_datetime = "2021-01-27T00:00:00Z"
+    end_datetime = "2021-01-27T00:00:00Z"
+    collection = ImageryCollection(title, description)
+    collection.update_spatial_extent(bbox)
+    collection.update_temporal_extent(start_datetime, end_datetime)
+    assert collection.stac["extent"]["temporal"]["interval"] == [start_datetime, end_datetime]
+
+
+def test_imagery_stac_collection_update_twice() -> None:
+    title = "Test Urban Imagery"
+    description = "Test Urban Imagery Description"
+    collection = ImageryCollection(title, description)
+
+    bbox_one = [174.889641, -41.217532, 174.902344, -41.203521]
+    start_datetime_one = "2021-01-27T00:00:00Z"
+    end_datetime_one = "2021-01-27T00:00:00Z"
+    collection.update_spatial_extent(bbox_one)
+    collection.update_temporal_extent(start_datetime_one, end_datetime_one)
+
+    bbox_two = [174.917643, -41.211157, 174.922965, -41.205490]
+    start_datetime_two = "2021-02-01T00:00:00Z"
+    end_datetime_two = "2021-02-20T00:00:00Z"
+    collection.update_spatial_extent(bbox_two)
+    collection.update_temporal_extent(start_datetime_two, end_datetime_two)
+
+    assert collection.stac["extent"]["temporal"]["interval"] == [start_datetime_one, end_datetime_two]
+    assert collection.stac["extent"]["spatial"]["bbox"] == [174.889641, -41.217532, 174.922965, -41.203521]

scripts/standardise_validate.py

@@ -10,10 +10,10 @@
 def main() -> None:
 
     concurrency: int = 1
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--preset", dest="preset", required=True)
-    parser.add_argument("--source", dest="source", nargs="+", required=True)
-    arguments = parser.parse_args()
+    parse_args = argparse.ArgumentParser()
+    parse_args.add_argument("--preset", dest="preset", required=True)
+    parse_args.add_argument("--source", dest="source", nargs="+", required=True)
+    arguments = parse_args.parse_args()
 
     source = format_source(arguments.source)
 

scripts/update_stac_collection.py

@@ -0,0 +1,49 @@
+import argparse
+import json
+from typing import List
+
+from linz_logger import get_log
+
+from scripts.cli.cli_helper import format_source
+from scripts.files.files_helper import is_json
+from scripts.files.fs import read, write
+from scripts.logging.time_helper import time_in_ms
+from scripts.stac.imagery.collection import ImageryCollection
+from scripts.stac.imagery.item import ImageryItem
+
+
+def update_imagery_collection(files: List[str], collection_path: str) -> None:
+    start_time = time_in_ms()
+    get_log().info("finalise_stac_collection_imagery_start", collection=collection_path)
+
+    collection = ImageryCollection(stac=json.loads(read(collection_path)))
+
+    for file in files:
+        if not is_json(file):
+            get_log().trace("create_stac_file_not_tiff_skipped", file=file)
+        item = ImageryItem(stac=json.loads(read(file)))
+        collection.add_link(href=file)
+        collection.update_temporal_extent(item.stac["properties"]["start_datetime"], item.stac["properties"]["end_datetime"])
+        collection.update_spatial_extent(item.stac["bbox"])
+
+    write(collection_path, json.dumps(collection.stac).encode("utf-8"))
+
+    get_log().info(
+        "update_stac_collection_imagery_complete", collection=collection.stac, source=files, duration=time_in_ms() - start_time
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source", dest="source", nargs="+", required=True)
+    parser.add_argument("--collection", dest="collection", help="path to collection.json", required=True)
+    arguments = parser.parse_args()
+
+    files = format_source(arguments.source)
+    collection_path = arguments.collection
+
+    update_imagery_collection(files, collection_path)
+
+
+if __name__ == "__main__":
+    main()