support additional from_parameter in property filter

* seems functional, needs work #327

* clean up and add tests #327

* retain backwards compatibility wrt/ openeo-geopyspark-driver #327

* update CHANGELOG #327

* improve error message #327

* update version and CHANGELOG #327

Author: bossie

CHANGELOG.md

@@ -21,9 +21,11 @@ and start a new "In Progress" section above it.
 
 ## In progress
 
+## 0.124.0
+
 - Better argument validation in `resample_spatial`/`resample_cube_spatial` (related to [Open-EO/openeo-python-client#690](https://github.com/Open-EO/openeo-python-client/issues/690))
 - Improve `resample_spatial`/`resample_cube_spatial` metadata tracking in dry-run ([#348](https://github.com/Open-EO/openeo-python-driver/issues/348))
-
+- `load_collection`/`load_stac`: support parameters in `properties` ([#327](https://github.com/Open-EO/openeo-python-driver/issues/327))
 
 ## 0.123.0
 

openeo_driver/ProcessGraphDeserializer.py

@@ -759,15 +759,18 @@ def load_collection(args: dict, env: EvalEnv) -> DriverDataCube:
 
     dry_run_tracer: DryRunDataTracer = env.get(ENV_DRY_RUN_TRACER)
     if dry_run_tracer:
-        return dry_run_tracer.load_collection(collection_id=collection_id, arguments=arguments, metadata=metadata)
+        return dry_run_tracer.load_collection(
+            collection_id=collection_id, arguments=arguments, metadata=metadata, env=env
+        )
     else:
         # Extract basic source constraints.
         # TODO #275: eliminate this VITO specific handling?
         properties = {**CollectionMetadata(metadata).get("_vito", "properties", default={}),
                       **arguments.get("properties", {})}
 
-        source_id = dry_run.DataSource.load_collection(collection_id=collection_id,
-                                                       properties=properties,bands = arguments.get("bands",[])).get_source_id()
+        source_id = dry_run.DataSource.load_collection(
+            collection_id=collection_id, properties=properties, bands=arguments.get("bands", []), env=env
+        ).get_source_id()
         load_params = _extract_load_parameters(env, source_id=source_id)
         # Override with explicit arguments
         load_params.update(arguments)
@@ -2416,9 +2419,11 @@ def load_stac(args: Dict, env: EvalEnv) -> DriverDataCube:
 
     dry_run_tracer: DryRunDataTracer = env.get(ENV_DRY_RUN_TRACER)
     if dry_run_tracer:
-        return dry_run_tracer.load_stac(url, arguments)
+        return dry_run_tracer.load_stac(url, arguments, env)
     else:
-        source_id = dry_run.DataSource.load_stac(url, properties=arguments.get("properties", {}), bands=arguments.get("bands",[])).get_source_id()
+        source_id = dry_run.DataSource.load_stac(
+            url, properties=arguments.get("properties", {}), bands=arguments.get("bands", []), env=env
+        ).get_source_id()
         load_params = _extract_load_parameters(env, source_id=source_id)
         load_params.update(arguments)
 

openeo_driver/_version.py

@@ -1 +1 @@
-__version__ = "0.123.0a1"
+__version__ = "0.124.0a1"

openeo_driver/dry_run.py

@@ -167,10 +167,10 @@ def describe(self) -> str:
         return self._process
 
     @classmethod
-    def load_collection(cls, collection_id, properties={}, bands=[]) -> "DataSource":
+    def load_collection(cls, collection_id, properties={}, bands=[], env=EvalEnv()) -> "DataSource":
         """Factory for a `load_collection` DataSource."""
         exact_property_matches = {
-            property_name: filter_properties.extract_literal_match(condition)
+            property_name: filter_properties.extract_literal_match(condition, env)
             for property_name, condition in properties.items()
         }
 
@@ -192,10 +192,10 @@ def load_result(cls, job_id: str) -> "DataSource":
         return cls(process="load_result", arguments=(job_id,))
 
     @classmethod
-    def load_stac(cls, url: str, properties={}, bands=[]) -> "DataSource":
+    def load_stac(cls, url: str, properties={}, bands=[], env=EvalEnv()) -> "DataSource":
         """Factory for a `load_stac` DataSource."""
         exact_property_matches = {
-            property_name: filter_properties.extract_literal_match(condition)
+            property_name: filter_properties.extract_literal_match(condition, env)
             for property_name, condition in properties.items()
         }
 
@@ -273,7 +273,9 @@ def process_traces(self, traces: List[DataTraceBase], operation: str, arguments:
         """Process given traces with an operation (and keep track of the results)."""
         return [self.add_trace(DataTrace(parent=t, operation=operation, arguments=arguments)) for t in traces]
 
-    def load_collection(self, collection_id: str, arguments: dict, metadata: dict = None) -> "DryRunDataCube":
+    def load_collection(
+        self, collection_id: str, arguments: dict, metadata: dict = None, env: EvalEnv = EvalEnv()
+    ) -> "DryRunDataCube":
         """Create a DryRunDataCube from a `load_collection` process."""
         # TODO #275 avoid VITO/Terrascope specific handling here?
         properties = {
@@ -282,7 +284,7 @@ def load_collection(self, collection_id: str, arguments: dict, metadata: dict =
         }
 
         trace = DataSource.load_collection(
-            collection_id=collection_id, properties=properties, bands=arguments.get("bands", [])
+            collection_id=collection_id, properties=properties, bands=arguments.get("bands", []), env=env
         )
         self.add_trace(trace)
 
@@ -329,10 +331,10 @@ def load_result(self, job_id: str, arguments: dict) -> "DryRunDataCube":
 
         return cube
 
-    def load_stac(self, url: str, arguments: dict) -> "DryRunDataCube":
+    def load_stac(self, url: str, arguments: dict, env: EvalEnv = EvalEnv()) -> "DryRunDataCube":
         properties = arguments.get("properties", {})
 
-        trace = DataSource.load_stac(url=url, properties=properties, bands=arguments.get("bands", []))
+        trace = DataSource.load_stac(url=url, properties=properties, bands=arguments.get("bands", []), env=env)
         self.add_trace(trace)
 
         metadata = CollectionMetadata(

openeo_driver/filter_properties.py

@@ -2,6 +2,7 @@
 
 from openeo.internal.process_graph_visitor import ProcessGraphVisitor
 from openeo_driver.errors import OpenEOApiException
+from openeo_driver.utils import EvalEnv
 
 
 class PropertyConditionException(OpenEOApiException):
@@ -10,11 +11,12 @@ class PropertyConditionException(OpenEOApiException):
     message = "Property condition is invalid."
 
 
-def extract_literal_match(condition: dict, parameter_name="value") -> Dict[str, Any]:
+def extract_literal_match(condition: dict, env=EvalEnv()) -> Dict[str, Any]:
     """
     Turns a condition as defined by the load_collection process into a set of criteria ((operator, value) pairs).
     Conditions are currently limited to processes "eq", "lte" and "gte" so they will be turned into a single criterion.
     """
+    callback_parameter_name = "value"  # as in: {"from_parameter": "value"}
 
     class LiteralMatchExtractingGraphVisitor(ProcessGraphVisitor):
 
@@ -32,11 +34,20 @@ def enterProcess(self, process_id: str, arguments: dict, namespace: Union[str, N
 
             self.result["operator"] = "in" if process_id == "array_contains" else process_id
 
-        def enterArgument(self, argument_id: str, value):
-            self.result["parameter"] = value.get("from_parameter")
+        def enterArgument(self, argument_id: str, value: dict):
+            parameter_name = value.get("from_parameter")
+
+            if parameter_name == callback_parameter_name:
+                self.result["parameter"] = parameter_name
+            else:
+                env_parameters = env.collect_parameters()
+                if parameter_name not in env_parameters:
+                    raise PropertyConditionException(f"Unknown parameter {parameter_name}")
+
+                self.result["value"] = env_parameters[parameter_name]
 
         def constantArgument(self, argument_id: str, value):
-            self.result["constant"] = value
+            self.result["value"] = value
 
             if argument_id == 'x':
                 if self.result["operator"] == "lte":
@@ -45,21 +56,21 @@ def constantArgument(self, argument_id: str, value):
                     self.result["operator"] = "lte"
 
         def constantArrayElement(self, value):
-            if "constant" not in self.result:
-                self.result["constant"] = []
+            if "value" not in self.result:
+                self.result["value"] = []
 
-            self.result["constant"].append(value)
+            self.result["value"].append(value)
 
     visitor = LiteralMatchExtractingGraphVisitor()
     visitor.accept_process_graph(condition['process_graph'])
 
     if "parameter" not in visitor.result:
-        raise PropertyConditionException(f"No parameter {parameter_name!r} found")
-    if visitor.result["parameter"] != parameter_name:
+        raise PropertyConditionException(f"No parameter {callback_parameter_name!r} found")
+    if visitor.result["parameter"] != callback_parameter_name:
         raise PropertyConditionException(
-            f"Expected parameter {parameter_name!r} but got {visitor.result['parameter']!r}"
+            f"Expected parameter {callback_parameter_name!r} but got {visitor.result['parameter']!r}"
         )
-    if "constant" not in visitor.result:
-        raise PropertyConditionException(f"No comparison with constant")
+    if "value" not in visitor.result:
+        raise PropertyConditionException(f"No comparison with constant/value from parameter")
 
-    return {visitor.result["operator"]: visitor.result["constant"]}
+    return {visitor.result["operator"]: visitor.result["value"]}

tests/data/pg/1.0/udp/load_collection_property_from_parameter.json

@@ -0,0 +1,38 @@
+{
+  "id": "load_collection_property_from_parameter",
+  "parameters": [
+    {
+      "name": "orbit_state",
+      "schema": {
+        "type": "string"
+      }
+    }
+  ],
+  "process_graph": {
+    "loadcollection1": {
+      "process_id": "load_collection",
+      "arguments": {
+        "id": "SENTINEL1_GRD",
+        "properties": {
+          "sat:orbit_state": {
+            "process_graph": {
+              "eq1": {
+                "arguments": {
+                  "x": {
+                    "from_parameter": "value"
+                  },
+                  "y": {
+                    "from_parameter": "orbit_state"
+                  }
+                },
+                "process_id": "eq",
+                "result": true
+              }
+            }
+          }
+        }
+      },
+      "result": true
+    }
+  }
+}

tests/data/pg/1.0/udp_load_collection_property_from_parameter.json

@@ -0,0 +1,9 @@
+{
+  "udp1": {
+    "process_id": "load_collection_property_from_parameter",
+    "arguments": {
+      "orbit_state": "ASCENDING"
+    },
+    "result": true
+  }
+}

tests/test_filter_properties.py

@@ -1,6 +1,7 @@
 import pytest
 
 from openeo_driver.filter_properties import extract_literal_match, PropertyConditionException
+from openeo_driver.utils import EvalEnv
 
 
 def test_extract_literal_match_basic():
@@ -47,9 +48,10 @@ def test_extract_literal_match_gte_reverse():
 ])
 def test_extract_literal_match_failures(node):
     node["result"] = True
+    env = EvalEnv().push_parameters({"data": "some parameter"})
     pg = {"process_graph": {"p": node}}
     with pytest.raises(PropertyConditionException):
-        extract_literal_match(pg)
+        extract_literal_match(pg, env)
 
 
 def test_array_contains():
@@ -67,3 +69,22 @@ def test_array_contains():
     }
 
     assert extract_literal_match(pg) == {"in": ["31UES", "31UFS"]}
+
+
+def test_parameter():
+    env = EvalEnv().push_parameters({"orbit_state": "ASCENDING"})
+
+    pg = {
+        "process_graph": {
+            "eq1": {
+                "process_id": "eq",
+                "arguments": {
+                    "x": {"from_parameter": "value"},
+                    "y": {"from_parameter": "orbit_state"}
+                },
+                "result": True
+            }
+        }
+    }
+
+    assert extract_literal_match(pg, env) == {"eq": "ASCENDING"}

tests/test_views_execute.py

@@ -4672,3 +4672,18 @@ def i_spy_with_my_little_eye(args: ProcessArgs, env: EvalEnv):
     api.ensure_auth_header()
     res = api.post(path="/result", json=post_data)
     assert res.assert_status_code(200).json == 123
+
+
+def test_load_collection_property_from_parameter(api, udp_registry):
+    # You can't execute a top-level process graph with parameters; instead, you execute a top-level process graph
+    # that calls a UDP with parameters.
+
+    api.set_auth_bearer_token(TEST_USER_BEARER_TOKEN)
+    udp_spec = api.load_json("udp/load_collection_property_from_parameter.json")
+    udp_registry.save(user_id=TEST_USER, process_id="load_collection_property_from_parameter", spec=udp_spec)
+    pg = api.load_json("udp_load_collection_property_from_parameter.json")
+    api.check_result(pg)
+
+    params = dummy_backend.last_load_collection_call("SENTINEL1_GRD")
+
+    assert "sat:orbit_state" in params.properties