Merge pull request #143 from biggraph/feature/task-solver-box

lynxkite-app/src/lynxkite_app/crdt.py

@@ -273,6 +273,7 @@ async def execute(name: str, ws_crdt: pycrdt.Map, ws_pyd: workspace.Workspace, d
             nc["data"]["status"] = "planned"
             # Nodes get a reference to their CRDT maps, so they can update them as the results come in.
             np._crdt = nc
+    ws_pyd = ws_pyd.normalize()
     await workspace.execute(ws_pyd)
     workspace.save(ws_pyd, path)
     print(f"Finished running {name} in {ws_pyd.env}.")

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-core/src/lynxkite/core/ops.py

@@ -144,6 +144,7 @@ def _param_to_type(name, value, type):
         assert value != "", f"{name} is unset."
         return float(value)
     if isinstance(type, enum.EnumMeta):
+        assert value in type.__members__, f"{value} is not an option for {name}."
         return type[value]
     if isinstance(type, types.UnionType):
         match type.__args__:

lynxkite-core/src/lynxkite/core/workspace.py

@@ -97,6 +97,25 @@ class Workspace(BaseConfig):
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
     _crdt: pycrdt.Map
 
+    def normalize(self):
+        if self.env not in ops.CATALOGS:
+            return self
+        catalog = ops.CATALOGS[self.env]
+        _ops = {n.id: catalog[n.data.title] for n in self.nodes if n.data.title in catalog}
+        valid_targets = set(
+            (n.id, h) for n in self.nodes for h in _ops[n.id].inputs if n.id in _ops
+        )
+        valid_sources = set(
+            (n.id, h) for n in self.nodes for h in _ops[n.id].outputs if n.id in _ops
+        )
+        edges = [
+            edge
+            for edge in self.edges
+            if (edge.source, edge.sourceHandle) in valid_sources
+            and (edge.target, edge.targetHandle) in valid_targets
+        ]
+        return self.model_copy(update={"edges": edges})
+
 
 async def execute(ws: Workspace):
     if ws.env in ops.EXECUTORS:

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")