Support collection_property helper in load_stac #246

Author: soxofaan

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Support `collection_property` based property filtering in `load_stac` ([#246](https://github.com/Open-EO/openeo-python-client/issues/246))
+
 ### Changed
 
 - Eliminate deprecated `utcnow` usage patterns. Introduce `Rfc3339.now_utc()` method (as replacement for deprecated `utcnow()` method) to simplify finding deprecated `utcnow` usage in user code. ([#760](https://github.com/Open-EO/openeo-python-client/issues/760))

openeo/rest/connection.py

@@ -1097,7 +1097,7 @@ def load_collection(
         temporal_extent: Union[Sequence[InputDate], Parameter, str, None] = None,
         bands: Union[Iterable[str], Parameter, str, None] = None,
         properties: Union[
-            None, Dict[str, Union[str, PGNode, Callable]], List[CollectionProperty], CollectionProperty
+            Dict[str, Union[PGNode, Callable]], List[CollectionProperty], CollectionProperty, None
         ] = None,
         max_cloud_cover: Optional[float] = None,
         fetch_metadata: bool = True,
@@ -1198,7 +1198,9 @@ def load_stac(
         spatial_extent: Union[dict, Parameter, shapely.geometry.base.BaseGeometry, str, Path, None] = None,
         temporal_extent: Union[Sequence[InputDate], Parameter, str, None] = None,
         bands: Union[Iterable[str], Parameter, str, None] = None,
-        properties: Optional[Dict[str, Union[str, PGNode, Callable]]] = None,
+        properties: Union[
+            Dict[str, Union[PGNode, Callable]], List[CollectionProperty], CollectionProperty, None
+        ] = None,
     ) -> DataCube:
         """
         Loads data from a static STAC catalog or a STAC API Collection and returns the data as a processable :py:class:`DataCube`.
@@ -1292,6 +1294,8 @@ def load_stac(
             The value must be a condition (user-defined process) to be evaluated against a STAC API.
             This parameter is not supported for static STAC.
 
+            See :py:func:`~openeo.rest.graph_building.collection_property` for easy construction of property filters.
+
         .. versionadded:: 0.17.0
 
         .. versionchanged:: 0.23.0
@@ -1300,6 +1304,9 @@ def load_stac(
 
         .. versionchanged:: 0.37.0
             Argument ``spatial_extent``: add support for passing a Shapely geometry or a local path to a GeoJSON file.
+
+        .. versionchanged:: 0.41.0
+            Add support for :py:func:`~openeo.rest.graph_building.collection_property` based property filters
         """
         return DataCube.load_stac(
             url=url,

openeo/rest/datacube.py

@@ -175,7 +175,7 @@ def load_collection(
         bands: Union[Iterable[str], Parameter, str, None] = None,
         fetch_metadata: bool = True,
         properties: Union[
-            None, Dict[str, Union[str, PGNode, typing.Callable]], List[CollectionProperty], CollectionProperty
+            Dict[str, Union[PGNode, typing.Callable]], List[CollectionProperty], CollectionProperty, None
         ] = None,
         max_cloud_cover: Optional[float] = None,
     ) -> DataCube:
@@ -250,27 +250,13 @@ def load_collection(
                 metadata = metadata.filter_bands(bands)
             arguments['bands'] = bands
 
-        if isinstance(properties, list):
-            # TODO: warn about items that are not CollectionProperty objects instead of silently dropping them.
-            properties = {p.name: p.from_node() for p in properties if isinstance(p, CollectionProperty)}
-        if isinstance(properties, CollectionProperty):
-            properties = {properties.name: properties.from_node()}
-        elif properties is None:
-            properties = {}
-        if max_cloud_cover:
-            properties["eo:cloud_cover"] = lambda v: v <= max_cloud_cover
-        if properties:
-            summaries = metadata and metadata.get("summaries") or {}
-            undefined_properties = set(properties.keys()).difference(summaries.keys())
-            if undefined_properties:
-                warnings.warn(
-                    f"{collection_id} property filtering with properties that are undefined "
-                    f"in the collection metadata (summaries): {', '.join(undefined_properties)}.",
-                    stacklevel=2,
-                )
-            arguments["properties"] = {
-                prop: build_child_callback(pred, parent_parameters=["value"]) for prop, pred in properties.items()
-            }
+        properties = cls._build_load_properties_argument(
+            properties=properties,
+            supported_properties=(metadata.get("summaries", default={}).keys() if metadata else None),
+            max_cloud_cover=max_cloud_cover,
+        )
+        if properties is not None:
+            arguments["properties"] = properties
 
         pg = PGNode(
             process_id='load_collection',
@@ -282,6 +268,50 @@ def load_collection(
         load_collection, name="create_collection", since="0.4.6"
     )
 
+    @classmethod
+    def _build_load_properties_argument(
+        cls,
+        properties: Union[
+            Dict[str, Union[PGNode, typing.Callable]],
+            List[CollectionProperty],
+            CollectionProperty,
+            None,
+        ],
+        *,
+        supported_properties: Optional[typing.Collection[str]] = None,
+        max_cloud_cover: Optional[float] = None,
+    ) -> Union[Dict[str, PGNode], None]:
+        """
+        Helper to convert/build the ``properties`` argument
+        for ``load_collection`` and ``load_stac`` from user input
+        """
+        if isinstance(properties, CollectionProperty):
+            properties = [properties]
+        if isinstance(properties, list):
+            if not all(isinstance(p, CollectionProperty) for p in properties):
+                raise ValueError(
+                    f"When providing load properties as a list, all items must be CollectionProperty objects, but got {properties}"
+                )
+            properties = {p.name: p.from_node() for p in properties}
+
+        if max_cloud_cover is not None:
+            properties = properties or {}
+            properties["eo:cloud_cover"] = lambda v: v <= max_cloud_cover
+
+        if isinstance(properties, dict):
+            if supported_properties:
+                unsupported_properties = set(properties.keys()).difference(supported_properties)
+                if unsupported_properties:
+                    warnings.warn(
+                        f"Property filtering with unsupported properties according to collection/STAC metadata: {unsupported_properties} (supported: {supported_properties}).",
+                        stacklevel=3,
+                    )
+            properties = {
+                prop: build_child_callback(pred, parent_parameters=["value"]) for prop, pred in properties.items()
+            }
+
+        return properties
+
     @classmethod
     @deprecated(reason="Depends on non-standard process, replace with :py:meth:`openeo.rest.connection.Connection.load_stac` where possible.",version="0.25.0")
     def load_disk_collection(cls, connection: Connection, file_format: str, glob_pattern: str, **options) -> DataCube:
@@ -314,7 +344,9 @@ def load_stac(
         spatial_extent: Union[dict, Parameter, shapely.geometry.base.BaseGeometry, str, pathlib.Path, None] = None,
         temporal_extent: Union[Sequence[InputDate], Parameter, str, None] = None,
         bands: Union[Iterable[str], Parameter, str, None] = None,
-        properties: Optional[Dict[str, Union[str, PGNode, Callable]]] = None,
+        properties: Union[
+            Dict[str, Union[PGNode, typing.Callable]], List[CollectionProperty], CollectionProperty, None
+        ] = None,
         connection: Optional[Connection] = None,
     ) -> DataCube:
         """
@@ -409,14 +441,19 @@ def load_stac(
             The value must be a condition (user-defined process) to be evaluated against a STAC API.
             This parameter is not supported for static STAC.
 
+            See :py:func:`~openeo.rest.graph_building.collection_property` for easy construction of property filters.
+
         :param connection: The connection to use to connect with the backend.
 
         .. versionadded:: 0.33.0
 
         .. versionchanged:: 0.37.0
             Argument ``spatial_extent``: add support for passing a Shapely geometry or a local path to a GeoJSON file.
+
+        .. versionchanged:: 0.41.0
+            Add support for :py:func:`~openeo.rest.graph_building.collection_property` based property filters
         """
-        arguments = {"url": url}
+        arguments: dict = {"url": url}
         if spatial_extent:
             arguments["spatial_extent"] = _get_geometry_argument(
                 argument=spatial_extent,
@@ -434,10 +471,11 @@ def load_stac(
         bands = cls._get_bands(bands, process_id="load_stac")
         if bands is not None:
             arguments["bands"] = bands
-        if properties:
-            arguments["properties"] = {
-                prop: build_child_callback(pred, parent_parameters=["value"]) for prop, pred in properties.items()
-            }
+
+        properties = cls._build_load_properties_argument(properties=properties)
+        if properties is not None:
+            arguments["properties"] = properties
+
         graph = PGNode("load_stac", arguments=arguments)
         try:
             metadata = metadata_from_stac(url)

tests/rest/datacube/test_datacube100.py

@@ -2129,7 +2129,7 @@ def test_load_collection_max_cloud_cover_summaries_warning(
     if expect_warning:
         assert len(recwarn.list) == 1
         assert re.search(
-            "property filtering.*undefined.*collection metadata.*eo:cloud_cover",
+            "Property filtering.*unsupported.*collection.*metadata.*eo:cloud_cover",
             str(recwarn.pop(UserWarning).message),
         )
     else:

tests/rest/test_connection.py

@@ -2788,24 +2788,94 @@ def test_extents(self, con120):
             }
         }
 
-    def test_properties(self, con120):
-        cube = con120.load_stac("https://provider.test/dataset", properties={"platform": lambda p: p == "S2A"})
+    @pytest.mark.parametrize(
+        ["properties", "expected"],
+        [
+            (
+                # dict with callable
+                {"platform": lambda p: p == "S2A"},
+                {
+                    "platform": {
+                        "process_graph": {
+                            "eq1": {
+                                "process_id": "eq",
+                                "arguments": {"x": {"from_parameter": "value"}, "y": "S2A"},
+                                "result": True,
+                            }
+                        }
+                    }
+                },
+            ),
+            (
+                # Single collection_property
+                (openeo.collection_property("platform") == "S2A"),
+                {
+                    "platform": {
+                        "process_graph": {
+                            "eq1": {
+                                "process_id": "eq",
+                                "arguments": {"x": {"from_parameter": "value"}, "y": "S2A"},
+                                "result": True,
+                            }
+                        }
+                    }
+                },
+            ),
+            (
+                # List of collection_properties
+                [
+                    openeo.collection_property("platform") == "S2A",
+                    openeo.collection_property("flavor") == "salty",
+                ],
+                {
+                    "platform": {
+                        "process_graph": {
+                            "eq1": {
+                                "process_id": "eq",
+                                "arguments": {"x": {"from_parameter": "value"}, "y": "S2A"},
+                                "result": True,
+                            }
+                        }
+                    },
+                    "flavor": {
+                        "process_graph": {
+                            "eq2": {
+                                "process_id": "eq",
+                                "arguments": {"x": {"from_parameter": "value"}, "y": "salty"},
+                                "result": True,
+                            }
+                        }
+                    },
+                },
+            ),
+            (
+                # Advanced PGNode usage
+                {"platform": PGNode(process_id="custom_foo", arguments={"mode": "bar"})},
+                {
+                    "platform": {
+                        "process_graph": {
+                            "customfoo1": {
+                                "process_id": "custom_foo",
+                                "arguments": {"mode": "bar"},
+                                "result": True,
+                            }
+                        }
+                    },
+                },
+            ),
+        ],
+    )
+    def test_property_filtering(self, con120, properties, expected):
+        cube = con120.load_stac(
+            "https://provider.test/dataset",
+            properties=properties,
+        )
         assert cube.flat_graph() == {
             "loadstac1": {
                 "process_id": "load_stac",
                 "arguments": {
                     "url": "https://provider.test/dataset",
-                    "properties": {
-                        "platform": {
-                            "process_graph": {
-                                "eq1": {
-                                    "arguments": {"x": {"from_parameter": "value"}, "y": "S2A"},
-                                    "process_id": "eq",
-                                    "result": True,
-                                }
-                            }
-                        }
-                    },
+                    "properties": expected,
                 },
                 "result": True,
             }