Issue #391 Initial implementation of MultiResult

Author: soxofaan

openeo/internal/graph_building.py

@@ -14,7 +14,7 @@
 import sys
 from contextlib import nullcontext
 from pathlib import Path
-from typing import Any, Dict, Iterator, Optional, Tuple, Union
+from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
 
 from openeo.api.process import Parameter
 from openeo.internal.process_graph_visitor import (
@@ -438,3 +438,24 @@ def _process_from_parameter(self, name: str) -> Any:
         if name not in self._parameters:
             raise ProcessGraphVisitException("No substitution value for parameter {p!r}.".format(p=name))
         return self._parameters[name]
+
+
+class MultiResult(FlatGraphableMixin):
+    """
+    Handler of use cases where there are multiple result nodes
+    (or other leaf nodes) in a process graph.
+    """
+
+    def __init__(self, leaves: List[FlatGraphableMixin]):
+        self._leaves = leaves
+
+    def flat_graph(self) -> Dict[str, dict]:
+        result = {}
+        for leaf in self._leaves:
+            leaf_graph = leaf.flat_graph()
+            existing = set(leaf_graph.keys()).intersection(result.keys())
+            if existing:
+                # TODO: automatic renaming of duplicate node ids?
+                raise ValueError(f"Duplicate node ids while building multi-result process graph: {sorted(existing)}")
+            result.update(leaf_graph)
+        return result

tests/internal/test_graphbuilding.py

@@ -1,4 +1,5 @@
 import io
+import re
 import textwrap
 
 import pytest
@@ -7,6 +8,7 @@
 from openeo.api.process import Parameter
 from openeo.internal.graph_building import (
     FlatGraphNodeIdGenerator,
+    MultiResult,
     PGNode,
     PGNodeGraphUnflattener,
     ReduceNode,
@@ -412,3 +414,19 @@ def test_walk_nodes_nested():
     walk = list(node.walk_nodes())
     assert all(isinstance(n, PGNode) for n in walk)
     assert set(n.process_id for n in walk) == {"load1", "max", "foo", "load2", "add", "five"}
+
+
+class TestMultiResult:
+    def test_simple(self):
+        multi = MultiResult([PGNode("foo"), PGNode("bar")])
+        assert multi.flat_graph() == {
+            "foo1": {"process_id": "foo", "arguments": {}, "result": True},
+            "bar1": {"process_id": "bar", "arguments": {}, "result": True},
+        }
+
+    def test_simple_duplicates(self):
+        multi = MultiResult([PGNode("foo"), PGNode("foo")])
+        with pytest.raises(
+            ValueError, match=re.escape("Duplicate node ids while building multi-result process graph: ['foo1']")
+        ):
+            multi.flat_graph()