Initial implementation of CWL support in run_udf.

openeo_driver/ProcessGraphDeserializer.py

@@ -1769,7 +1769,7 @@ def merge_cubes(args: ProcessArgs, env: EvalEnv) -> DriverDataCube:
 def run_udf(args: ProcessArgs, env: EvalEnv):
     # TODO: note: this implements a non-standard usage of `run_udf`: processing "vector" cube (direct JSON or from aggregate_spatial, ...)
     dry_run_tracer: DryRunDataTracer = env.get(ENV_DRY_RUN_TRACER)
-    data = args.get_required(name="data")
+    data = args.get_optional(name="data")
     udf, runtime = _get_udf(args, env=env)
     context = args.get_optional(name="context", default={})
 
@@ -1781,6 +1781,14 @@ def run_udf(args: ProcessArgs, env: EvalEnv):
         # This way a weak_spatial_extent can be calculated from the UDF's output.
         return data.run_udf()
 
+    if runtime.lower() == "EOAP-CWL".lower():
+        return env.backend_implementation.run_cwl(
+            data=data,
+            env=env,
+            cwl=udf,
+            context=context,
+        )
+
     if env.get("validation", False):
         raise FeatureUnsupportedException("run_udf is not supported in validation mode.")
 

openeo_driver/backend.py

@@ -949,6 +949,16 @@ def load_ml_model(self, job_id: str) -> DriverMlModel:
     def vector_to_raster(self, input_vector_cube: DriverVectorCube, target: DriverDataCube) -> DriverDataCube:
         raise NotImplementedError
 
+    def run_cwl(
+        self,
+        *,
+        data,  # Will be None at atm
+        env: EvalEnv,
+        cwl: str,
+        context: dict,
+    ) -> DriverDataCube:
+        raise NotImplementedError
+
     def visit_process_graph(self, process_graph: dict) -> ProcessGraphVisitor:
         """Create a process graph visitor and accept given process graph"""
         return ProcessGraphVisitor().accept_process_graph(process_graph)