Add recursive process graph unrolling

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add `Connection.unroll_process_graph()` to recursively inline user-defined and remote process definitions. ([#749](https://github.com/Open-EO/openeo-python-client/issues/749))
 - `DataCube.resample_spatial()` now supports parameterized `resolution` and `projection` arguments.  ([#897](https://github.com/Open-EO/openeo-python-client/issues/897))
 - Sanitize asset download filenames (e.g. strip slashes, (semi)colon, hash, ...), instead of blindly using the asset key as filename. ([#820](https://github.com/Open-EO/openeo-python-client/issues/820))
 - Support parameters in `DataCube` apply- and band-math operations ([#903](https://github.com/Open-EO/openeo-python-client/issues/903))

openeo/internal/graph_unrolling.py

@@ -0,0 +1,220 @@
+from __future__ import annotations
+
+import copy
+from typing import Callable, Dict, Iterable, Mapping, Optional, Tuple
+
+
+class ProcessGraphUnrollError(ValueError):
+    """Raised when a process graph can not be unrolled safely."""
+
+
+ProcessDefinitionResolver = Callable[[Mapping], Optional[Mapping]]
+
+
+class ProcessGraphUnroller:
+    """Inline user-defined process definitions into a flat process graph."""
+
+    def __init__(self, *, resolve_process_definition: ProcessDefinitionResolver):
+        self._resolve_process_definition = resolve_process_definition
+
+    def unroll(self, process_graph: Mapping[str, Mapping]) -> Dict[str, dict]:
+        return self._unroll_graph(copy.deepcopy(dict(process_graph)), stack=())
+
+    def _unroll_graph(
+        self, process_graph: Dict[str, dict], *, stack: Tuple[Tuple[str, Optional[str]], ...]
+    ) -> Dict[str, dict]:
+        process_graph = self._unroll_callback_graphs(process_graph=process_graph, stack=stack)
+
+        while True:
+            expandable = next(
+                (
+                    (node_id, node, definition)
+                    for node_id, node in process_graph.items()
+                    if (definition := self._resolve_process_definition(node)) is not None
+                ),
+                None,
+            )
+            if expandable is None:
+                return process_graph
+
+            node_id, node, definition = expandable
+            process_key = (str(node.get("process_id")), node.get("namespace"))
+            if process_key in stack:
+                chain = " -> ".join(pid for pid, _ in (*stack, process_key))
+                raise ProcessGraphUnrollError(f"Recursive process definition detected: {chain}")
+
+            inlined = self._instantiate_definition(
+                definition=definition,
+                arguments=node.get("arguments", {}),
+                invocation_id=node_id,
+                used_node_ids=process_graph,
+            )
+            inlined = self._unroll_graph(inlined, stack=(*stack, process_key))
+            result_id = self._get_result_node_id(inlined)
+
+            if node.get("result"):
+                inlined[result_id]["result"] = True
+            else:
+                inlined[result_id].pop("result", None)
+
+            process_graph = self._replace_node(
+                process_graph=process_graph,
+                node_id=node_id,
+                replacement=inlined,
+                replacement_result_id=result_id,
+            )
+            process_graph = self._unroll_callback_graphs(process_graph=process_graph, stack=stack)
+
+    def _instantiate_definition(
+        self,
+        *,
+        definition: Mapping,
+        arguments: Mapping,
+        invocation_id: str,
+        used_node_ids: Iterable[str],
+    ) -> Dict[str, dict]:
+        process_graph = definition.get("process_graph")
+        if not isinstance(process_graph, Mapping) or not process_graph:
+            raise ProcessGraphUnrollError(
+                f"Process definition {definition.get('id', invocation_id)!r} has no process graph"
+            )
+
+        bindings = self._build_parameter_bindings(definition=definition, arguments=arguments)
+        node_id_map = self._build_node_id_map(
+            inner_node_ids=process_graph,
+            invocation_id=invocation_id,
+            used_node_ids=used_node_ids,
+        )
+
+        instantiated = {}
+        for inner_id, inner_node in copy.deepcopy(dict(process_graph)).items():
+            inner_node = self._rewrite_references(inner_node, node_id_map)
+            inner_node = self._bind_parameters(inner_node, bindings)
+            instantiated[node_id_map[inner_id]] = inner_node
+        return instantiated
+
+    @staticmethod
+    def _build_parameter_bindings(*, definition: Mapping, arguments: Mapping) -> Dict[str, object]:
+        parameters = definition.get("parameters", [])
+        if not isinstance(parameters, list):
+            raise ProcessGraphUnrollError(f"Invalid parameters in process definition {definition.get('id')!r}")
+
+        bindings = {}
+        missing = []
+        for parameter in parameters:
+            name = parameter.get("name")
+            if not isinstance(name, str):
+                raise ProcessGraphUnrollError(f"Invalid parameter in process definition {definition.get('id')!r}")
+            if name in arguments:
+                bindings[name] = copy.deepcopy(arguments[name])
+            elif "default" in parameter:
+                bindings[name] = copy.deepcopy(parameter["default"])
+            elif parameter.get("optional", False):
+                bindings[name] = None
+            else:
+                missing.append(name)
+
+        if missing:
+            raise ProcessGraphUnrollError(
+                f"Missing required arguments {sorted(missing)} for process {definition.get('id')!r}"
+            )
+        return bindings
+
+    @staticmethod
+    def _build_node_id_map(
+        *, inner_node_ids: Iterable[str], invocation_id: str, used_node_ids: Iterable[str]
+    ) -> Dict[str, str]:
+        used = set(used_node_ids)
+        result = {}
+        for inner_id in inner_node_ids:
+            base = f"{invocation_id}_{inner_id}"
+            candidate = base
+            index = 2
+            while candidate in used:
+                candidate = f"{base}_{index}"
+                index += 1
+            result[inner_id] = candidate
+            used.add(candidate)
+        return result
+
+    def _unroll_callback_graphs(
+        self, *, process_graph: Dict[str, dict], stack: Tuple[Tuple[str, Optional[str]], ...]
+    ) -> Dict[str, dict]:
+        def visit(value):
+            if isinstance(value, dict):
+                return {
+                    key: (
+                        self._unroll_graph(copy.deepcopy(child), stack=stack)
+                        if key == "process_graph" and isinstance(child, dict)
+                        else visit(child)
+                    )
+                    for key, child in value.items()
+                }
+            if isinstance(value, list):
+                return [visit(child) for child in value]
+            return value
+
+        return {node_id: visit(node) for node_id, node in process_graph.items()}
+
+    @staticmethod
+    def _rewrite_references(value, node_id_map: Mapping[str, str]):
+        if isinstance(value, dict):
+            if set(value) == {"from_node"} and value["from_node"] in node_id_map:
+                return {"from_node": node_id_map[value["from_node"]]}
+            return {
+                key: child if key == "process_graph" else ProcessGraphUnroller._rewrite_references(child, node_id_map)
+                for key, child in value.items()
+            }
+        if isinstance(value, list):
+            return [ProcessGraphUnroller._rewrite_references(child, node_id_map) for child in value]
+        return value
+
+    @staticmethod
+    def _bind_parameters(value, bindings: Mapping[str, object]):
+        if isinstance(value, dict):
+            if set(value) == {"from_parameter"}:
+                parameter = value["from_parameter"]
+                if parameter not in bindings:
+                    raise ProcessGraphUnrollError(f"Unknown process parameter {parameter!r}")
+                return copy.deepcopy(bindings[parameter])
+            return {
+                key: child if key == "process_graph" else ProcessGraphUnroller._bind_parameters(child, bindings)
+                for key, child in value.items()
+            }
+        if isinstance(value, list):
+            return [ProcessGraphUnroller._bind_parameters(child, bindings) for child in value]
+        return value
+
+    @staticmethod
+    def _get_result_node_id(process_graph: Mapping[str, Mapping]) -> str:
+        result_nodes = [node_id for node_id, node in process_graph.items() if node.get("result") is True]
+        if len(result_nodes) != 1:
+            raise ProcessGraphUnrollError(f"Expected one result node, found {len(result_nodes)}")
+        return result_nodes[0]
+
+    @staticmethod
+    def _replace_node(
+        *,
+        process_graph: Dict[str, dict],
+        node_id: str,
+        replacement: Dict[str, dict],
+        replacement_result_id: str,
+    ) -> Dict[str, dict]:
+        def replace_reference(value):
+            if isinstance(value, dict):
+                if value == {"from_node": node_id}:
+                    return {"from_node": replacement_result_id}
+                return {
+                    key: child if key == "process_graph" else replace_reference(child) for key, child in value.items()
+                }
+            if isinstance(value, list):
+                return [replace_reference(child) for child in value]
+            return value
+
+        result = {}
+        for current_id, current_node in process_graph.items():
+            if current_id == node_id:
+                result.update(replacement)
+            else:
+                result[current_id] = replace_reference(current_node)
+        return result

openeo/rest/connection.py

@@ -41,6 +41,10 @@
     _FromNodeMixin,
     as_flat_graph,
 )
+from openeo.internal.graph_unrolling import (
+    ProcessGraphUnroller,
+    ProcessGraphUnrollError,
+)
 from openeo.internal.jupyter import VisualDict, VisualList
 from openeo.internal.processes.builder import ProcessBuilderBase
 from openeo.internal.warnings import deprecated, legacy_alias
@@ -1024,6 +1028,81 @@ def describe_process(self, id: str, namespace: Optional[str] = None) -> dict:
 
         raise OpenEoClientException("Process does not exist.")
 
+    def unroll_process_graph(
+        self,
+        process_graph: Union[dict, FlatGraphableMixin, str, Path, List[FlatGraphableMixin]],
+    ) -> Dict[str, dict]:
+        """
+        Inline user-defined and remote process definitions in a process graph.
+
+        The returned flat process graph only references processes advertised by the
+        backend. Registered user-defined processes are loaded from ``/process_graphs``;
+        process nodes with an HTTP(S) ``namespace`` are loaded as remote process
+        definitions. Nested definitions are expanded recursively and parameter
+        defaults are applied.
+
+        :param process_graph: process graph or graph-building object to unroll.
+        :return: a new flat process graph. The input is not modified.
+
+        .. versionadded:: 0.51.0
+        """
+        graph = as_flat_graph(process_graph)
+        if isinstance(graph.get("process"), dict) and isinstance(graph["process"].get("process_graph"), dict):
+            graph = graph["process"]["process_graph"]
+        elif isinstance(graph.get("process_graph"), dict):
+            graph = graph["process_graph"]
+
+        definition_cache: Dict[Tuple[str, Optional[str]], Optional[dict]] = {}
+        predefined_process_ids: Optional[Set[str]] = None
+
+        def extract_remote_definition(data: dict, *, process_id: str, namespace: str) -> dict:
+            if "process_graph" in data:
+                definition = data
+            elif isinstance(data.get("processes"), list):
+                matches = [process for process in data["processes"] if process.get("id") == process_id]
+                if len(matches) != 1:
+                    raise ProcessGraphUnrollError(
+                        f"Expected one definition for process {process_id!r} at {namespace!r}, found {len(matches)}"
+                    )
+                definition = matches[0]
+            else:
+                raise ProcessGraphUnrollError(f"No process definition found at {namespace!r}")
+
+            if definition.get("id") not in {None, process_id}:
+                raise ProcessGraphUnrollError(
+                    f"Expected process {process_id!r} at {namespace!r}, found {definition.get('id')!r}"
+                )
+            return definition
+
+        def resolve(node: Mapping) -> Optional[dict]:
+            nonlocal predefined_process_ids
+
+            process_id = node.get("process_id")
+            namespace = node.get("namespace")
+            if not isinstance(process_id, str):
+                return None
+            key = (process_id, namespace)
+            if key in definition_cache:
+                return definition_cache[key]
+
+            if isinstance(namespace, str) and urllib.parse.urlparse(namespace).scheme in {"http", "https"}:
+                response = requests.get(namespace)
+                response.raise_for_status()
+                definition = extract_remote_definition(response.json(), process_id=process_id, namespace=namespace)
+            elif namespace is not None:
+                definition = None
+            else:
+                if predefined_process_ids is None:
+                    predefined_process_ids = {process["id"] for process in self.list_processes()}
+                definition = (
+                    None if process_id in predefined_process_ids else self.user_defined_process(process_id).describe()
+                )
+
+            definition_cache[key] = definition
+            return definition
+
+        return ProcessGraphUnroller(resolve_process_definition=resolve).unroll(graph)
+
     def list_jobs(self, limit: Union[int, None] = 100) -> JobListingResponse:
         """
         Lists (batch) jobs metadata of the authenticated user.