Fix node with image.

server/executors/one_by_one.py

@@ -7,142 +7,166 @@ import traceback
 import inspect
 import typing
 
+
 class Context(ops.BaseConfig):
-  '''Passed to operation functions as "_ctx" if they have such a parameter.'''
-  node: workspace.WorkspaceNode
-  last_result: typing.Any = None
+    """Passed to operation functions as "_ctx" if they have such a parameter."""
+
+    node: workspace.WorkspaceNode
+    last_result: typing.Any = None
+
 
 class Output(ops.BaseConfig):
-  '''Return this to send values to specific outputs of a node.'''
-  output_handle: str
-  value: dict
+    """Return this to send values to specific outputs of a node."""
+
+    output_handle: str
+    value: dict
 
 
 def df_to_list(df):
-  return df.to_dict(orient='records')
+    return df.to_dict(orient="records")
+
 
 def has_ctx(op):
-  sig = inspect.signature(op.func)
-  return '_ctx' in sig.parameters
+    sig = inspect.signature(op.func)
+    return "_ctx" in sig.parameters
+
 
 CACHES = {}
 
+
 def register(env: str, cache: bool = True):
-  '''Registers the one-by-one executor.'''
-  if cache:
-    CACHES[env] = {}
-    cache = CACHES[env]
-  else:
-    cache = None
-  ops.EXECUTORS[env] = lambda ws: execute(ws, ops.CATALOGS[env], cache=cache)
+    """Registers the one-by-one executor."""
+    if cache:
+        CACHES[env] = {}
+        cache = CACHES[env]
+    else:
+        cache = None
+    ops.EXECUTORS[env] = lambda ws: execute(ws, ops.CATALOGS[env], cache=cache)
+
 
 def get_stages(ws, catalog):
-  '''Inputs on top/bottom are batch inputs. We decompose the graph into a DAG of components along these edges.'''
-  nodes = {n.id: n for n in ws.nodes}
-  batch_inputs = {}
-  inputs = {}
-  for edge in ws.edges:
-    inputs.setdefault(edge.target, []).append(edge.source)
-    node = nodes[edge.target]
-    op = catalog[node.data.title]
-    i = op.inputs[edge.targetHandle]
-    if i.position in 'top or bottom':
-      batch_inputs.setdefault(edge.target, []).append(edge.source)
-  stages = []
-  for bt, bss in batch_inputs.items():
-    upstream = set(bss)
-    new = set(bss)
-    while new:
-      n = new.pop()
-      for i in inputs.get(n, []):
-        if i not in upstream:
-          upstream.add(i)
-          new.add(i)
-    stages.append(upstream)
-  stages.sort(key=lambda s: len(s))
-  stages.append(set(nodes))
-  return stages
+    """Inputs on top/bottom are batch inputs. We decompose the graph into a DAG of components along these edges."""
+    nodes = {n.id: n for n in ws.nodes}
+    batch_inputs = {}
+    inputs = {}
+    for edge in ws.edges:
+        inputs.setdefault(edge.target, []).append(edge.source)
+        node = nodes[edge.target]
+        op = catalog[node.data.title]
+        i = op.inputs[edge.targetHandle]
+        if i.position in "top or bottom":
+            batch_inputs.setdefault(edge.target, []).append(edge.source)
+    stages = []
+    for bt, bss in batch_inputs.items():
+        upstream = set(bss)
+        new = set(bss)
+        while new:
+            n = new.pop()
+            for i in inputs.get(n, []):
+                if i not in upstream:
+                    upstream.add(i)
+                    new.add(i)
+        stages.append(upstream)
+    stages.sort(key=lambda s: len(s))
+    stages.append(set(nodes))
+    return stages
 
 
 def _default_serializer(obj):
-  if isinstance(obj, pydantic.BaseModel):
-    return obj.dict()
-  return {"__nonserializable__": id(obj)}
+    if isinstance(obj, pydantic.BaseModel):
+        return obj.dict()
+    return {"__nonserializable__": id(obj)}
+
 
 def make_cache_key(obj):
-  return orjson.dumps(obj, default=_default_serializer)
+    return orjson.dumps(obj, default=_default_serializer)
+
 
 EXECUTOR_OUTPUT_CACHE = {}
 
+
 async def await_if_needed(obj):
-  if inspect.isawaitable(obj):
-    return await obj
-  return obj
+    if inspect.isawaitable(obj):
+        return await obj
+    return obj
+
 
 async def execute(ws, catalog, cache=None):
-  nodes = {n.id: n for n in ws.nodes}
-  contexts = {n.id: Context(node=n) for n in ws.nodes}
-  edges = {n.id: [] for n in ws.nodes}
-  for e in ws.edges:
-    edges[e.source].append(e)
-  tasks = {}
-  NO_INPUT = object() # Marker for initial tasks.
-  for node in ws.nodes:
-    node.data.error = None
-    op = catalog[node.data.title]
-    # Start tasks for nodes that have no non-batch inputs.
-    if all([i.position in 'top or bottom' for i in op.inputs.values()]):
-      tasks[node.id] = [NO_INPUT]
-  batch_inputs = {}
-  # Run the rest until we run out of tasks.
-  stages = get_stages(ws, catalog)
-  for stage in stages:
-    next_stage = {}
-    while tasks:
-      n, ts = tasks.popitem()
-      if n not in stage:
-        next_stage.setdefault(n, []).extend(ts)
-        continue
-      node = nodes[n]
-      data = node.data
-      op = catalog[data.title]
-      params = {**data.params}
-      if has_ctx(op):
-        params['_ctx'] = contexts[node.id]
-      results = []
-      for task in ts:
-        try:
-          inputs = [
-            batch_inputs[(n, i.name)] if i.position in 'top or bottom' else task
-            for i in op.inputs.values()]
-          if cache is not None:
-            key = make_cache_key((inputs, params))
-            if key not in cache:
-              cache[key] = await await_if_needed(op(*inputs, **params))
-            result = cache[key]
-          else:
-            result = await await_if_needed(op(*inputs, **params))
-        except Exception as e:
-          traceback.print_exc()
-          data.error = str(e)
-          break
-        contexts[node.id].last_result = result
-        # Returned lists and DataFrames are considered multiple tasks.
-        if isinstance(result, pd.DataFrame):
-          result = df_to_list(result)
-        elif not isinstance(result, list):
-          result = [result]
-        results.extend(result)
-      else: # Finished all tasks without errors.
-        if op.type == 'visualization' or op.type == 'table_view' or op.type == 'image':
-          data.display = results[0]
-        for edge in edges[node.id]:
-          t = nodes[edge.target]
-          op = catalog[t.data.title]
-          i = op.inputs[edge.targetHandle]
-          if i.position in 'top or bottom':
-            batch_inputs.setdefault((edge.target, edge.targetHandle), []).extend(results)
-          else:
-            tasks.setdefault(edge.target, []).extend(results)
-    tasks = next_stage
-  return contexts
+    nodes = {n.id: n for n in ws.nodes}
+    contexts = {n.id: Context(node=n) for n in ws.nodes}
+    edges = {n.id: [] for n in ws.nodes}
+    for e in ws.edges:
+        edges[e.source].append(e)
+    tasks = {}
+    NO_INPUT = object()  # Marker for initial tasks.
+    for node in ws.nodes:
+        node.data.error = None
+        op = catalog.get(node.data.title)
+        if op is None:
+            node.data.error = f'Operation "{node.data.title}" not found.'
+            continue
+        # Start tasks for nodes that have no non-batch inputs.
+        if all([i.position in "top or bottom" for i in op.inputs.values()]):
+            tasks[node.id] = [NO_INPUT]
+    batch_inputs = {}
+    # Run the rest until we run out of tasks.
+    stages = get_stages(ws, catalog)
+    for stage in stages:
+        next_stage = {}
+        while tasks:
+            n, ts = tasks.popitem()
+            if n not in stage:
+                next_stage.setdefault(n, []).extend(ts)
+                continue
+            node = nodes[n]
+            data = node.data
+            op = catalog[data.title]
+            params = {**data.params}
+            if has_ctx(op):
+                params["_ctx"] = contexts[node.id]
+            results = []
+            for task in ts:
+                try:
+                    inputs = [
+                        batch_inputs[(n, i.name)]
+                        if i.position in "top or bottom"
+                        else task
+                        for i in op.inputs.values()
+                    ]
+                    if cache is not None:
+                        key = make_cache_key((inputs, params))
+                        if key not in cache:
+                            cache[key] = await await_if_needed(op(*inputs, **params))
+                        result = cache[key]
+                    else:
+                        result = await await_if_needed(op(*inputs, **params))
+                except Exception as e:
+                    traceback.print_exc()
+                    data.error = str(e)
+                    break
+                contexts[node.id].last_result = result
+                # Returned lists and DataFrames are considered multiple tasks.
+                if isinstance(result, pd.DataFrame):
+                    result = df_to_list(result)
+                elif not isinstance(result, list):
+                    result = [result]
+                results.extend(result)
+            else:  # Finished all tasks without errors.
+                if (
+                    op.type == "visualization"
+                    or op.type == "table_view"
+                    or op.type == "image"
+                ):
+                    data.display = results[0]
+                for edge in edges[node.id]:
+                    t = nodes[edge.target]
+                    op = catalog[t.data.title]
+                    i = op.inputs[edge.targetHandle]
+                    if i.position in "top or bottom":
+                        batch_inputs.setdefault(
+                            (edge.target, edge.targetHandle), []
+                        ).extend(results)
+                    else:
+                        tasks.setdefault(edge.target, []).extend(results)
+        tasks = next_stage
+    return contexts

web/src/workspace/Workspace.tsx

@@ -15,7 +15,6 @@ import {
   type Node,
   type Edge,
   type Connection,
-  type NodeTypes,
   useReactFlow,
   MiniMap,
 } from '@xyflow/react';
@@ -28,17 +27,14 @@ import Atom from '~icons/tabler/atom.jsx';
 import { syncedStore, getYjsDoc } from "@syncedstore/core";
 import { WebsocketProvider } from "y-websocket";
 import NodeWithParams from './nodes/NodeWithParams';
-// import NodeWithVisualization from './NodeWithVisualization';
-// import NodeWithImage from './NodeWithImage';
 // import NodeWithTableView from './NodeWithTableView';
-// import NodeWithSubFlow from './NodeWithSubFlow';
-// import NodeWithArea from './NodeWithArea';
-// import NodeSearch from './NodeSearch';
 import EnvironmentSelector from './EnvironmentSelector';
 import { LynxKiteState } from './LynxKiteState';
 import '@xyflow/react/dist/style.css';
 import { Workspace, WorkspaceNode } from "../apiTypes.ts";
 import NodeSearch, { OpsOp, Catalog, Catalogs } from "./NodeSearch.tsx";
+import NodeWithVisualization from "./nodes/NodeWithVisualization.tsx";
+import NodeWithImage from "./nodes/NodeWithImage.tsx";
 
 export default function (props: any) {
   return (
@@ -144,6 +140,8 @@ function LynxKiteFlow() {
   const nodeTypes = useMemo(() => ({
     basic: NodeWithParams,
     table_view: NodeWithParams,
+    visualization: NodeWithVisualization,
+    image: NodeWithImage,
   }), []);
   const closeNodeSearch = useCallback(() => {
     setNodeSearchSettings(undefined);
@@ -160,7 +158,7 @@ function LynxKiteFlow() {
       pos: { x: event.clientX, y: event.clientY },
       boxes: catalog.data![state.workspace.env!],
     });
-  }, [setNodeSearchSettings, suppressSearchUntil]);
+  }, [catalog, state, setNodeSearchSettings, suppressSearchUntil]);
   const addNode = useCallback((meta: OpsOp) => {
     const node: Partial<WorkspaceNode> = {
       type: meta.type,
@@ -184,7 +182,7 @@ function LynxKiteFlow() {
     wnodes.push(node as WorkspaceNode);
     setNodes([...nodes, node as WorkspaceNode]);
     closeNodeSearch();
-  }, [state, reactFlow, setNodes]);
+  }, [nodeSearchSettings, state, reactFlow, setNodes]);
 
   const onConnect = useCallback((connection: Connection) => {
     setSuppressSearchUntil(Date.now() + 200);

web/src/workspace/nodes/NodeWithImage.tsx

@@ -0,0 +1,11 @@
+import NodeWithParams from './NodeWithParams';
+
+const NodeWithImage = (props: any) => {
+  return (
+    <NodeWithParams {...props}>
+      {props.data.display && <img src={props.data.display} alt="Node Display" />}
+    </NodeWithParams>
+  );
+};
+
+export default NodeWithImage;

web/src/workspace/nodes/NodeWithParams.tsx

@@ -23,6 +23,7 @@ function NodeWithParams(props: any) {
           onChange={(value: any, opts?: UpdateOptions) => setParam(name, value, opts || {})}
         />
       )}
+      {props.children}
     </LynxKiteNode >
   );
 }

web/src/workspace/nodes/NodeWithVisualization.tsx

@@ -0,0 +1,28 @@
+import React, { useEffect } from 'react';
+import NodeWithParams from './NodeWithParams';
+import * as echarts from 'echarts';
+
+const NodeWithVisualization = (props: any) => {
+  const chartsRef = React.useRef<HTMLDivElement>(null);
+  const chartsInstanceRef = React.useRef<echarts.ECharts>();
+  useEffect(() => {
+    const opts = props.data?.display?.value;
+    if (!opts || !chartsRef.current) return;
+    console.log(chartsRef.current);
+    chartsInstanceRef.current = echarts.init(chartsRef.current, null, { renderer: 'canvas', width: 250, height: 250 });
+    chartsInstanceRef.current.setOption(opts);
+    const onResize = () => chartsInstanceRef.current?.resize();
+    window.addEventListener('resize', onResize);
+    return () => {
+      window.removeEventListener('resize', onResize);
+      chartsInstanceRef.current?.dispose();
+    };
+  }, [props.data?.display?.value]);
+  return (
+    <NodeWithParams {...props}>
+      <div className="box" draggable={false} ref={chartsRef} />;
+    </NodeWithParams>
+  );
+};
+
+export default NodeWithVisualization;