File size: 1,638 Bytes
593b8c0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
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
|