Backend can now put results in the CRDT.

server/crdt.py

@@ -1,4 +1,5 @@
 '''CRDT is used to synchronize workspace state for backend and frontend(s).'''
+import asyncio
 import contextlib
 import fastapi
 import os.path
@@ -10,6 +11,7 @@ import pycrdt_websocket.ystore
 router = fastapi.APIRouter()
 
 def ws_exception_handler(exception, log):
+    print('exception', exception)
     log.exception(exception)
     return True
 
@@ -18,34 +20,93 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
         ystore = pycrdt_websocket.ystore.FileYStore(f'crdt_data/{name}.crdt')
         ydoc = pycrdt.Doc()
         ydoc['workspace'] = ws = pycrdt.Map()
-        ws['nodes'] = pycrdt.Array()
-        ws['edges'] = pycrdt.Array()
-        ws['env'] = 'unset'
         # Replay updates from the store.
         try:
             for update, timestamp in [(item[0], item[-1]) async for item in ystore.read()]:
                 ydoc.apply_update(update)
         except pycrdt_websocket.ystore.YDocNotFound:
             pass
-        print('init_room', name, ws)
-        _subscription_id = ws.observe_deep(handle_deep_changes)
-        return pycrdt_websocket.YRoom(ystore=ystore, ydoc=ydoc)
+        if 'nodes' not in ws:
+            ws['nodes'] = pycrdt.Array()
+        if 'edges' not in ws:
+            ws['edges'] = pycrdt.Array()
+        if 'env' not in ws:
+            ws['env'] = 'unset'
+        room = pycrdt_websocket.YRoom(ystore=ystore, ydoc=ydoc)
+        room.ws = ws
+        def on_change(changes):
+            asyncio.create_task(workspace_changed(changes, ws))
+        ws.observe_deep(on_change)
+        return room
 
     async def get_room(self, name: str) -> pycrdt_websocket.YRoom:
-        print('get_room', name, self.rooms)
         if name not in self.rooms:
             self.rooms[name] = await self.init_room(name)
-        print('get_room2', name, self.rooms)
         room = self.rooms[name]
         await self.start_room(room)
         return room
 
-print('new WebsocketServer')
 websocket_server = WebsocketServer(exception_handler=ws_exception_handler, auto_clean_rooms=False)
 asgi_server = pycrdt_websocket.ASGIServer(websocket_server)
 
-def handle_deep_changes(events):
-    print('events', events)
+last_ws_input = None
+def clean_input(ws_pyd):
+    for node in ws_pyd.nodes:
+        node.data.display = None
+        node.position.x = 0
+        node.position.y = 0
+        if node.model_extra:
+            for key in list(node.model_extra.keys()):
+                delattr(node, key)
+
+def crdt_update(crdt_obj, python_obj):
+    if isinstance(python_obj, dict):
+        for key, value in python_obj.items():
+            if isinstance(value, dict):
+                if crdt_obj.get(key) is None:
+                    crdt_obj[key] = pycrdt.Map()
+                crdt_update(crdt_obj[key], value)
+            elif isinstance(value, list):
+                if crdt_obj.get(key) is None:
+                    crdt_obj[key] = pycrdt.Array()
+                crdt_update(crdt_obj[key], value)
+            else:
+                print('set', key, value)
+                crdt_obj[key] = value
+    elif isinstance(python_obj, list):
+        for i, value in enumerate(python_obj):
+            if isinstance(value, dict):
+                if i >= len(crdt_obj):
+                    crdt_obj.append(pycrdt.Map())
+                crdt_update(crdt_obj[i], value)
+            elif isinstance(value, list):
+                if i >= len(crdt_obj):
+                    crdt_obj.append(pycrdt.Array())
+                crdt_update(crdt_obj[i], value)
+            else:
+                if i >= len(crdt_obj):
+                    crdt_obj.append(value)
+                else:
+                    print('set', i, value)
+                    crdt_obj[i] = value
+    else:
+        raise ValueError('Invalid type:', python_obj)
+
+async def workspace_changed(e, ws_crdt):
+    global last_ws_input
+    from . import workspace
+    ws_pyd = workspace.Workspace.model_validate(ws_crdt.to_py())
+    clean_input(ws_pyd)
+    if ws_pyd == last_ws_input:
+        return
+    print('ws changed')
+    last_ws_input = ws_pyd.model_copy(deep=True)
+    workspace.execute(ws_pyd)
+    for nc, np in zip(ws_crdt['nodes'], ws_pyd.nodes):
+        if 'data' not in nc:
+            nc['data'] = pycrdt.Map()
+        # Display is added as an opaque Box.
+        nc['data']['display'] = np.data.display
 
 @contextlib.asynccontextmanager
 async def lifespan(app):
@@ -58,5 +119,4 @@ def sanitize_path(path):
 @router.websocket("/ws/crdt/{room_name}")
 async def crdt_websocket(websocket: fastapi.WebSocket, room_name: str):
     room_name = sanitize_path(room_name)
-    print('room_name', room_name)
     await asgi_server({'path': room_name}, websocket._receive, websocket._send)

server/lynxkite_ops.py

@@ -105,11 +105,6 @@ def execute(ws):
                 data = node.data
                 op = catalog[data.title]
                 params = {**data.params}
-                if op.sub_nodes:
-                    sub_nodes = children.get(node.id, [])
-                    sub_node_ids = [node.id for node in sub_nodes]
-                    sub_edges = [edge for edge in ws.edges if edge.source in sub_node_ids]
-                    params['sub_flow'] = {'nodes': sub_nodes, 'edges': sub_edges}
                 # Convert inputs.
                 for i, (x, p) in enumerate(zip(inputs, op.inputs.values())):
                   if p.type == nx.Graph and isinstance(x, Bundle):

web/src/LynxKiteFlow.svelte

@@ -136,7 +136,9 @@
     const edge = {
       id: `${params.source} ${params.target}`,
       source: params.source,
+      sourceHandle: params.sourceHandle,
       target: params.target,
+      targetHandle: params.targetHandle,
     };
     $store.workspace.edges.push(edge);
   }

web/src/NodeWithParams.svelte

@@ -15,10 +15,11 @@
     $store.workspace.nodes[i].data.params[name] = newValue;
     updateNodeInternals();
   }
+  $: params = data?.params ? Object.entries(data.params) : [];
 </script>
 
 <LynxKiteNode {...$$props}>
-  {#each Object.entries(data.params) as [name, value]}
+  {#each params as [name, value]}
     <NodeParameter
       {name}
       {value}

web/src/NodeWithVisualization.svelte

@@ -5,11 +5,12 @@
   import { init } from 'echarts';
   type $$Props = NodeProps;
   export let data: $$Props['data'];
+  $: display = JSON.parse(JSON.stringify(data?.display?.value));
 </script>
 
 <NodeWithParams {...$$props}>
   {#if data.display}
-    <Chart {init} options={data.display} initOptions={{renderer: 'canvas', width: 250, height: 250}}/>
+    <Chart {init} options={display} initOptions={{renderer: 'canvas', width: 250, height: 250}}/>
   {/if}
 </NodeWithParams>
 <style>