remove delayedvector and fiona

CHANGELOG.md

@@ -25,6 +25,11 @@ and start a new "In Progress" section above it.
 
 - Require at least Python 3.11 (drop support for Python 3.8, 3.9 and 3.10 [#370](https://github.com/Open-EO/openeo-python-driver/issues/370))
 - Require at least PySTAC 1.10 ([#396](https://github.com/Open-EO/openeo-python-driver/issues/396))
+- Replace `DelayedVector` with `DriverVectorCube` in `read_vector`, `get_geometries`, `vector_buffer` and the dummy backend's `aggregate_spatial` (vector files are now loaded through `DriverVectorCube.from_fiona`) and remove the `DelayedVector` handling from `apply_polygon`, `chunk_polygon`, `mask_polygon`, `filter_spatial`, `aggregate_spatial` and `run_udf`
+- `read_vector` and `get_geometries` (with `filename`) now return a `DriverVectorCube` (built with the backend's `vector_cube_cls`) instead of a `DelayedVector`.
+- Vector files are now loaded eagerly (instead of lazily) and loading errors are reported as `load_url_reading_error` instead of the former GeoJSON parsing errors.
+- **Backwards incompatible changes:**
+    - `run_udf` on the result of `read_vector`/`get_geometries` no longer passes the geometries as feature collection to the UDF, but follows the generic vector cube handling (like `load_uploaded_files`).
 
 
 ## 0.141.0

openeo_driver/datacube.py

@@ -714,10 +714,6 @@ def to_legacy_save_result(self) -> Union["AggregatePolygonResult", "JSONResult"]
         # TODO: eliminate these legacy, non-standard formats?
         from openeo_driver.save_result import AggregatePolygonResult, JSONResult
 
-        if self._cube is None or self._cube.attrs.get(self.CUBE_ATTR_VECTOR_CUBE_DUMMY):
-            # No cube: no real data to return (in legacy style), so let's just return a `null` per geometry.
-            return JSONResult(data=[None] * self.geometry_count())
-
         cube = self._cube
         # TODO: more flexible temporal/band dimension detection?
         if cube.dims == (self.DIM_GEOMETRY, self.DIM_TIME):
@@ -733,11 +729,9 @@ def to_legacy_save_result(self) -> Union["AggregatePolygonResult", "JSONResult"]
             # we keep it simple here with a basic JSONResult result.
             cube = cube.transpose(self.DIM_GEOMETRY, self.DIM_BANDS)
             return JSONResult(data=cube.values.tolist())
-        elif cube.dims == (self.DIM_GEOMETRY, self.DIM_PROPERTIES):
+        else:
             return JSONResult(data=self.to_geojson(include_properties=True))
-        raise ValueError(
-            f"Unsupported cube configuration {cube.dims} for _write_legacy_aggregate_polygon_result_json"
-        )
+
 
     def get_dimension_names(self) -> List[str]:
         if self._cube is None:

openeo_driver/delayed_vector.py

@@ -6,7 +6,6 @@
 from typing import Iterable, List, Dict
 from urllib.parse import urlparse
 
-import fiona
 import geopandas as gpd
 import pyproj
 import requests
@@ -261,14 +260,18 @@ def save_as(src_url: str, dest_path: str):
     @staticmethod
     def _read_shapefile_geometries(shp_path: str) -> List[BaseGeometry]:
         # FIXME: returned as a list for safety but possible to return as an iterable?
-        with fiona.open(shp_path) as collection:
-            [validate_geojson_coordinates(record["geometry"]) for record in collection]
-            return [shapely.geometry.shape(record["geometry"]) for record in collection]
+        geometries = [
+            geometry
+            for geometry in pyogrio.read_dataframe(shp_path, read_geometry=True).geometry
+            if geometry is not None
+        ]
+        for geometry in geometries:
+            validate_geojson_coordinates(shapely.geometry.mapping(geometry))
+        return geometries
 
     @staticmethod
-    def _read_shapefile_bounds(shp_path: str) -> List[BaseGeometry]:
-        with fiona.open(shp_path) as collection:
-            return collection.bounds
+    def _read_shapefile_bounds(shp_path: str) -> tuple:
+        return tuple(pyogrio.read_info(shp_path)["total_bounds"])
 
     @staticmethod
     def _read_shapefile_crs(shp_path: str) -> pyproj.CRS:
@@ -277,8 +280,8 @@ def _read_shapefile_crs(shp_path: str) -> pyproj.CRS:
         @param shp_path:
         @return: CRS as a proj4 dict
         """
-        with fiona.open(shp_path) as collection:
-            return collection.crs
+        crs = pyogrio.read_info(shp_path)["crs"]
+        return None if crs is None else pyproj.CRS.from_user_input(crs)
 
     @staticmethod
     def _as_geometry_collection(feature_collection: Dict) -> Dict:

openeo_driver/dry_run.py

@@ -68,7 +68,6 @@
 from openeo_driver.constants import RESAMPLE_SPATIAL_ALIGN_DEFAULT
 from openeo_driver.datacube import DriverDataCube, DriverVectorCube
 from openeo_driver.datastructs import ResolutionMergeArgs, SarBackscatterArgs
-from openeo_driver.delayed_vector import DelayedVector
 from openeo_driver.errors import FeatureUnsupportedException, OpenEOApiException
 from openeo_driver.util.geometry import (
     BoundingBox,
@@ -559,8 +558,8 @@ def get_source_constraints(self, merge=True) -> List[SourceConstraint]:
 
     def get_geometries(
         self, operation="aggregate_spatial"
-    ) -> List[Union[shapely.geometry.base.BaseGeometry, DelayedVector, DriverVectorCube]]:
-        """Get geometries (polygons or DelayedVector), as used by aggregate_spatial"""
+    ) -> List[Union[shapely.geometry.base.BaseGeometry, DriverVectorCube]]:
+        """Get geometries (polygons or vector cube), as used by aggregate_spatial"""
         geometries_by_id = {}
         for leaf in self.get_trace_leaves():
             for args in leaf.get_arguments_by_operation(operation):
@@ -572,8 +571,8 @@ def get_geometries(
 
     def get_last_geometry(
         self, operation="aggregate_spatial"
-    ) -> Union[shapely.geometry.base.BaseGeometry, DelayedVector, DriverVectorCube]:
-        """Get geometries (polygons or DelayedVector), as used by aggregate_spatial"""
+    ) -> Union[shapely.geometry.base.BaseGeometry, DriverVectorCube]:
+        """Get geometries (polygons or vector cube), as used by aggregate_spatial"""
 
         for leaf in self.get_trace_leaves():
             args = leaf.get_arguments_by_operation(operation)
@@ -705,12 +704,12 @@ def aggregate_spatial(
 
     def _normalize_geometry(
         self, geometries, target_crs=None, target_resolution = None
-    ) -> Tuple[Union[DriverVectorCube, DelayedVector, BaseGeometry], dict]:
+    ) -> Tuple[Union[DriverVectorCube, BaseGeometry], dict]:
         """
         Helper to preprocess geometries (as used in aggregate_spatial and mask_polygon)
         and extract bbox (e.g. for filter_bbox)
 
-        :param geometries: geometries as BaseGeometry, GeoJSON dict, DelayedVector or DriverVectorCube
+        :param geometries: geometries as BaseGeometry, GeoJSON dict, path (str) or DriverVectorCube
         :param target_crs: target CRS to reproject geometries to
         :param target_resolution: target resolution for geometries, in units of the target CRS, None by default which will use 10m
         """
@@ -756,9 +755,7 @@ def _normalize_geometry(
         elif isinstance(geometries, dict):
             return self._normalize_geometry(geojson_to_geometry(geometries),target_crs, target_resolution)
         elif isinstance(geometries, str):
-            return self._normalize_geometry(DelayedVector(geometries),target_crs, target_resolution)
-        elif isinstance(geometries, DelayedVector):
-            bbox = geometries.bounds
+            return self._normalize_geometry(DriverVectorCube.from_fiona([geometries]), target_crs, target_resolution)
         elif isinstance(geometries, shapely.geometry.base.BaseGeometry):
             # TODO: buffer distance of 10m assumes certain resolution (e.g. sentinel2 pixels)
             # TODO: use proper distance for collection resolution instead of using a default distance?

openeo_driver/dummy/dummy_backend.py

@@ -54,7 +54,6 @@
 from openeo_driver.constants import JOB_STATUS, STAC_EXTENSION, DEFAULT_LOG_LEVEL_RETRIEVAL
 from openeo_driver.datacube import DriverDataCube, DriverMlModel, DriverVectorCube, SaveResultHints
 from openeo_driver.datastructs import StacAsset
-from openeo_driver.delayed_vector import DelayedVector
 from openeo_driver.dry_run import SourceConstraint
 from openeo_driver.errors import (
     FeatureUnsupportedException,
@@ -310,7 +309,7 @@ def assert_polygon_sequence(geometries: Union[Sequence, BaseMultipartGeometry])
             )
             return geometries.with_cube(cube=cube)
         elif isinstance(geometries, str):
-            geometries = [geometry for geometry in DelayedVector(geometries).geometries]
+            geometries = list(DriverVectorCube.from_fiona([geometries]).get_geometries())
             n_geometries = assert_polygon_sequence(geometries)
         elif isinstance(geometries, GeometryCollection):
             # TODO #71 #114 EP-3981: GeometryCollection is deprecated

openeo_driver/processgraph/process_implementations/cubes.py

@@ -4,6 +4,7 @@
 
 import geopandas as gpd
 import numpy as np
+import pyproj
 import shapely.geometry
 import shapely.ops
 from shapely.geometry import GeometryCollection, MultiPolygon
@@ -12,7 +13,6 @@
 from openeo.util import rfc3339
 
 from openeo_driver.datacube import DriverDataCube, DriverVectorCube, SupportsRunUdf
-from openeo_driver.delayed_vector import DelayedVector
 from openeo_driver.dry_run import DryRunDataCube, DryRunDataTracer
 from openeo_driver.errors import (
     FeatureUnsupportedException,
@@ -155,16 +155,7 @@ def apply_polygon(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
     mask_value = args.get_optional("mask_value", expected_type=(int, float), default=None)
     context = args.get_optional("context", default=None)
 
-    if isinstance(geometries, DelayedVector):
-        geometries = list(geometries.geometries)
-        for p in geometries:
-            if not isinstance(p, shapely.geometry.Polygon):
-                raise ProcessParameterInvalidException(
-                    parameter="polygons", process="apply_polygon",
-                    reason="{m!s} is not a polygon.".format(m=p)
-                )
-        polygon = MultiPolygon(geometries)
-    elif isinstance(geometries, DriverVectorCube):
+    if isinstance(geometries, DriverVectorCube):
         polygon = geometries.to_multipolygon()
     elif isinstance(geometries, shapely.geometry.base.BaseGeometry):
         polygon = MultiPolygon(geometries)
@@ -196,8 +187,8 @@ def chunk_polygon(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
     mask_value = args.get_optional("mask_value", expected_type=(int, float), default=None)
     context = args.get_optional("context", default=None)
 
-    if isinstance(chunks, DelayedVector):
-        polygons = list(chunks.geometries)
+    if isinstance(chunks, DriverVectorCube):
+        polygons = list(chunks.get_geometries())
         for p in polygons:
             if not isinstance(p, shapely.geometry.Polygon):
                 raise ProcessParameterInvalidException(
@@ -271,9 +262,7 @@ def mask_polygon(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
     replacement = args.get_optional("replacement", default=None)
     inside = args.get_optional("inside", default=False)
 
-    if isinstance(mask_geom, DelayedVector):
-        polygon = shapely.ops.unary_union(list(mask_geom.geometries))
-    elif isinstance(mask_geom, DriverVectorCube):
+    if isinstance(mask_geom, DriverVectorCube):
         polygon = mask_geom.to_multipolygon()
     elif isinstance(mask_geom, dict) and "type" in mask_geom:
         polygon = geojson_to_multipolygon(mask_geom)
@@ -372,8 +361,6 @@ def filter_spatial(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
                     for geom in geometries.geoms
                 ]
                 geometries = MultiPolygon(polygons)
-    elif isinstance(geometries, DelayedVector):
-        geometries = DriverVectorCube.from_fiona([geometries.path]).to_multipolygon()
     elif isinstance(geometries, DriverVectorCube):
         pass
     else:
@@ -463,7 +450,7 @@ def aggregate_spatial(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
                 parameter="geometries", process="aggregate_spatial",
                 reason="Failed to parse inline GeoJSON",
             )
-    elif isinstance(geoms, (AggregatePolygonResult, DelayedVector)):
+    elif isinstance(geoms, AggregatePolygonResult):
         geoms = geoms.to_driver_vector_cube()
     else:
         raise ProcessParameterInvalidException(
@@ -493,6 +480,17 @@ def reduce_spatial(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
     return cube.reduce_spatial(reducer=reduce_pg, context=context, env=env)
 
 
+def _geojson_geometries(geojson: dict) -> List[shapely.geometry.base.BaseGeometry]:
+    """Extract geometries (ignoring properties) from a GeoJSON dict."""
+    geojson_type = geojson.get("type")
+    if geojson_type == "FeatureCollection":
+        return [shapely.geometry.shape(f["geometry"]) for f in geojson["features"]]
+    elif geojson_type == "GeometryCollection":
+        return [shapely.geometry.shape(g) for g in geojson["geometries"]]
+    elif geojson_type == "Feature":
+        return [shapely.geometry.shape(geojson["geometry"])]
+    return [shapely.geometry.shape(geojson)]
+
 @process
 def run_udf(args: ProcessArgs, env: EvalEnv):
     dry_run_tracer: DryRunDataTracer = env.get(ENV_DRY_RUN_TRACER)
@@ -518,14 +516,16 @@ def run_udf(args: ProcessArgs, env: EvalEnv):
         pass
     if isinstance(data, DriverVectorCube):
         data = data.to_legacy_save_result()
-
-    if isinstance(data, (DelayedVector, dict)):
-        if isinstance(data, dict):
-            data = DelayedVector.from_json_dict(data)
-        collection = openeo.udf.FeatureCollection(id='VectorCollection', data=data.as_geodataframe())
+    if isinstance(data, dict):
+        # Inline GeoJSON: pass geometries as feature collection to the UDF.
+        crs = pyproj.CRS(data.get("crs", {}).get("properties", {}).get("name", "EPSG:4326"))
+        gdf = gpd.GeoDataFrame(geometry=_geojson_geometries(data), crs=crs)
+        collection = openeo.udf.FeatureCollection(id="VectorCollection", data=gdf)
         data = openeo.udf.UdfData(
-            proj={"EPSG": data.crs.to_epsg()}, feature_collection_list=[collection], user_context=context
+            proj={"EPSG": crs.to_epsg()}, feature_collection_list=[collection], user_context=context
         )
+    elif isinstance(data, openeo.udf.UdfData):
+        pass
     elif isinstance(data, JSONResult):
         st = openeo.udf.StructuredData(description="Dictionary data", data=data.get_data(), type="dict")
         data = openeo.udf.UdfData(structured_data_list=[st], user_context=context)

openeo_driver/processgraph/process_implementations/geometry.py

@@ -8,7 +8,6 @@
 import shapely.geometry
 
 from openeo_driver.datacube import DriverDataCube, DriverVectorCube
-from openeo_driver.delayed_vector import DelayedVector
 from openeo_driver.dry_run import DryRunDataCube, DryRunDataTracer
 from openeo_driver.errors import FeatureUnsupportedException, ProcessParameterInvalidException
 from openeo_driver.processes import ProcessArgs, ProcessSpec
@@ -33,6 +32,18 @@ def _check_geometry_path_assumption(path: str, process: str, parameter: str):
         )
 
 
+def _read_vector_cube(path: str, env: EvalEnv) -> DriverVectorCube:
+    """Load vector file as vector cube (expanding a top-level GeoJSON GeometryCollection into separate geometries)."""
+    vector_cube_cls = env.backend_implementation.vector_cube_cls
+    vector_cube = vector_cube_cls.from_fiona([path])
+    geometries = vector_cube.get_geometries()
+    if len(geometries) == 1 and isinstance(geometries.iloc[0], shapely.geometry.GeometryCollection):
+        # TODO #71 #114 Deprecate/avoid usage of GeometryCollection
+        _log.warning("Vector data of deprecated type 'GeometryCollection', please use a FeatureCollection instead.")
+        gdf = gpd.GeoDataFrame(geometry=list(geometries.iloc[0].geoms), crs=vector_cube.get_crs())
+        vector_cube = vector_cube_cls.from_geodataframe(gdf)
+    return vector_cube
+
 @non_standard_process(
     ProcessSpec(id="vector_buffer", description="Add a buffer around a geometry.")
     .param(
@@ -69,7 +80,9 @@ def vector_buffer(args: ProcessArgs, env: EvalEnv) -> dict:
         input_crs = geometry.get_crs()
     elif isinstance(geometry, str):
         _check_geometry_path_assumption(path=geometry, process="vector_buffer", parameter="geometry")
-        geoms = list(DelayedVector(geometry).geometries)
+        vector_cube = DriverVectorCube.from_fiona([geometry])
+        geoms = list(vector_cube.get_geometries())
+        input_crs = vector_cube.get_crs()
     elif isinstance(geometry, dict) and "type" in geometry:
         geometry_type = geometry["type"]
         if geometry_type == "FeatureCollection":
@@ -142,7 +155,7 @@ def get_geometries(args: ProcessArgs, env: EvalEnv):
     path = args.get_optional("filename", None)
     if path is not None:
         _check_geometry_path_assumption(path=path, process="get_geometries", parameter="filename")
-        return DelayedVector(path)
+        return _read_vector_cube(path, env)
     else:
         return feature_collection
 
@@ -169,11 +182,11 @@ def get_geometries(args: ProcessArgs, env: EvalEnv):
     )
     .returns("GeoJSON-style feature collection", schema={"type": "object", "subtype": "geojson"})
 )
-def read_vector(args: ProcessArgs, env: EvalEnv) -> DelayedVector:
+def read_vector(args: ProcessArgs, env: EvalEnv) -> DriverVectorCube:
     _log.warning("DEPRECATED: read_vector usage")
     path = args.get_required("filename")
     _check_geometry_path_assumption(path=path, process="read_vector", parameter="filename")
-    return DelayedVector(path)
+    return _read_vector_cube(path, env)
 
 
 @non_standard_process(