Some backend execution. Errors.

.gitignore

@@ -0,0 +1,11 @@
+# Editor directories and files
+.vscode/*
+!.vscode/extensions.json
+.idea
+.DS_Store
+*.suo
+*.ntvs*
+*.njsproj
+*.sln
+*.sw?
+__pycache__

main.py

@@ -1,68 +0,0 @@
-from typing import Union
-import fastapi
-import pydantic
-import networkx as nx
-
-class Position(pydantic.BaseModel):
-    x: float
-    y: float
-
-class WorkspaceNodeData(pydantic.BaseModel):
-    title: str
-    params: dict
-
-class WorkspaceNode(pydantic.BaseModel):
-    id: str
-    type: str
-    data: WorkspaceNodeData
-    position: Position
-
-class WorkspaceEdge(pydantic.BaseModel):
-    id: str
-    source: str
-    target: str
-
-class Workspace(pydantic.BaseModel):
-    nodes: list[WorkspaceNode]
-    edges: list[WorkspaceEdge]
-
-
-app = fastapi.FastAPI()
-
-
-@app.get("/")
-def read_root():
-    return {"Hello": "World"}
-
-
-@app.get("/items/{item_id}")
-def read_item(item_id: int, q: Union[str, None] = None):
-    return {"item_id": item_id, "q": q}
-
-
-@app.post("/api/save")
-def save(ws: Workspace):
-    print(ws)
-    G = nx.scale_free_graph(4)
-    return {'graph':{
-    'attributes': {
-      'name': 'My Graph'
-    },
-    'options': {
-      'allowSelfLoops': True,
-      'multi': False,
-      'type': 'mixed'
-    },
-    'nodes': [
-      {'key': 'Thomas'},
-      {'key': 'Eric'}
-    ],
-    'edges': [
-      {
-        'key': 'T->E',
-        'source': 'Thomas',
-        'target': 'Eric',
-      }
-    ]
-  }}
-    return {"graph": list(nx.to_edgelist(G))}

run.sh

@@ -0,0 +1,2 @@
+#!/bin/bash -xue
+uvicorn server.main:app --reload

server/basic_ops.py

@@ -0,0 +1,42 @@
+'''Some operations. To be split into separate files when we have more.'''
+from . import ops
+import pandas as pd
+import networkx as nx
+
+@ops.op("Import Parquet")
+def import_parquet(*, filename: str):
+  '''Imports a parquet file.'''
+  return pd.read_parquet(filename)
+
+@ops.op("Create scale-free graph")
+def create_scale_free_graph(*, nodes: int):
+  '''Creates a scale-free graph with the given number of nodes.'''
+  return nx.scale_free_graph(nodes)
+
+@ops.op("Compute PageRank")
+def compute_pagerank(graph, *, damping: 0.85, iterations: 3):
+  return nx.pagerank(graph)
+
+@ops.op("Visualize graph")
+def visualize_graph(graph) -> 'graphviz':
+  return {'graph':{
+    'attributes': {
+      'name': 'My Graph'
+    },
+    'options': {
+      'allowSelfLoops': True,
+      'multi': False,
+      'type': 'mixed'
+    },
+    'nodes': [
+      {'key': 'Thomas'},
+      {'key': 'Eric'}
+    ],
+    'edges': [
+      {
+        'key': 'T->E',
+        'source': 'Thomas',
+        'target': 'Eric',
+      }
+    ]
+  }}

server/main.py

@@ -0,0 +1,81 @@
+from typing import Optional
+import fastapi
+import pydantic
+from . import ops
+from . import basic_ops
+
+class BaseConfig(pydantic.BaseModel):
+    model_config = pydantic.ConfigDict(
+        extra='allow',
+    )
+
+class Position(BaseConfig):
+    x: float
+    y: float
+
+class WorkspaceNodeData(BaseConfig):
+    title: str
+    params: dict
+    display: Optional[object] = None
+    error: Optional[str] = None
+
+class WorkspaceNode(BaseConfig):
+    id: str
+    type: str
+    data: WorkspaceNodeData
+    position: Position
+
+class WorkspaceEdge(BaseConfig):
+    id: str
+    source: str
+    target: str
+
+class Workspace(BaseConfig):
+    nodes: list[WorkspaceNode]
+    edges: list[WorkspaceEdge]
+
+
+app = fastapi.FastAPI()
+
+
+@app.get("/api/catalog")
+def get_catalog():
+    return [
+        {
+          'type': op.type,
+          'data': { 'title': op.name, 'params': op.params },
+          'targetPosition': 'left' if op.inputs else None,
+          'sourcePosition': 'right' if op.outputs else None,
+        }
+        for op in ops.ALL_OPS.values()]
+
+def execute(ws):
+    nodes = ws.nodes
+    outputs = {}
+    failed = 0
+    while len(outputs) + failed < len(nodes):
+        for node in nodes:
+            if node.id in outputs:
+                continue
+            inputs = [edge.source for edge in ws.edges if edge.target == node.id]
+            if all(input in outputs for input in inputs):
+                inputs = [outputs[input] for input in inputs]
+                data = node.data
+                op = ops.ALL_OPS[data.title]
+                try:
+                  output = op(*inputs, **data.params)
+                except Exception as e:
+                  data.error = str(e)
+                  failed += 1
+                  continue
+                outputs[node.id] = output
+                if op.type == 'graphviz':
+                    data.graph = output
+
+
+@app.post("/api/save")
+def save(ws: Workspace):
+    print(ws)
+    execute(ws)
+    print('exec done', ws)
+    return ws

server/ops.py

@@ -0,0 +1,53 @@
+'''API for implementing LynxKite operations.'''
+import dataclasses
+import inspect
+
+ALL_OPS = {}
+
+@dataclasses.dataclass
+class Op:
+  func: callable
+  name: str
+  params: dict
+  inputs: dict
+  outputs: dict
+  type: str
+
+  def __call__(self, *inputs, **params):
+    # Do conversions here.
+    res = self.func(*inputs, **params)
+    return res
+
+@dataclasses.dataclass
+class EdgeDefinition:
+  df: str
+  source_column: str
+  target_column: str
+  source_table: str
+  target_table: str
+  source_key: str
+  target_key: str
+
+@dataclasses.dataclass
+class Bundle:
+  dfs: dict
+  edges: list[EdgeDefinition]
+
+def op(name):
+  '''Decorator for defining an operation.'''
+  def decorator(func):
+    type = func.__annotations__.get('return') or 'basic'
+    sig = inspect.signature(func)
+    # Positional arguments are inputs.
+    inputs = {
+      name: param.annotation
+      for name, param in sig.parameters.items()
+      if param.kind != param.KEYWORD_ONLY}
+    params = {
+      name: param.default if param.default is not inspect._empty else None
+      for name, param in sig.parameters.items()
+      if param.kind == param.KEYWORD_ONLY}
+    op = Op(func, name, params=params, inputs=inputs, outputs={'output': 'yes'}, type=type)
+    ALL_OPS[name] = op
+    return func
+  return decorator

web/.gitignore

@@ -11,14 +11,3 @@ node_modules
 dist
 dist-ssr
 *.local
-
-# Editor directories and files
-.vscode/*
-!.vscode/extensions.json
-.idea
-.DS_Store
-*.suo
-*.ntvs*
-*.njsproj
-*.sln
-*.sw?

web/src/LynxKiteFlow.svelte

@@ -37,7 +37,7 @@
       id: '3',
       type: 'basic',
       data: { title: 'Import Parquet', params: { filename: '/tmp/x.parquet' } },
-      position: { x: -300, y: 0 },
+      position: { x: -400, y: 0 },
       sourcePosition: Position.Right,
     },
     {
@@ -54,17 +54,13 @@
       id: '3-1',
       source: '3',
       target: '1',
-      markerEnd: {
-        type: MarkerType.ArrowClosed,
-      },
+      // markerEnd: { type: MarkerType.ArrowClosed },
     },
     {
       id: '3-4',
       source: '1',
       target: '4',
-      markerEnd: {
-        type: MarkerType.ArrowClosed,
-      },
+      // markerEnd: { type: MarkerType.ArrowClosed },
     },
   ]);
 
@@ -105,30 +101,13 @@
     });
     closeNodeSearch();
   }
-
-  const boxes = [
-    {
-      type: 'basic',
-      data: { title: 'Import Parquet', params: { filename: '/tmp/x.parquet' } },
-      sourcePosition: Position.Right,
-    },
-    {
-      type: 'basic',
-      data: { title: 'Export Parquet', params: { filename: '/tmp/x.parquet' } },
-      targetPosition: Position.Left,
-    },
-    {
-      type: 'graphviz',
-      data: { title: 'Visualize graph', params: {} },
-      targetPosition: Position.Left,
-    },
-    {
-      type: 'basic',
-      data: { title: 'Compute PageRank', params: { damping: 0.85, iterations: 3 } },
-      sourcePosition: Position.Right,
-      targetPosition: Position.Left,
-    },
-  ];
+  const boxes = writable([]);
+  async function getBoxes() {
+    const res = await fetch('/api/catalog');
+    const j = await res.json();
+    boxes.set(j);
+  }
+  getBoxes();
 
   let nodeSearchPos: XYPosition | undefined = undefined;
 
@@ -143,6 +122,7 @@
     }
     const ws = JSON.stringify(g);
     if (ws === backendWorkspace) return;
+    console.log('current vs backend', '\n' + ws, '\n' + backendWorkspace);
     backendWorkspace = ws;
     const res = await fetch('/api/save', {
       method: 'POST',
@@ -152,9 +132,8 @@
       body: JSON.stringify(g),
     });
     const j = await res.json();
-    g.nodes[2].data.graph = j.graph;
-    backendWorkspace = JSON.stringify(g);
-    nodes.set(g.nodes);
+    backendWorkspace = JSON.stringify(j);
+    nodes.set(j.nodes);
   });
 </script>
 
@@ -165,10 +144,9 @@
     maxZoom={1.5}
     minZoom={0.3}
     >
-    <Background />
+    <Background patternColor="#39bcf3" />
     <Controls />
-    <Background />
     <MiniMap />
-    {#if nodeSearchPos}<NodeSearch boxes={boxes} on:cancel={closeNodeSearch} on:add={addNode} pos={nodeSearchPos} />{/if}
+    {#if nodeSearchPos}<NodeSearch boxes={$boxes} on:cancel={closeNodeSearch} on:add={addNode} pos={nodeSearchPos} />{/if}
   </SvelteFlow>
 </div>

web/src/LynxKiteNode.svelte

@@ -28,8 +28,12 @@
   <div class="lynxkite-node">
     <div class="title" on:click={titleClicked}>
       {data.title}
+      {#if data.error}<span class="error-sign">⚠️</span>{/if}
     </div>
     {#if expanded}
+      {#if data.error}
+        <div class="error">{data.error}</div>
+      {/if}
       <slot />
     {/if}
     {#if sourcePosition}
@@ -42,8 +46,18 @@
 </div>
 
 <style>
+  .error {
+    background: #ffdddd;
+    padding: 8px;
+    font-size: 12px;
+  }
+  .error-sign {
+    float: right;
+  }
   .node-container {
     padding: 8px;
+    min-width: 170px;
+    max-width: 300px;
   }
   .lynxkite-node {
     box-shadow: 0px 5px 50px 0px rgba(0, 0, 0, 0.3);
@@ -53,7 +67,5 @@
     background: #ff8800;
     font-weight: bold;
     padding: 8px;
-    min-width: 170px;
-    max-width: 300px;
   }
 </style>

web/src/NodeSearch.svelte

@@ -8,7 +8,7 @@
   let hits = [];
   let selectedIndex = 0;
   onMount(() => searchBox.focus());
-  const fuse = new Fuse(boxes, {
+  $: fuse = new Fuse(boxes, {
     keys: ['data.title']
   })
   function onInput() {