Update node display/error in CRDT one by one during execution.

lynxkite-app/src/lynxkite_app/crdt.py

@@ -224,17 +224,16 @@ async def execute(
     assert path.is_relative_to(config.DATA_PATH), "Provided workspace path is invalid"
     # Save user changes before executing, in case the execution fails.
     workspace.save(ws_pyd, path)
+    add_crdt_bindings(ws_pyd, ws_crdt)
     await workspace.execute(ws_pyd)
     workspace.save(ws_pyd, path)
-    # Execution happened on the Python object, we need to replicate
-    # the results to the CRDT object.
-    with ws_crdt.doc.transaction():
-        for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
-            if "data" not in nc:
-                nc["data"] = pycrdt.Map()
-            # Display is added as a non collaborative field.
-            nc["data"]["display"] = np.data.display
-            nc["data"]["error"] = np.data.error
+
+
+def add_crdt_bindings(ws_pyd: workspace.Workspace, ws_crdt: pycrdt.Map):
+    for nc, np in zip(ws_crdt["nodes"], ws_pyd.nodes):
+        if "data" not in nc:
+            nc["data"] = pycrdt.Map()
+        np._crdt = nc
 
 
 @contextlib.asynccontextmanager

lynxkite-app/src/lynxkite_app/main.py

@@ -5,6 +5,7 @@ import shutil
 import pydantic
 import fastapi
 import importlib
+import pandas as pd
 import pathlib
 import pkgutil
 from fastapi.staticfiles import StaticFiles
@@ -18,6 +19,8 @@ if os.environ.get("NX_CUGRAPH_AUTOCONFIG", "").strip().lower() == "true":
 
     cudf.pandas.install()
 
+pd.options.mode.copy_on_write = True  # Prepare for Pandas 3.0.
+
 
 def detect_plugins():
     plugins = {}

lynxkite-core/src/lynxkite/core/ops.py

@@ -94,8 +94,9 @@ class Result:
     JSON-serializable.
     """
 
-    output: typing.Any
+    output: typing.Any = None
     display: ReadOnlyJSON | None = None
+    error: str | None = None
 
 
 MULTI_INPUT = Input(name="multi", type="*")

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

@@ -4,6 +4,7 @@ import json
 from typing import Optional
 import dataclasses
 import os
+import pycrdt
 import pydantic
 import tempfile
 from . import ops
@@ -36,6 +37,16 @@ class WorkspaceNode(BaseConfig):
     type: str
     data: WorkspaceNodeData
     position: Position
+    _crdt: pycrdt.Map
+
+    def publish_result(self, result: ops.Result):
+        """Sends the result to the frontend. Call this in an executor when the result is available."""
+        with self._crdt.doc.transaction():
+            self._crdt["data"]["display"] = result.display
+            self._crdt["data"]["error"] = result.error
+
+    def publish_error(self, error: Exception | str):
+        self.publish_result(ops.Result(error=str(error)))
 
 
 class WorkspaceEdge(BaseConfig):

lynxkite-graph-analytics/src/lynxkite_graph_analytics/__init__.py

@@ -1,3 +1,4 @@
-from .lynxkite_ops import *  # noqa (imported to trigger registration)
+from .core import *  # noqa (easier access for core classes)
+from . import lynxkite_ops  # noqa (imported to trigger registration)
 from . import networkx_ops  # noqa (imported to trigger registration)
 from . import pytorch_model_ops  # noqa (imported to trigger registration)

lynxkite-graph-analytics/src/lynxkite_graph_analytics/core.py

@@ -158,11 +158,10 @@ async def execute(ws):
             if all(input in outputs for input in inputs):
                 # All inputs for this node are ready, we can compute the output.
                 inputs = [outputs[input] for input in inputs]
-                data = node.data
-                params = {**data.params}
-                op = catalog.get(data.title)
+                params = {**node.data.params}
+                op = catalog.get(node.data.title)
                 if not op:
-                    data.error = "Operation not found in catalog"
+                    node.publish_error("Operation not found in catalog")
                     failed += 1
                     continue
                 try:
@@ -177,16 +176,11 @@ async def execute(ws):
                     result = op(*inputs, **params)
                 except Exception as e:
                     traceback.print_exc()
-                    data.error = str(e)
+                    node.publish_error(e)
                     failed += 1
                     continue
-                if len(op.inputs) == 1 and op.inputs.get("multi") == "*":
-                    # It's a flexible input. Create n+1 handles.
-                    data.inputs = {f"input{i}": None for i in range(len(inputs) + 1)}
-                data.error = None
                 outputs[node.id] = result.output
-                if result.display:
-                    data.display = result.display
+                node.publish_result(result)
 
 
 def df_for_frontend(df: pd.DataFrame, limit: int) -> pd.DataFrame:

lynxkite-graph-analytics/src/lynxkite_graph_analytics/lynxkite_ops.py

@@ -152,7 +152,7 @@ def _map_color(value):
     if pd.api.types.is_numeric_dtype(value):
         cmap = matplotlib.cm.get_cmap("viridis")
         value = (value - value.min()) / (value.max() - value.min())
-        rgba = cmap(value)
+        rgba = cmap(value.values)
         return [
             "#{:02x}{:02x}{:02x}".format(int(r * 255), int(g * 255), int(b * 255))
             for r, g, b in rgba[:, :3]