Create graph_models.py

engine_03/flow/graph_models.py

@@ -0,0 +1,64 @@
+from dataclasses import dataclass, field
+from typing import List, Dict, Any, Optional
+
+
+@dataclass
+class Node:
+    id: str
+    label: str = ""
+    type: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class Edge:
+    source: str
+    target: str
+    label: str = ""
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class FlowGraph:
+    nodes: List[Node] = field(default_factory=list)
+    edges: List[Edge] = field(default_factory=list)
+
+    def find_node(self, node_id: str) -> Optional[Node]:
+        for n in self.nodes:
+            if n.id == node_id:
+                return n
+        return None
+
+
+def table_to_flowgraph(nodes_table, edges_table) -> FlowGraph:
+    """
+    Convert Gradio Dataframe tables into a FlowGraph instance.
+    nodes_table: list of [id, label, type]
+    edges_table: list of [source, target, label]
+    """
+    graph = FlowGraph()
+
+    if nodes_table is None:
+        nodes_table = []
+    if edges_table is None:
+        edges_table = []
+
+    for row in nodes_table:
+        if not row:
+            continue
+        row = list(row) + ["", "", ""]
+        nid, label, ntype = row[0], row[1], row[2]
+        if not nid:
+            continue
+        graph.nodes.append(Node(id=str(nid), label=str(label or nid), type=str(ntype or "")))
+
+    for row in edges_table:
+        if not row:
+            continue
+        row = list(row) + ["", "", ""]
+        src, dst, elabel = row[0], row[1], row[2]
+        if not src or not dst:
+            continue
+        graph.edges.append(Edge(source=str(src), target=str(dst), label=str(elabel or "")))
+
+    return graph