Parse flowchart lines into nodes and edges. Only creates nodes from proper shape definitions (not from edge references). Only includes edges where both endpoints are properly defined nodes.
(lines: List[str], metadata: Dict[str, Any])
| 240 | |
| 241 | |
| 242 | def parse_flowchart(lines: List[str], metadata: Dict[str, Any]) -> Tuple[List[GraphNode], List[GraphEdge]]: |
| 243 | """Parse flowchart lines into nodes and edges. |
| 244 | |
| 245 | Only creates nodes from proper shape definitions (not from edge references). |
| 246 | Only includes edges where both endpoints are properly defined nodes. |
| 247 | """ |
| 248 | nodes: Dict[str, GraphNode] = {} |
| 249 | raw_edges: List[Tuple[str, str, Optional[str]]] = [] # (source, target, label) |
| 250 | |
| 251 | # Patterns for node shapes - ORDER MATTERS (more specific first) |
| 252 | # [label] = step, ([label]) = llm, {label} = decision |
| 253 | # Node ID pattern: matches path::function or path::function::line format |
| 254 | # Uses [^\s\[\](){}]+ to match IDs like "main.py::handle", "backend/client.py::call_llm::42" |
| 255 | node_id = r'([^\s\[\](){}]+)' |
| 256 | node_patterns = [ |
| 257 | (node_id + r'\[\[([^\]]+)\]\]', 'step'), # A[[label]] - subroutine |
| 258 | (node_id + r'\(\[([^\]]+)\]\)', 'llm'), # A([label]) - stadium/llm |
| 259 | (node_id + r'\{([^}]+)\}', 'decision'), # A{label} - diamond |
| 260 | (node_id + r'\[([^\]]+)\]', 'step'), # A[label] - rectangle |
| 261 | (node_id + r'\(([^)]+)\)', 'step'), # A(label) - rounded |
| 262 | ] |
| 263 | |
| 264 | # Edge pattern: A --> B, A -->|label| B |
| 265 | # Node IDs can contain path/function/line separators (. / :: -) |
| 266 | # Match ID chars including '-', relying on shape suffix or whitespace to delimit |
| 267 | edge_id = r'[^\s\[\](){}|>]+' |
| 268 | edge_pattern = rf'({edge_id})(?:\[[^\]]*\]|\(\[[^\]]*\]\)|\{{[^}}]*\}}|\([^)]*\))?\s*-->\s*(?:\|([^|]*)\|)?\s*({edge_id})' |
| 269 | |
| 270 | for line in lines: |
| 271 | # First pass: Extract node definitions with shapes |
| 272 | for pattern, node_type in node_patterns: |
| 273 | for match in re.finditer(pattern, line): |
| 274 | node_id = match.group(1) |
| 275 | label = match.group(2).strip() |
| 276 | # Strip any remaining square/curly brackets from label (but NOT parentheses - used in model names) |
| 277 | label = label.strip('[]{}') |
| 278 | |
| 279 | # Create node (first match wins - patterns are ordered specific to general) |
| 280 | if node_id not in nodes: |
| 281 | nodes[node_id] = GraphNode(id=node_id, label=label, type=node_type) |
| 282 | |
| 283 | # Second pass: Extract edges |
| 284 | edge_matches = re.findall(edge_pattern, line) |
| 285 | for match in edge_matches: |
| 286 | source = match[0] |
| 287 | label = match[1] if len(match) > 1 and match[1] else None |
| 288 | target = match[2] if len(match) > 2 else None |
| 289 | |
| 290 | if source and target: |
| 291 | raw_edges.append((source, target, label.strip() if label else None)) |
| 292 | |
| 293 | # Valid node types - normalize anything else to 'step' |
| 294 | VALID_TYPES = {'step', 'llm', 'decision'} |
| 295 | |
| 296 | # Enrich nodes with metadata |
| 297 | for node_id, node in nodes.items(): |
| 298 | if node_id in metadata: |
| 299 | meta = metadata[node_id] |
no test coverage detected