WIP

scripts/stac/imagery/capture_area.py

@@ -4,6 +4,7 @@
 from linz_logger import get_log
 from shapely import BufferCapStyle, BufferJoinStyle, Geometry, to_geojson, union_all
 from shapely.geometry import Polygon
+from shapely.geometry.base import BaseGeometry
 
 from scripts.logging.time_helper import time_in_ms
 
@@ -43,19 +44,19 @@ def get_buffer_distance(gsd: float) -> float:
     return gsd * 2 * DECIMAL_DEGREES_1M
 
 
-def to_feature(geometry: Geometry) -> Dict[str, Any]:
-    """Transform a Geometry to a GeoJSON feature.
+def to_feature(geometry: BaseGeometry) -> Dict[str, Any]:
+    """Transform a BaseGeometry to a GeoJSON feature.
 
     Args:
-        geometry: a Geometry
+        geometry: a BaseGeometry
 
     Returns:
         a GeoJSON document.
     """
     return {"geometry": json.loads(to_geojson(geometry)), "type": "Feature", "properties": {}}
 
 
-def merge_polygons(polygons: List[Polygon], buffer_distance: float) -> Geometry:
+def merge_polygons(polygons: List[Polygon], buffer_distance: float) -> BaseGeometry:
     """Merge a list of polygons by converting them to a single geometry that covers the same area.
     A buffer distance is used to buffer out the polygons before dissolving them together and then negative buffer them back in.
     The merged geometry is simplify (rounded) to the decimal used for the buffer.
@@ -80,8 +81,8 @@ def merge_polygons(polygons: List[Polygon], buffer_distance: float) -> Geometry:
     return union_simplified
 
 
-def generate_capture_area(polygons: List[Polygon], gsd: float) -> Dict[str, Any]:
-    """Generate the capture area from a list of polygons.
+def generate_capture_area(polygons: List[BaseGeometry], gsd: float) -> Dict[str, Any]:
+    """Generate the capture area from a list of BaseGeometries.
     Providing the `gsd` allows to round the geometry as we've seen some tiffs
     geometry being slightly off, sometimes due to rounding issue in their
     creation process (before delivery).
@@ -92,7 +93,7 @@ def generate_capture_area(polygons: List[Polygon], gsd: float) -> Dict[str, Any]
     but < buffer_factor*2 will be closed.
 
     Args:
-        polygons: list of polygons of the area
+        polygons: list of BaseGeometries of the area
         gsd: Ground Sample Distance in meters
 
     Returns:

scripts/stac/imagery/collection.py

@@ -1,8 +1,8 @@
 import os
 from datetime import datetime
+from shapely.geometry.base import BaseGeometry
 from typing import Any, Callable, Dict, List, Optional
 
-import shapely.ops
 import ulid
 
 from scripts.datetimes import format_rfc_3339_datetime_string, parse_rfc_3339_datetime
@@ -87,12 +87,12 @@ def __init__(
 
         self.add_providers(providers)
 
-    def add_capture_area(self, polygons: List[shapely.geometry.shape], target: str, artifact_target: str = "/tmp") -> None:
+    def add_capture_area(self, polygons: List[BaseGeometry], target: str, artifact_target: str = "/tmp") -> None:
         """Add the capture area of the Collection.
         The `href` or path of the capture-area.geojson is always set as the relative `./capture-area.geojson`
 
         Args:
-            polygons: list of geometries
+            polygons: list of BaseGeometries
             target: location where the capture-area.geojson file will be saved
             artifact_target: location where the capture-area.geojson artifact file will be saved.
             This is useful for Argo Workflow in order to expose the file to the user for testing/validation purpose.