|
|
from typing import List, Set |
|
|
|
|
|
from .graph_models import FlowGraph, Node, Edge |
|
|
|
|
|
|
|
|
def _find_start_nodes(flow: FlowGraph) -> List[Node]: |
|
|
targets = {e.target for e in flow.edges} |
|
|
return [n for n in flow.nodes if n.id not in targets] or (flow.nodes[:1] if flow.nodes else []) |
|
|
|
|
|
|
|
|
def simulate_flowgraph(flow: FlowGraph) -> str: |
|
|
if not flow.nodes: |
|
|
return "No nodes defined. Add some nodes first." |
|
|
|
|
|
start_nodes = _find_start_nodes(flow) |
|
|
visited: Set[str] = set() |
|
|
logs: List[str] = [] |
|
|
|
|
|
def dfs(node: Node, depth: int = 0): |
|
|
if node.id in visited: |
|
|
logs.append(f"{' '*depth}- Node `{node.id}` (already visited, skipping to avoid loops)") |
|
|
return |
|
|
visited.add(node.id) |
|
|
|
|
|
type_tag = f" [{node.type}]" if node.type else "" |
|
|
logs.append(f"{' '*depth}- Node `{node.id}`{type_tag}: {node.label}") |
|
|
|
|
|
|
|
|
for edge in flow.edges: |
|
|
if edge.source == node.id: |
|
|
next_node = flow.find_node(edge.target) |
|
|
if next_node: |
|
|
dfs(next_node, depth + 1) |
|
|
|
|
|
for start in start_nodes: |
|
|
dfs(start, 0) |
|
|
|
|
|
if not logs: |
|
|
return "No execution path found. Check your edges." |
|
|
|
|
|
return "### Simulated Flow Execution\n" + "\n".join(logs) |
|
|
|
