EP-3981 #114 DriverVectorCube: add from_geojson and to_wkt

Author: soxofaan

openeo_driver/datacube.py

@@ -182,14 +182,30 @@ def with_cube(self, cube: xarray.DataArray, flatten_prefix: str = FLATTEN_PREFIX
         return DriverVectorCube(geometries=self._geometries, cube=cube, flatten_prefix=flatten_prefix)
 
     @classmethod
-    def from_fiona(cls, paths: List[str], driver: str, options: dict):
+    def from_fiona(cls, paths: List[str], driver: str, options: dict) -> "DriverVectorCube":
         """Factory to load vector cube data using fiona/GeoPandas."""
         if len(paths) != 1:
             # TODO #114 EP-3981: support multiple paths
             raise FeatureUnsupportedException(message="Loading a vector cube from multiple files is not supported")
         # TODO #114 EP-3981: lazy loading like/with DelayedVector
         return cls(geometries=gpd.read_file(paths[0], driver=driver))
 
+    @classmethod
+    def from_geojson(cls, geojson: dict) -> "DriverVectorCube":
+        """Construct vector cube from GeoJson dict structure"""
+        # TODO support more geojson types?
+        if geojson["type"] in {"Polygon", "MultiPolygon"}:
+            features = [{"type": "Feature", "geometry": geojson, "properties": {}}]
+        elif geojson["type"] in {"Feature"}:
+            features = [geojson]
+        elif geojson["type"] in {"FeatureCollection"}:
+            features = geojson
+        else:
+            raise FeatureUnsupportedException(
+                f"Can not construct DriverVectorCube from {geojson.get('type', type(geojson))!r}"
+            )
+        return cls(geometries=gpd.GeoDataFrame.from_features(features))
+
     def _as_geopandas_df(self) -> gpd.GeoDataFrame:
         """Join geometries and cube as a geopandas dataframe"""
         # TODO: avoid copy?
@@ -213,6 +229,11 @@ def _as_geopandas_df(self) -> gpd.GeoDataFrame:
     def to_geojson(self):
         return shapely.geometry.mapping(self._as_geopandas_df())
 
+    def to_wkt(self) -> Tuple[List[str], str]:
+        wkts = [str(g) for g in self._geometries.geometry]
+        crs = self._geometries.crs.to_string() if self._geometries.crs else "EPSG:4326"
+        return wkts, crs
+
     def write_assets(
             self, directory: Union[str, Path], format: str, options: Optional[dict] = None
     ) -> Dict[str, StacAsset]:

tests/test_vectorcube.py

@@ -60,6 +60,13 @@ def test_to_geojson(self, gdf):
             ]
         })
 
+    def test_to_wkt(self, gdf):
+        vc = DriverVectorCube(gdf)
+        assert vc.to_wkt() == (
+            ['POLYGON ((1 1, 3 1, 2 3, 1 1))', 'POLYGON ((4 2, 5 4, 3 4, 4 2))'],
+            'EPSG:4326',
+        )
+
     def test_with_cube_to_geojson(self, gdf):
         vc1 = DriverVectorCube(gdf)
         dims, coords = vc1.get_xarray_cube_basics()
@@ -97,3 +104,75 @@ def test_with_cube_to_geojson(self, gdf):
                 }),
             ]
         })
+
+    @pytest.mark.parametrize(["geojson", "expected"], [
+        (
+                {"type": "Polygon", "coordinates": [[(1, 1), (3, 1), (2, 3), (1, 1)]]},
+                [
+                    DictSubSet({
+                        "type": "Feature",
+                        "geometry": {"type": "Polygon", "coordinates": (((1, 1), (3, 1), (2, 3), (1, 1),),)},
+                        "properties": {},
+                    }),
+                ],
+        ),
+        (
+                {"type": "MultiPolygon", "coordinates": [[[(1, 1), (3, 1), (2, 3), (1, 1)]]]},
+                [
+                    DictSubSet({
+                        "type": "Feature",
+                        "geometry": {"type": "MultiPolygon", "coordinates": [(((1, 1), (3, 1), (2, 3), (1, 1),),)]},
+                        "properties": {},
+                    }),
+                ],
+        ),
+        (
+                {
+                    "type": "Feature",
+                    "geometry": {"type": "MultiPolygon", "coordinates": [[[(1, 1), (3, 1), (2, 3), (1, 1)]]]},
+                    "properties": {"id": "12_3"},
+                },
+                [
+                    DictSubSet({
+                        "type": "Feature",
+                        "geometry": {"type": "MultiPolygon", "coordinates": [(((1, 1), (3, 1), (2, 3), (1, 1),),)]},
+                        "properties": {"id": "12_3"},
+                    }),
+                ],
+        ),
+        (
+                {
+                    "type": "FeatureCollection",
+                    "features": [
+                        {
+                            "type": "Feature",
+                            "geometry": {"type": "Polygon", "coordinates": [[(1, 1), (3, 1), (2, 3), (1, 1)]]},
+                            "properties": {"id": 1},
+                        },
+                        {
+                            "type": "Feature",
+                            "geometry": {"type": "MultiPolygon", "coordinates": [[[(1, 1), (3, 1), (2, 3), (1, 1)]]]},
+                            "properties": {"id": 2},
+                        },
+                    ],
+                },
+                [
+                    DictSubSet({
+                        "type": "Feature",
+                        "geometry": {"type": "Polygon", "coordinates": (((1, 1), (3, 1), (2, 3), (1, 1),),)},
+                        "properties": {"id": 1},
+                    }),
+                    DictSubSet({
+                        "type": "Feature",
+                        "geometry": {"type": "MultiPolygon", "coordinates": [(((1, 1), (3, 1), (2, 3), (1, 1),),)]},
+                        "properties": {"id": 2},
+                    }),
+                ],
+        ),
+    ])
+    def test_from_geojson(self, geojson, expected):
+        vc = DriverVectorCube.from_geojson(geojson)
+        assert vc.to_geojson() == DictSubSet({
+            "type": "FeatureCollection",
+            "features": expected,
+        })