Clean up ndvi in ProcessGraphDeserializer

related to #346

Author: soxofaan

openeo_driver/ProcessGraphDeserializer.py

@@ -1570,21 +1570,12 @@ def apply_kernel(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
 
 
 @process
-def ndvi(args: dict, env: EvalEnv) -> DriverDataCube:
-    image_collection = extract_arg(args, 'data')
-    if not isinstance(image_collection, DriverDataCube):
-        raise ProcessParameterInvalidException(
-            parameter="data", process="ndvi",
-            reason=f"Invalid data type {type(image_collection)!r} expected raster-cube."
-        )
-    if ComparableVersion("1.0.0").or_higher(env["version"]):
-        red = args.get("red")
-        nir = args.get("nir")
-        target_band = args.get("target_band")
-        return image_collection.ndvi(nir=nir, red=red, target_band=target_band)
-    else:
-        name = args.get("name", "ndvi")
-        return image_collection.ndvi(name=name)
+def ndvi(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
+    cube = args.get_required("data", expected_type=DriverDataCube)
+    nir = args.get_optional("nir", default="nir")
+    red = args.get_optional("red", default="red")
+    target_band = args.get_optional("target_band", default=None)
+    return cube.ndvi(nir=nir, red=red, target_band=target_band)
 
 
 @process

openeo_driver/datacube.py

@@ -199,7 +199,7 @@ def timeseries(self, x, y, srs="EPSG:4326") -> dict:
         # TODO #47: remove this non-standard process
         self._not_implemented()
 
-    def ndvi(self, nir: str = "nir", red: str = "red", target_band: str = None) -> 'DriverDataCube':
+    def ndvi(self, nir: str = "nir", red: str = "red", target_band: Optional[str] = None) -> "DriverDataCube":
         self._not_implemented()
 
     def save_result(self, filename: str, format: str, format_options: dict = None) -> str:

openeo_driver/dry_run.py

@@ -759,8 +759,8 @@ def reduce_dimension(
             "reduce_dimension", arguments={}
         )
 
-    def ndvi(self, nir: str = "nir", red: str = "red", target_band: str = None) -> "DriverDataCube":
-        if target_band == None and self.metadata.has_band_dimension():
+    def ndvi(self, nir: str = "nir", red: str = "red", target_band: Optional[str] = None) -> "DryRunDataCube":
+        if target_band is None and self.metadata.has_band_dimension():
             return self._process_metadata(
                 self.metadata.reduce_dimension(dimension_name=self.metadata.band_dimension.name)
             )

tests/test_views_execute.py

@@ -2779,7 +2779,7 @@ def test_user_defined_process_udp_vs_pdp_priority(api, udp_registry):
     api.check_result("udp_ndvi.json")
     dummy = dummy_backend.get_collection("S2_FOOBAR")
     assert dummy.ndvi.call_count == 1
-    dummy.ndvi.assert_called_with(nir=None, red=None, target_band=None)
+    dummy.ndvi.assert_called_with(nir="nir", red="red", target_band=None)
     assert dummy.reduce_dimension.call_count == 0
 
     # Overload ndvi with UDP.