Add "simple" executor.

lynxkite-core/src/lynxkite/core/executors/simple.py

@@ -0,0 +1,64 @@
+"""A LynxKite executor that simply passes the output of one box to the other."""
+
+import os
+from .. import ops
+from .. import workspace
+import traceback
+import inspect
+import graphlib
+
+
+def register(env: str):
+    """Registers the one-by-one executor."""
+    ops.EXECUTORS[env] = lambda ws: execute(ws, ops.CATALOGS[env])
+
+
+async def await_if_needed(obj):
+    if inspect.isawaitable(obj):
+        return await obj
+    return obj
+
+
+async def execute(ws: workspace.Workspace, catalog: ops.Catalog):
+    nodes = {n.id: n for n in ws.nodes}
+    dependencies = {n: [] for n in nodes}
+    in_edges = {n: {} for n in nodes}
+    for e in ws.edges:
+        dependencies[e.target].append(e.source)
+        assert e.targetHandle not in in_edges[e.target], f"Duplicate input for {e.target}"
+        in_edges[e.target][e.targetHandle] = e.source, e.sourceHandle
+    outputs = {}
+    ts = graphlib.TopologicalSorter(dependencies)
+    for node_id in ts.static_order():
+        node = nodes[node_id]
+        op = catalog[node.data.title]
+        params = {**node.data.params}
+        node.publish_started()
+        try:
+            inputs = []
+            missing = []
+            for i in op.inputs.values():
+                edges = in_edges[node_id]
+                if i.name in edges and edges[i.name] in outputs:
+                    inputs.append(outputs[edges[i.name]])
+                else:
+                    missing.append(i.name)
+            if missing:
+                node.publish_error(f"Missing input: {', '.join(missing)}")
+                continue
+            result = op(*inputs, **params)
+            result.output = await await_if_needed(result.output)
+            result.display = await await_if_needed(result.display)
+            if len(op.outputs) == 1:
+                [output] = list(op.outputs.values())
+                outputs[node_id, output.name] = result.output
+            elif len(op.outputs) > 1:
+                assert type(result.output) is dict, "An op with multiple outputs must return a dict"
+                for output in op.outputs.values():
+                    outputs[node_id, output.name] = result.output[output.name]
+            node.publish_result(result)
+        except Exception as e:
+            if not os.environ.get("LYNXKITE_SUPPRESS_OP_ERRORS"):
+                traceback.print_exc()
+            node.publish_error(e)
+    return outputs

lynxkite-pillow-example/src/lynxkite_pillow_example/__init__.py

@@ -1,7 +1,7 @@
 """Demo for how easily we can provide a UI for popular open-source tools."""
 
 from lynxkite.core import ops
-from lynxkite.core.executors import one_by_one
+from lynxkite.core.executors import simple
 from PIL import Image, ImageFilter
 import base64
 import fsspec
@@ -9,7 +9,7 @@ import io
 
 ENV = "Pillow"
 op = ops.op_registration(ENV)
-one_by_one.register(ENV, cache=False)
+simple.register(ENV)
 
 
 @op("Open image")