Remove sub_nodes/sub_flow/parentId.

server/lynxkite_ops.py

@@ -1,4 +1,5 @@
-'''Some operations. To be split into separate files when we have more.'''
+"""Some operations. To be split into separate files when we have more."""
+
 from . import ops
 from collections import deque
 import dataclasses
@@ -9,72 +10,85 @@ import pandas as pd
 import traceback
 import typing
 
-op = ops.op_registration('LynxKite')
+op = ops.op_registration("LynxKite")
+
 
 @dataclasses.dataclass
 class RelationDefinition:
-  '''Defines a set of edges.'''
-  df: str # The DataFrame that contains the edges.
-  source_column: str # The column in the edge DataFrame that contains the source node ID.
-  target_column: str # The column in the edge DataFrame that contains the target node ID.
-  source_table: str # The DataFrame that contains the source nodes.
-  target_table: str # The DataFrame that contains the target nodes.
-  source_key: str # The column in the source table that contains the node ID.
-  target_key: str # The column in the target table that contains the node ID.
+    """Defines a set of edges."""
+
+    df: str  # The DataFrame that contains the edges.
+    source_column: (
+        str  # The column in the edge DataFrame that contains the source node ID.
+    )
+    target_column: (
+        str  # The column in the edge DataFrame that contains the target node ID.
+    )
+    source_table: str  # The DataFrame that contains the source nodes.
+    target_table: str  # The DataFrame that contains the target nodes.
+    source_key: str  # The column in the source table that contains the node ID.
+    target_key: str  # The column in the target table that contains the node ID.
+
 
 @dataclasses.dataclass
 class Bundle:
-  '''A collection of DataFrames and other data.
-
-  Can efficiently represent a knowledge graph (homogeneous or heterogeneous) or tabular data.
-  It can also carry other data, such as a trained model.
-  '''
-  dfs: dict[str, pd.DataFrame] = dataclasses.field(default_factory=dict)
-  relations: list[RelationDefinition] = dataclasses.field(default_factory=list)
-  other: dict[str, typing.Any] = None
-
-  @classmethod
-  def from_nx(cls, graph: nx.Graph):
-    edges = nx.to_pandas_edgelist(graph)
-    d = dict(graph.nodes(data=True))
-    nodes = pd.DataFrame(d.values(), index=d.keys())
-    nodes['id'] = nodes.index
-    return cls(
-      dfs={'edges': edges, 'nodes': nodes},
-      relations=[
-        RelationDefinition(
-          df='edges',
-          source_column='source',
-          target_column='target',
-          source_table='nodes',
-          target_table='nodes',
-          source_key='id',
-          target_key='id',
+    """A collection of DataFrames and other data.
+
+    Can efficiently represent a knowledge graph (homogeneous or heterogeneous) or tabular data.
+    It can also carry other data, such as a trained model.
+    """
+
+    dfs: dict[str, pd.DataFrame] = dataclasses.field(default_factory=dict)
+    relations: list[RelationDefinition] = dataclasses.field(default_factory=list)
+    other: dict[str, typing.Any] = None
+
+    @classmethod
+    def from_nx(cls, graph: nx.Graph):
+        edges = nx.to_pandas_edgelist(graph)
+        d = dict(graph.nodes(data=True))
+        nodes = pd.DataFrame(d.values(), index=d.keys())
+        nodes["id"] = nodes.index
+        return cls(
+            dfs={"edges": edges, "nodes": nodes},
+            relations=[
+                RelationDefinition(
+                    df="edges",
+                    source_column="source",
+                    target_column="target",
+                    source_table="nodes",
+                    target_table="nodes",
+                    source_key="id",
+                    target_key="id",
+                )
+            ],
         )
-      ]
-    )
 
-  def to_nx(self):
-    graph = nx.from_pandas_edgelist(self.dfs['edges'])
-    nx.set_node_attributes(graph, self.dfs['nodes'].set_index('id').to_dict('index'))
-    return graph
+    def to_nx(self):
+        graph = nx.from_pandas_edgelist(self.dfs["edges"])
+        nx.set_node_attributes(
+            graph, self.dfs["nodes"].set_index("id").to_dict("index")
+        )
+        return graph
 
 
 def nx_node_attribute_func(name):
-  '''Decorator for wrapping a function that adds a NetworkX node attribute.'''
-  def decorator(func):
-    @functools.wraps(func)
-    def wrapper(graph: nx.Graph, **kwargs):
-      graph = graph.copy()
-      attr = func(graph, **kwargs)
-      nx.set_node_attributes(graph, attr, name)
-      return graph
-    return wrapper
-  return decorator
+    """Decorator for wrapping a function that adds a NetworkX node attribute."""
+
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(graph: nx.Graph, **kwargs):
+            graph = graph.copy()
+            attr = func(graph, **kwargs)
+            nx.set_node_attributes(graph, attr, name)
+            return graph
+
+        return wrapper
+
+    return decorator
 
 
 def disambiguate_edges(ws):
-    '''If an input plug is connected to multiple edges, keep only the last edge.'''
+    """If an input plug is connected to multiple edges, keep only the last edge."""
     seen = set()
     for edge in reversed(ws.edges):
         if (edge.target, edge.targetHandle) in seen:
@@ -82,20 +96,15 @@ def disambiguate_edges(ws):
         seen.add((edge.target, edge.targetHandle))
 
 
-@ops.register_executor('LynxKite')
+@ops.register_executor("LynxKite")
 async def execute(ws):
-    catalog = ops.CATALOGS['LynxKite']
+    catalog = ops.CATALOGS["LynxKite"]
     # Nodes are responsible for interpreting/executing their child nodes.
-    nodes = [n for n in ws.nodes if not n.parentId]
     disambiguate_edges(ws)
-    children = {}
-    for n in ws.nodes:
-        if n.parentId:
-            children.setdefault(n.parentId, []).append(n)
     outputs = {}
     failed = 0
-    while len(outputs) + failed < len(nodes):
-        for node in nodes:
+    while len(outputs) + failed < len(ws.nodes):
+        for node in ws.nodes:
             if node.id in outputs:
                 continue
             # TODO: Take the input/output handles into account.
@@ -107,118 +116,138 @@ async def execute(ws):
                 params = {**data.params}
                 # Convert inputs.
                 for i, (x, p) in enumerate(zip(inputs, op.inputs.values())):
-                  if p.type == nx.Graph and isinstance(x, Bundle):
-                    inputs[i] = x.to_nx()
-                  elif p.type == Bundle and isinstance(x, nx.Graph):
-                    inputs[i] = Bundle.from_nx(x)
+                    if p.type == nx.Graph and isinstance(x, Bundle):
+                        inputs[i] = x.to_nx()
+                    elif p.type == Bundle and isinstance(x, nx.Graph):
+                        inputs[i] = Bundle.from_nx(x)
                 try:
-                  output = op(*inputs, **params)
+                    output = op(*inputs, **params)
                 except Exception as e:
-                  traceback.print_exc()
-                  data.error = str(e)
-                  failed += 1
-                  continue
-                if len(op.inputs) == 1 and op.inputs.get('multi') == '*':
+                    traceback.print_exc()
+                    data.error = str(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.inputs = {f"input{i}": None for i in range(len(inputs) + 1)}
                 data.error = None
                 outputs[node.id] = output
-                if op.type == 'visualization' or op.type == 'table_view' or op.type == 'image':
+                if (
+                    op.type == "visualization"
+                    or op.type == "table_view"
+                    or op.type == "image"
+                ):
                     data.display = output
 
+
 @op("Import Parquet")
 def import_parquet(*, filename: str):
-  '''Imports a parquet file.'''
-  return pd.read_parquet(filename)
+    """Imports a parquet file."""
+    return pd.read_parquet(filename)
+
 
 @op("Create scale-free graph")
 def create_scale_free_graph(*, nodes: int = 10):
-  '''Creates a scale-free graph with the given number of nodes.'''
-  return nx.scale_free_graph(nodes)
+    """Creates a scale-free graph with the given number of nodes."""
+    return nx.scale_free_graph(nodes)
+
 
 @op("Compute PageRank")
-@nx_node_attribute_func('pagerank')
+@nx_node_attribute_func("pagerank")
 def compute_pagerank(graph: nx.Graph, *, damping=0.85, iterations=100):
-  return nx.pagerank(graph, alpha=damping, max_iter=iterations)
+    return nx.pagerank(graph, alpha=damping, max_iter=iterations)
+
 
 @op("Discard loop edges")
 def discard_loop_edges(graph: nx.Graph):
-  graph = graph.copy()
-  graph.remove_edges_from(nx.selfloop_edges(graph))
-  return graph
+    graph = graph.copy()
+    graph.remove_edges_from(nx.selfloop_edges(graph))
+    return graph
+
 
 @op("Sample graph")
 def sample_graph(graph: nx.Graph, *, nodes: int = 100):
-  '''Takes a (preferably connected) subgraph.'''
-  sample = set()
-  to_expand = deque([0])
-  while to_expand and len(sample) < nodes:
-    node = to_expand.pop()
-    for n in graph.neighbors(node):
-      if n not in sample:
-        sample.add(n)
-        to_expand.append(n)
-      if len(sample) == nodes:
-        break
-  return nx.Graph(graph.subgraph(sample))
+    """Takes a (preferably connected) subgraph."""
+    sample = set()
+    to_expand = deque([0])
+    while to_expand and len(sample) < nodes:
+        node = to_expand.pop()
+        for n in graph.neighbors(node):
+            if n not in sample:
+                sample.add(n)
+                to_expand.append(n)
+            if len(sample) == nodes:
+                break
+    return nx.Graph(graph.subgraph(sample))
+
 
 def _map_color(value):
-  cmap = matplotlib.cm.get_cmap('viridis')
-  value = (value - value.min()) / (value.max() - value.min())
-  rgba = cmap(value)
-  return ['#{:02x}{:02x}{:02x}'.format(int(r*255), int(g*255), int(b*255)) for r, g, b in rgba[:, :3]]
+    cmap = matplotlib.cm.get_cmap("viridis")
+    value = (value - value.min()) / (value.max() - value.min())
+    rgba = cmap(value)
+    return [
+        "#{:02x}{:02x}{:02x}".format(int(r * 255), int(g * 255), int(b * 255))
+        for r, g, b in rgba[:, :3]
+    ]
+
 
 @op("Visualize graph", view="visualization")
 def visualize_graph(graph: Bundle, *, color_nodes_by: ops.NodeAttribute = None):
-  nodes = graph.dfs['nodes'].copy()
-  if color_nodes_by:
-    nodes['color'] = _map_color(nodes[color_nodes_by])
-  nodes = nodes.to_records()
-  edges = graph.dfs['edges'].drop_duplicates(['source', 'target'])
-  edges = edges.to_records()
-  pos = nx.spring_layout(graph.to_nx(), iterations=max(1, int(10000/len(nodes))))
-  v = {
-    'animationDuration': 500,
-    'animationEasingUpdate': 'quinticInOut',
-    'series': [
-      {
-        'type': 'graph',
-        'roam': True,
-        'lineStyle': {
-          'color': 'gray',
-          'curveness': 0.3,
-        },
-        'emphasis': {
-          'focus': 'adjacency',
-          'lineStyle': {
-            'width': 10,
-          }
-        },
-        'data': [
-          {
-            'id': str(n.id),
-            'x': float(pos[n.id][0]), 'y': float(pos[n.id][1]),
-            # Adjust node size to cover the same area no matter how many nodes there are.
-            'symbolSize': 50 / len(nodes) ** 0.5,
-            'itemStyle': {'color': n.color} if color_nodes_by else {},
-          }
-          for n in nodes],
-        'links': [
-          {'source': str(r.source), 'target': str(r.target)}
-          for r in edges],
-      },
-    ],
-  }
-  return v
+    nodes = graph.dfs["nodes"].copy()
+    if color_nodes_by:
+        nodes["color"] = _map_color(nodes[color_nodes_by])
+    nodes = nodes.to_records()
+    edges = graph.dfs["edges"].drop_duplicates(["source", "target"])
+    edges = edges.to_records()
+    pos = nx.spring_layout(graph.to_nx(), iterations=max(1, int(10000 / len(nodes))))
+    v = {
+        "animationDuration": 500,
+        "animationEasingUpdate": "quinticInOut",
+        "series": [
+            {
+                "type": "graph",
+                "roam": True,
+                "lineStyle": {
+                    "color": "gray",
+                    "curveness": 0.3,
+                },
+                "emphasis": {
+                    "focus": "adjacency",
+                    "lineStyle": {
+                        "width": 10,
+                    },
+                },
+                "data": [
+                    {
+                        "id": str(n.id),
+                        "x": float(pos[n.id][0]),
+                        "y": float(pos[n.id][1]),
+                        # Adjust node size to cover the same area no matter how many nodes there are.
+                        "symbolSize": 50 / len(nodes) ** 0.5,
+                        "itemStyle": {"color": n.color} if color_nodes_by else {},
+                    }
+                    for n in nodes
+                ],
+                "links": [
+                    {"source": str(r.source), "target": str(r.target)} for r in edges
+                ],
+            },
+        ],
+    }
+    return v
+
 
 @op("View tables", view="table_view")
 def view_tables(bundle: Bundle):
-  v = {
-    'dataframes': { name: {
-      'columns': [str(c) for c in df.columns],
-      'data': df.values.tolist(),
-    } for name, df in bundle.dfs.items() },
-    'relations': bundle.relations,
-    'other': bundle.other,
-  }
-  return v
+    v = {
+        "dataframes": {
+            name: {
+                "columns": [str(c) for c in df.columns],
+                "data": df.values.tolist(),
+            }
+            for name, df in bundle.dfs.items()
+        },
+        "relations": bundle.relations,
+        "other": bundle.other,
+    }
+    return v

server/ops.py

@@ -1,4 +1,5 @@
-'''API for implementing LynxKite operations.'''
+"""API for implementing LynxKite operations."""
+
 from __future__ import annotations
 import enum
 import functools
@@ -10,180 +11,214 @@ from typing_extensions import Annotated
 CATALOGS = {}
 EXECUTORS = {}
 
-typeof = type # We have some arguments called "type".
+typeof = type  # We have some arguments called "type".
+
+
 def type_to_json(t):
-  if isinstance(t, type) and issubclass(t, enum.Enum):
-    return {'enum': list(t.__members__.keys())}
-  if getattr(t, '__metadata__', None):
-    return t.__metadata__[-1]
-  return {'type': str(t)}
-Type = Annotated[
-  typing.Any, pydantic.PlainSerializer(type_to_json, return_type=dict)
-]
-LongStr = Annotated[
-  str, {'format': 'textarea'}
-]
-PathStr = Annotated[
-  str, {'format': 'path'}
-]
-CollapsedStr = Annotated[
-  str, {'format': 'collapsed'}
-]
-NodeAttribute = Annotated[
-  str, {'format': 'node attribute'}
-]
-EdgeAttribute = Annotated[
-  str, {'format': 'edge attribute'}
-]
+    if isinstance(t, type) and issubclass(t, enum.Enum):
+        return {"enum": list(t.__members__.keys())}
+    if getattr(t, "__metadata__", None):
+        return t.__metadata__[-1]
+    return {"type": str(t)}
+
+
+Type = Annotated[typing.Any, pydantic.PlainSerializer(type_to_json, return_type=dict)]
+LongStr = Annotated[str, {"format": "textarea"}]
+PathStr = Annotated[str, {"format": "path"}]
+CollapsedStr = Annotated[str, {"format": "collapsed"}]
+NodeAttribute = Annotated[str, {"format": "node attribute"}]
+EdgeAttribute = Annotated[str, {"format": "edge attribute"}]
+
+
 class BaseConfig(pydantic.BaseModel):
-  model_config = pydantic.ConfigDict(
-    arbitrary_types_allowed=True,
-  )
+    model_config = pydantic.ConfigDict(
+        arbitrary_types_allowed=True,
+    )
 
 
 class Parameter(BaseConfig):
-  '''Defines a parameter for an operation.'''
-  name: str
-  default: typing.Any
-  type: Type = None
-
-  @staticmethod
-  def options(name, options, default=None):
-    e = enum.Enum(f'OptionsFor_{name}', options)
-    return Parameter.basic(name, e[default or options[0]], e)
-
-  @staticmethod
-  def collapsed(name, default, type=None):
-    return Parameter.basic(name, default, CollapsedStr)
-
-  @staticmethod
-  def basic(name, default=None, type=None):
-    if default is inspect._empty:
-      default = None
-    if type is None or type is inspect._empty:
-      type = typeof(default) if default is not None else None
-    return Parameter(name=name, default=default, type=type)
+    """Defines a parameter for an operation."""
+
+    name: str
+    default: typing.Any
+    type: Type = None
+
+    @staticmethod
+    def options(name, options, default=None):
+        e = enum.Enum(f"OptionsFor_{name}", options)
+        return Parameter.basic(name, e[default or options[0]], e)
+
+    @staticmethod
+    def collapsed(name, default, type=None):
+        return Parameter.basic(name, default, CollapsedStr)
+
+    @staticmethod
+    def basic(name, default=None, type=None):
+        if default is inspect._empty:
+            default = None
+        if type is None or type is inspect._empty:
+            type = typeof(default) if default is not None else None
+        return Parameter(name=name, default=default, type=type)
+
 
 class Input(BaseConfig):
-  name: str
-  type: Type
-  position: str = 'left'
+    name: str
+    type: Type
+    position: str = "left"
+
 
 class Output(BaseConfig):
-  name: str
-  type: Type
-  position: str = 'right'
+    name: str
+    type: Type
+    position: str = "right"
+
+
+MULTI_INPUT = Input(name="multi", type="*")
+
 
-MULTI_INPUT = Input(name='multi', type='*')
 def basic_inputs(*names):
-  return {name: Input(name=name, type=None) for name in names}
+    return {name: Input(name=name, type=None) for name in names}
+
+
 def basic_outputs(*names):
-  return {name: Output(name=name, type=None) for name in names}
+    return {name: Output(name=name, type=None) for name in names}
 
 
 class Op(BaseConfig):
-  func: typing.Callable = pydantic.Field(exclude=True)
-  name: str
-  params: dict[str, Parameter]
-  inputs: dict[str, Input]
-  outputs: dict[str, Output]
-  type: str = 'basic' # The UI to use for this operation.
-  sub_nodes: list[Op] = None # If set, these nodes can be placed inside the operation's node.
-
-  def __call__(self, *inputs, **params):
-    # Convert parameters.
-    for p in params:
-      if p in self.params:
-        if self.params[p].type == int:
-          params[p] = int(params[p])
-        elif self.params[p].type == float:
-          params[p] = float(params[p])
-        elif isinstance(self.params[p].type, enum.EnumMeta):
-          params[p] = self.params[p].type[params[p]]
-    res = self.func(*inputs, **params)
-    return res
-
-
-def op(env: str, name: str, *, view='basic', sub_nodes=None, outputs=None):
-  '''Decorator for defining an operation.'''
-  def decorator(func):
-    sig = inspect.signature(func)
-    # Positional arguments are inputs.
-    inputs = {
-      name: Input(name=name, type=param.annotation)
-      for name, param in sig.parameters.items()
-      if param.kind != param.KEYWORD_ONLY}
-    params = {}
-    for n, param in sig.parameters.items():
-      if param.kind == param.KEYWORD_ONLY and not n.startswith('_'):
-        params[n] = Parameter.basic(n, param.default, param.annotation)
-    if outputs:
-      _outputs = {name: Output(name=name, type=None) for name in outputs}
-    else:
-      _outputs = {'output': Output(name='output', type=None)} if view == 'basic' else {}
-    op = Op(func=func, name=name, params=params, inputs=inputs, outputs=_outputs, type=view)
-    if sub_nodes is not None:
-      op.sub_nodes = sub_nodes
-      op.type = 'sub_flow'
-    CATALOGS.setdefault(env, {})
-    CATALOGS[env][name] = op
-    func.__op__ = op
-    return func
-  return decorator
+    func: typing.Callable = pydantic.Field(exclude=True)
+    name: str
+    params: dict[str, Parameter]
+    inputs: dict[str, Input]
+    outputs: dict[str, Output]
+    type: str = "basic"  # The UI to use for this operation.
+
+    def __call__(self, *inputs, **params):
+        # Convert parameters.
+        for p in params:
+            if p in self.params:
+                if self.params[p].type == int:
+                    params[p] = int(params[p])
+                elif self.params[p].type == float:
+                    params[p] = float(params[p])
+                elif isinstance(self.params[p].type, enum.EnumMeta):
+                    params[p] = self.params[p].type[params[p]]
+        res = self.func(*inputs, **params)
+        return res
+
+
+def op(env: str, name: str, *, view="basic", outputs=None):
+    """Decorator for defining an operation."""
+
+    def decorator(func):
+        sig = inspect.signature(func)
+        # Positional arguments are inputs.
+        inputs = {
+            name: Input(name=name, type=param.annotation)
+            for name, param in sig.parameters.items()
+            if param.kind != param.KEYWORD_ONLY
+        }
+        params = {}
+        for n, param in sig.parameters.items():
+            if param.kind == param.KEYWORD_ONLY and not n.startswith("_"):
+                params[n] = Parameter.basic(n, param.default, param.annotation)
+        if outputs:
+            _outputs = {name: Output(name=name, type=None) for name in outputs}
+        else:
+            _outputs = (
+                {"output": Output(name="output", type=None)} if view == "basic" else {}
+            )
+        op = Op(
+            func=func,
+            name=name,
+            params=params,
+            inputs=inputs,
+            outputs=_outputs,
+            type=view,
+        )
+        CATALOGS.setdefault(env, {})
+        CATALOGS[env][name] = op
+        func.__op__ = op
+        return func
+
+    return decorator
+
 
 def input_position(**kwargs):
-  '''Decorator for specifying unusual positions for the inputs.'''
-  def decorator(func):
-    op = func.__op__
-    for k, v in kwargs.items():
-      op.inputs[k].position = v
-    return func
-  return decorator
+    """Decorator for specifying unusual positions for the inputs."""
+
+    def decorator(func):
+        op = func.__op__
+        for k, v in kwargs.items():
+            op.inputs[k].position = v
+        return func
+
+    return decorator
+
 
 def output_position(**kwargs):
-  '''Decorator for specifying unusual positions for the outputs.'''
-  def decorator(func):
-    op = func.__op__
-    for k, v in kwargs.items():
-      op.outputs[k].position = v
-    return func
-  return decorator
+    """Decorator for specifying unusual positions for the outputs."""
+
+    def decorator(func):
+        op = func.__op__
+        for k, v in kwargs.items():
+            op.outputs[k].position = v
+        return func
+
+    return decorator
+
 
 def no_op(*args, **kwargs):
-  if args:
-    return args[0]
-  return None
-
-def register_passive_op(env: str, name: str, inputs=[], outputs=['output'], params=[]):
-  '''A passive operation has no associated code.'''
-  op = Op(
-    func=no_op,
-    name=name,
-    params={p.name: p for p in params},
-    inputs=dict(
-      (i, Input(name=i, type=None)) if isinstance(i, str)
-      else (i.name, i) for i in inputs),
-    outputs=dict(
-      (o, Output(name=o, type=None)) if isinstance(o, str)
-      else (o.name, o) for o in outputs))
-  CATALOGS.setdefault(env, {})
-  CATALOGS[env][name] = op
-  return op
+    if args:
+        return args[0]
+    return None
+
+
+def register_passive_op(env: str, name: str, inputs=[], outputs=["output"], params=[]):
+    """A passive operation has no associated code."""
+    op = Op(
+        func=no_op,
+        name=name,
+        params={p.name: p for p in params},
+        inputs=dict(
+            (i, Input(name=i, type=None)) if isinstance(i, str) else (i.name, i)
+            for i in inputs
+        ),
+        outputs=dict(
+            (o, Output(name=o, type=None)) if isinstance(o, str) else (o.name, o)
+            for o in outputs
+        ),
+    )
+    CATALOGS.setdefault(env, {})
+    CATALOGS[env][name] = op
+    return op
+
 
 def register_executor(env: str):
-  '''Decorator for registering an executor.'''
-  def decorator(func):
-    EXECUTORS[env] = func
-    return func
-  return decorator
+    """Decorator for registering an executor."""
+
+    def decorator(func):
+        EXECUTORS[env] = func
+        return func
+
+    return decorator
+
 
 def op_registration(env: str):
-  return functools.partial(op, env)
+    return functools.partial(op, env)
+
 
 def passive_op_registration(env: str):
-  return functools.partial(register_passive_op, env)
+    return functools.partial(register_passive_op, env)
+
 
 def register_area(env, name, params=[]):
-  '''A node that represents an area. It can contain other nodes, but does not restrict movement in any way.'''
-  op = Op(func=no_op, name=name, params={p.name: p for p in params}, inputs={}, outputs={}, type='area')
-  CATALOGS[env][name] = op
+    """A node that represents an area. It can contain other nodes, but does not restrict movement in any way."""
+    op = Op(
+        func=no_op,
+        name=name,
+        params={p.name: p for p in params},
+        inputs={},
+        outputs={},
+        type="area",
+    )
+    CATALOGS[env][name] = op

server/workspace.py

@@ -1,4 +1,5 @@
-'''For working with LynxKite workspaces.'''
+"""For working with LynxKite workspaces."""
+
 from typing import Optional
 import dataclasses
 import os
@@ -6,15 +7,18 @@ import pydantic
 import tempfile
 from . import ops
 
+
 class BaseConfig(pydantic.BaseModel):
     model_config = pydantic.ConfigDict(
-        extra='allow',
+        extra="allow",
     )
 
+
 class Position(BaseConfig):
     x: float
     y: float
 
+
 class WorkspaceNodeData(BaseConfig):
     title: str
     params: dict
@@ -23,12 +27,13 @@ class WorkspaceNodeData(BaseConfig):
     # Also contains a "meta" field when going out.
     # This is ignored when coming back from the frontend.
 
+
 class WorkspaceNode(BaseConfig):
     id: str
     type: str
     data: WorkspaceNodeData
     position: Position
-    parentId: Optional[str] = None
+
 
 class WorkspaceEdge(BaseConfig):
     id: str
@@ -37,8 +42,9 @@ class WorkspaceEdge(BaseConfig):
     sourceHandle: str
     targetHandle: str
 
+
 class Workspace(BaseConfig):
-    env: str = ''
+    env: str = ""
     nodes: list[WorkspaceNode] = dataclasses.field(default_factory=list)
     edges: list[WorkspaceEdge] = dataclasses.field(default_factory=list)
 
@@ -52,7 +58,9 @@ def save(ws: Workspace, path: str):
     j = ws.model_dump_json(indent=2)
     dirname, basename = os.path.split(path)
     # Create temp file in the same directory to make sure it's on the same filesystem.
-    with tempfile.NamedTemporaryFile('w', prefix=f'.{basename}.', dir=dirname, delete_on_close=False) as f:
+    with tempfile.NamedTemporaryFile(
+        "w", prefix=f".{basename}.", dir=dirname, delete_on_close=False
+    ) as f:
         f.write(j)
         f.close()
         os.replace(f.name, path)
@@ -76,22 +84,13 @@ def _update_metadata(ws):
             if node.id in done:
                 continue
             data = node.data
-            if node.parentId is None:
-                op = catalog.get(data.title)
-            elif node.parentId not in nodes:
-                data.error = f'Parent not found: {node.parentId}'
-                done.add(node.id)
-                continue
-            elif node.parentId in done:
-                op = nodes[node.parentId].data.meta.sub_nodes[data.title]
-            else:
-                continue
+            op = catalog.get(data.title)
             if op:
                 data.meta = op
                 node.type = op.type
-                if data.error == 'Unknown operation.':
+                if data.error == "Unknown operation.":
                     data.error = None
             else:
-                data.error = 'Unknown operation.'
+                data.error = "Unknown operation."
             done.add(node.id)
     return ws

web/src/apiTypes.ts

@@ -24,7 +24,6 @@ export interface WorkspaceNode {
   type: string;
   data: WorkspaceNodeData;
   position: Position;
-  parentId?: string | null;
   [k: string]: unknown;
 }
 export interface WorkspaceNodeData {

web/src/index.css

@@ -15,7 +15,8 @@
   background: #002a4c;
 }
 
-img, svg {
+img,
+svg {
   display: inline-block;
 }
 
@@ -156,6 +157,43 @@ body {
       line-height: 10px;
     }
   }
+
+  .node-search {
+    position: fixed;
+    width: 300px;
+    z-index: 5;
+    padding: 4px;
+    border-radius: 4px;
+    border: 1px solid #888;
+    background-color: white;
+    max-height: -webkit-fill-available;
+    max-height: -moz-available;
+    display: flex;
+    flex-direction: column;
+
+    input {
+      width: calc(100% - 26px);
+      font-size: 20px;
+      padding: 8px;
+      border-radius: 4px;
+      border: 1px solid #eee;
+      margin: 4px;
+    }
+
+    .search-result {
+      padding: 4px;
+      cursor: pointer;
+    }
+
+    .search-result.selected {
+      background-color: oklch(75% 0.2 55);
+      border-radius: 4px;
+    }
+
+    .matches {
+      overflow-y: auto;
+    }
+  }
 }
 
 .directory {
@@ -265,29 +303,29 @@ path.react-flow__edge-path {
   stroke-width: 2;
   stroke: black;
 }
+
 .react-flow__edge.selected path.react-flow__edge-path {
   outline: var(--xy-selection-border, var(--xy-selection-border-default));
   outline-offset: 10px;
   border-radius: 1px;
 }
+
 .react-flow__handle {
   border-color: black;
   background: white;
   width: 10px;
   height: 10px;
 }
+
 .react-flow__arrowhead * {
   stroke: none;
   fill: black;
 }
-// We want the area node to be above the sub-flow node if its inside the sub-flow.
-// This will need some more thinking for a general solution.
-.react-flow__node-sub_flow {
-  z-index: -20 !important;
-}
+
 .react-flow__node-area {
   z-index: -10 !important;
 }
+
 .selected .lynxkite-node {
   outline: var(--xy-selection-border, var(--xy-selection-border-default));
   outline-offset: 7.5px;