Add a delayed execution mechanism for parameter edits.

server/crdt.py

@@ -50,7 +50,7 @@ class WebsocketServer(pycrdt_websocket.WebsocketServer):
         room.ws = ws
 
         def on_change(changes):
-            asyncio.create_task(workspace_changed(changes, ws))
+            asyncio.create_task(workspace_changed(name, changes, ws))
 
         ws.observe_deep(on_change)
         return room
@@ -69,6 +69,7 @@ last_ws_input = None
 def clean_input(ws_pyd):
     for node in ws_pyd.nodes:
         node.data.display = None
+        node.data.error = None
         node.position.x = 0
         node.position.y = 0
         if node.model_extra:
@@ -121,15 +122,39 @@ def try_to_load_workspace(ws, name):
         crdt_update(ws, ws_pyd.model_dump(), boxes={"display"})
 
 
-async def workspace_changed(e, ws_crdt):
-    global last_ws_input
+last_known_versions = {}
+delayed_executions = {}
+
+
+async def workspace_changed(name, changes, ws_crdt):
     from . import workspace
 
     ws_pyd = workspace.Workspace.model_validate(ws_crdt.to_py())
     clean_input(ws_pyd)
-    if ws_pyd == last_ws_input:
+    if ws_pyd == last_known_versions.get(name):
         return
-    last_ws_input = ws_pyd.model_copy(deep=True)
+    last_known_versions[name] = ws_pyd.model_copy(deep=True)
+    if name in delayed_executions:
+        delayed_executions[name].cancel()
+    delay = min(
+        change.keys.get("__execution_delay", {}).get("newValue", 0)
+        for change in changes
+    )
+    if delay > 0:
+        task = asyncio.create_task(execute(ws_crdt, ws_pyd, delay))
+        delayed_executions[name] = task
+    else:
+        await execute(ws_crdt, ws_pyd)
+
+
+async def execute(ws_crdt, ws_pyd, delay=0):
+    from . import workspace
+
+    if delay:
+        try:
+            await asyncio.sleep(delay)
+        except asyncio.CancelledError:
+            return
     await workspace.execute(ws_pyd)
     for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
         if "data" not in nc:
@@ -147,7 +172,7 @@ async def lifespan(app):
     )
     async with websocket_server:
         yield
-    print("closing websocket server for some reason")
+    print("closing websocket server")
 
 
 def sanitize_path(path):

web/src/workspace/Workspace.tsx

@@ -92,12 +92,20 @@ function LynxKiteFlow() {
       if (!node) continue;
       // Position events sometimes come with NaN values. Ignore them.
       if (ch.type === 'position' && !isNaN(ch.position.x) && !isNaN(ch.position.y)) {
-        Object.assign(node.position, ch.position);
+        getYjsDoc(state).transact(() => {
+          Object.assign(node.position, ch.position);
+        });
       } else if (ch.type === 'select') {
       } else if (ch.type === 'dimensions') {
       } else if (ch.type === 'replace') {
-        node.data.collapsed = ch.item.data.collapsed;
-        node.data.params = { ...ch.item.data.params };
+        // Ideally we would only update the parameter that changed. But ReactFlow does not give us that detail.
+        getYjsDoc(state).transact(() => {
+          Object.assign(node.data, {
+            collapsed: ch.item.data.collapsed,
+            params: { ...ch.item.data.params },
+            __execution_delay: ch.item.data.__execution_delay,
+          });
+        });
       } else {
         console.log('Unknown node change', ch);
       }

web/src/workspace/nodes/NodeParameter.tsx

@@ -42,7 +42,8 @@ export default function NodeParameter({ name, value, meta, onChange }) {
         <ParamName name={name} />
         <input className="input input-bordered w-full max-w-xs"
           value={value || ""}
-          onChange={(evt) => onChange(evt.currentTarget.value)}
+          onChange={(evt) => onChange(evt.currentTarget.value, { delay: 2 })}
+          onBlur={(evt) => onChange(evt.currentTarget.value, { delay: 0 })}
         />
       </>
       }

web/src/workspace/nodes/NodeWithParams.tsx

@@ -1,15 +1,14 @@
-import { useContext } from 'react';
-import { LynxKiteState } from '../LynxKiteState';
 import LynxKiteNode from './LynxKiteNode';
 import { useReactFlow } from '@xyflow/react';
 import NodeParameter from './NodeParameter';
 
+export type UpdateOptions = { delay?: number };
+
 function NodeWithParams(props: any) {
   const reactFlow = useReactFlow();
   const metaParams = props.data.meta?.params;
-  const state = useContext(LynxKiteState);
-  function setParam(name: string, newValue: any) {
-    reactFlow.updateNodeData(props.id, { params: { ...props.data.params, [name]: newValue } });
+  function setParam(name: string, newValue: any, opts: UpdateOptions) {
+    reactFlow.updateNodeData(props.id, { params: { ...props.data.params, [name]: newValue }, __execution_delay: opts.delay || 0 });
   }
   const params = props.data?.params ? Object.entries(props.data.params) : [];
 
@@ -21,7 +20,7 @@ function NodeWithParams(props: any) {
           key={name}
           value={value}
           meta={metaParams?.[name]}
-          onChange={(value: any) => setParam(name, value)}
+          onChange={(value: any, opts?: UpdateOptions) => setParam(name, value, opts || {})}
         />
       )}
     </LynxKiteNode >