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={})
 
@@ -1812,6 +1812,12 @@ def run_udf(args: ProcessArgs, env: EvalEnv):
             structured_data_list=[openeo.udf.StructuredData(description="Data list", data=data, type="list")],
             user_context=context
         )
+    elif runtime.lower() == "CWL-Calrissian".lower():
+        return env.backend_implementation.run_cwl(
+            env,
+            udf,
+            context,
+        )
     else:
         raise ProcessParameterInvalidException(
             parameter="data",

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,
+        env: EvalEnv,
+        cwl_url: str,
+        context: dict,
+        stac_root: Optional[str] = None,
+        direct_s3_mode: Optional[bool] = False,
+    ) -> 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)