Spaces:
Sleeping
Sleeping
File size: 3,912 Bytes
ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d ff2cc46 928873d |
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 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
from shapely.geometry import box, Point
from collections import defaultdict, deque
def map_arrows(nodes, arrows):
for node in nodes:
node["shape"] = box(*node["bbox"])
edges = []
for arrow in arrows:
tail_point = Point(arrow["tail"])
head_point = Point(arrow["head"])
source = None
target = None
for node in nodes:
if node["shape"].contains(tail_point):
source = node["id"]
if node["shape"].contains(head_point):
target = node["id"]
label = arrow.get("label", "")
if source and target and source != target:
edges.append((source, target, label))
return edges
def detect_node_type(text):
text_lower = text.lower()
if "start" in text_lower:
return "start"
if "end" in text_lower or "full" in text_lower:
return "end"
if "?" in text or "yes" in text_lower or "no" in text_lower:
return "decision"
return "process"
def build_flowchart_json(nodes, edges):
graph = {}
reverse_links = defaultdict(list)
edge_labels = defaultdict(list)
for node in nodes:
raw_text = node.get("text", "").strip()
node_type = node.get("type") or detect_node_type(raw_text)
graph[node["id"]] = {
"text": raw_text,
"type": node_type,
"next": []
}
for source, target, label in edges:
graph[source]["next"].append(target)
reverse_links[target].append(source)
edge_labels[(source, target)] = label.lower().strip()
start_nodes = [nid for nid in graph if len(reverse_links[nid]) == 0]
flowchart_json = {
"start": start_nodes[0] if start_nodes else None,
"steps": []
}
visited = set()
queue = deque(start_nodes)
while queue:
current = queue.popleft()
if current in visited:
continue
visited.add(current)
info = graph[current]
step = {
"id": current,
"text": info["text"],
"type": info["type"]
}
parents = reverse_links[current]
if len(parents) == 1:
step["parent"] = parents[0]
elif len(parents) > 1:
step["parents"] = parents
next_nodes = info["next"]
if info["type"] == "decision" and len(next_nodes) >= 2:
branches = {}
for target in next_nodes:
label = edge_labels.get((current, target), "")
if "yes" in label:
branches["yes"] = target
elif "no" in label:
branches["no"] = target
else:
branches.setdefault("unknown", []).append(target)
step["branches"] = branches
queue.extend(next_nodes)
elif len(next_nodes) == 1:
step["next"] = next_nodes[0]
queue.append(next_nodes[0])
elif len(next_nodes) > 1:
step["next"] = next_nodes
queue.extend(next_nodes)
flowchart_json["steps"].append(step)
return flowchart_json
if __name__ == "__main__":
nodes = [
{"id": "node1", "bbox": [100, 100, 200, 150], "text": "Start"},
{"id": "node2", "bbox": [300, 100, 400, 150], "text": "Is valid?"},
{"id": "node3", "bbox": [500, 50, 600, 100], "text": "Approve"},
{"id": "node4", "bbox": [500, 150, 600, 200], "text": "Reject"}
]
arrows = [
{"id": "arrow1", "tail": (200, 125), "head": (300, 125), "label": ""},
{"id": "arrow2", "tail": (400, 125), "head": (500, 75), "label": "Yes"},
{"id": "arrow3", "tail": (400, 125), "head": (500, 175), "label": "No"}
]
edges = map_arrows(nodes, arrows)
flowchart_json = build_flowchart_json(nodes, edges)
import json
print(json.dumps(flowchart_json, indent=2)) |