#!/usr/bin/env python3 """ Grid Blueprint Generator Generates Mermaid and ASCII diagrams from Grid plan YAML files. Used by Master Control to visualize mission topology before execution. Usage: python3 generate_blueprint.py plan.yaml python3 generate_blueprint.py plan.yaml --format ascii python3 generate_blueprint.py plan.yaml --format mermaid python3 generate_blueprint.py plan.yaml --output-dir .grid/ python3 generate_blueprint.py plan.yaml --unicode """ import argparse import json import sys from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional # Try to import yaml, provide fallback try: import yaml except ImportError: print("[Blueprint] PyYAML not found. Install with: pip3 install pyyaml") sys.exit(1) # ============================================================================= # Data Classes # ============================================================================= @dataclass class Port: """Node connection port.""" id: str type: str # input, output, bidirectional label: str position: str # top, right, bottom, left index: int = 0 @dataclass class Node: """Blueprint node representing MC, phase, block, agent, or checkpoint.""" id: str type: str # master_control, phase_coordinator, task_group, agent, checkpoint label: str sublabel: str = "" status: str = "pending" parent_id: Optional[str] = None metadata: dict = field(default_factory=dict) ports: list = field(default_factory=list) @dataclass class Edge: """Connection between nodes.""" source: str target: str type: str # control_flow, data_flow, dependency, verification label: str = "" status: str = "inactive" data_type: str = "" @dataclass class Blueprint: """Complete blueprint data structure.""" version: str = "1.0" generated_at: str = "" mission_name: str = "" mode: str = "AUTOPILOT" complexity: str = "MEDIUM" nodes: list = field(default_factory=list) edges: list = field(default_factory=list) layout_direction: str = "TB" spacing_h: int = 200 spacing_v: int = 100 node_width: int = 180 node_height: int = 80 # ============================================================================= # Character Sets # ============================================================================= ASCII_CHARS = { "h_line": "-", "v_line": "|", "h_heavy": "=", "corner_tl": "+", "corner_tr": "+", "corner_bl": "+", "corner_br": "+", "t_down": "+", "t_up": "+", "t_right": "+", "t_left": "+", "cross": "+", "arrow_down": "v", "arrow_right": ">", "arrow_up": "^", "arrow_left": "<", } UNICODE_CHARS = { "h_line": "\u2500", "v_line": "\u2502", "h_heavy": "\u2550", "corner_tl": "\u250C", "corner_tr": "\u2510", "corner_bl": "\u2514", "corner_br": "\u2518", "t_down": "\u252C", "t_up": "\u2534", "t_right": "\u251C", "t_left": "\u2524", "cross": "\u253C", "arrow_down": "\u25BC", "arrow_right": "\u25B6", "arrow_up": "\u25B2", "arrow_left": "\u25C0", } STATUS_ICONS = { "pending": {"ascii": ".", "unicode": "\u25CC"}, "running": {"ascii": "o", "unicode": "\u25C9"}, "complete": {"ascii": "*", "unicode": "\u25CF"}, "failed": {"ascii": "x", "unicode": "\u2715"}, } AGENT_ICONS = { "executor": {"ascii": "#", "unicode": "\u26A1"}, "recognizer": {"ascii": "v", "unicode": "\u2713"}, "scout": {"ascii": "@", "unicode": "\u2605"}, "planner": {"ascii": "^", "unicode": "\u2261"}, "upscaler": {"ascii": "U", "unicode": "\u2191"}, "default": {"ascii": "*", "unicode": "\u25C6"}, } # ============================================================================= # Blueprint Generator Class # ============================================================================= class BlueprintGenerator: """Generates Mermaid and ASCII diagrams from blueprint data.""" def __init__(self, plan_data: dict): """Initialize with plan YAML data.""" self.plan_data = plan_data self.blueprint = self._parse_blueprint() def _parse_blueprint(self) -> Blueprint: """Parse blueprint data from plan YAML.""" bp_data = self.plan_data.get("blueprint", {}) # If no explicit blueprint, generate from phases/blocks if not bp_data.get("nodes"): return self._generate_blueprint_from_plan() # Parse explicit blueprint blueprint = Blueprint( version=bp_data.get("version", "1.0"), generated_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), mission_name=self.plan_data.get("mission", "Mission"), mode=self.plan_data.get("mode", "AUTOPILOT"), complexity=self.plan_data.get("complexity", "MEDIUM"), ) # Parse layout layout = bp_data.get("layout", {}) blueprint.layout_direction = layout.get("direction", "TB") spacing = layout.get("spacing", {}) blueprint.spacing_h = spacing.get("horizontal", 200) blueprint.spacing_v = spacing.get("vertical", 100) blueprint.node_width = layout.get("node_width", 180) blueprint.node_height = layout.get("node_height", 80) # Parse nodes for n in bp_data.get("nodes", []): node = Node( id=n["id"], type=n["type"], label=n["label"], sublabel=n.get("sublabel", ""), status=n.get("status", "pending"), parent_id=n.get("parent_id"), metadata=n.get("metadata", {}), ) blueprint.nodes.append(node) # Parse edges for e in bp_data.get("edges", []): edge = Edge( source=e["source"], target=e["target"], type=e.get("type", "control_flow"), label=e.get("label", ""), status=e.get("status", "inactive"), data_type=e.get("data_type", ""), ) blueprint.edges.append(edge) return blueprint def _generate_blueprint_from_plan(self) -> Blueprint: """Generate blueprint from standard plan structure (phases/blocks).""" blueprint = Blueprint( generated_at=datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), mission_name=self.plan_data.get("mission", self.plan_data.get("cluster", "Mission")), mode=self.plan_data.get("mode", "AUTOPILOT"), complexity=self.plan_data.get("complexity", "MEDIUM"), ) # Add Master Control mc_node = Node( id="mc", type="master_control", label="MASTER CONTROL", sublabel="Orchestration", status="pending", ) blueprint.nodes.append(mc_node) # Parse phases phases = self.plan_data.get("phases", []) for i, phase in enumerate(phases, 1): phase_id = f"phase-{i:02d}" phase_name = phase.get("name", f"Phase {i}") phase_node = Node( id=phase_id, type="phase_coordinator", label=f"PHASE {i:02d}", sublabel=phase_name, status="pending", parent_id="mc", metadata={"wave": i}, ) blueprint.nodes.append(phase_node) # Edge from MC to phase blueprint.edges.append(Edge( source="mc", target=phase_id, type="control_flow", )) # Parse blocks in phase blocks = phase.get("blocks", []) for j, block in enumerate(blocks, 1): block_id = f"block-{i:02d}-{j:02d}" block_name = block.get("name", f"Block {j}") block_node = Node( id=block_id, type="task_group", label=f"BLOCK {j:02d}", sublabel=block_name, status="pending", parent_id=phase_id, metadata={ "wave": block.get("wave", 1), "autonomous": block.get("autonomous", True), }, ) blueprint.nodes.append(block_node) # Edge from phase to block blueprint.edges.append(Edge( source=phase_id, target=block_id, type="control_flow", )) # Add executor agent exec_id = f"exec-{i:02d}-{j:02d}" exec_node = Node( id=exec_id, type="agent", label="EXECUTOR", sublabel=f"Block {j:02d}", status="pending", parent_id=block_id, metadata={"agent_type": "executor"}, ) blueprint.nodes.append(exec_node) # Edge from block to executor blueprint.edges.append(Edge( source=block_id, target=exec_id, type="data_flow", label="PLAN", data_type="plan", )) # Add recognizer agent recog_id = f"recog-{i:02d}-{j:02d}" recog_node = Node( id=recog_id, type="agent", label="RECOGNIZER", sublabel="Verification", status="pending", parent_id=block_id, metadata={"agent_type": "recognizer"}, ) blueprint.nodes.append(recog_node) # Edge from executor to recognizer blueprint.edges.append(Edge( source=exec_id, target=recog_id, type="verification", label="SUMMARY", data_type="summary", )) # Add dependency edges between phases if i > 1: prev_phase_id = f"phase-{i-1:02d}" blueprint.edges.append(Edge( source=prev_phase_id, target=phase_id, type="dependency", )) return blueprint # ========================================================================= # Mermaid Generation # ========================================================================= def generate_mermaid(self) -> str: """Generate Mermaid flowchart diagram.""" lines = [] # Header with dark theme lines.append("%%{init: {'theme': 'dark', 'themeVariables': {") lines.append(" 'primaryColor': '#00ff88',") lines.append(" 'primaryBorderColor': '#00ff88',") lines.append(" 'primaryTextColor': '#e0e0e0',") lines.append(" 'lineColor': '#00ff88',") lines.append(" 'secondaryColor': '#102030',") lines.append(" 'tertiaryColor': '#0a1628',") lines.append(" 'background': '#0a1628'") lines.append("}}}%%") lines.append("") lines.append("flowchart TB") lines.append("") # Styling classes lines.append("%% Styling Classes") lines.append("classDef mc fill:#102030,stroke:#00ccff,stroke-width:3px,color:#e0e0e0") lines.append("classDef phase fill:#0a1628,stroke:#00ff88,stroke-width:2px,stroke-dasharray:5 5,color:#e0e0e0") lines.append("classDef block fill:#0a1628,stroke:#00ff88,stroke-width:2px,color:#e0e0e0") lines.append("classDef agent fill:#0a1628,stroke:#00ff88,stroke-width:1px,color:#e0e0e0") lines.append("classDef checkpoint fill:#0a1628,stroke:#ff6600,stroke-width:2px,stroke-dasharray:3 3,color:#e0e0e0") lines.append("classDef pending opacity:0.6") lines.append("classDef running stroke-width:4px") lines.append("classDef complete fill:#102030") lines.append("classDef failed stroke:#ff3366") lines.append("") # Find master control mc_node = next((n for n in self.blueprint.nodes if n.type == "master_control"), None) if mc_node: lines.append(f'{mc_node.id}["{mc_node.label}
{mc_node.sublabel}"]') lines.append("") # Group nodes by parent for subgraphs phases = [n for n in self.blueprint.nodes if n.type == "phase_coordinator"] for phase in phases: phase_label = f"{phase.label}: {phase.sublabel}" if phase.sublabel else phase.label lines.append(f'subgraph {phase.id}["{phase_label}"]') lines.append("direction TB") # Find blocks in this phase blocks = [n for n in self.blueprint.nodes if n.parent_id == phase.id] for block in blocks: block_label = f"{block.label}: {block.sublabel}" if block.sublabel else block.label lines.append(f' subgraph {block.id}["{block_label}"]') # Find agents in this block agents = [n for n in self.blueprint.nodes if n.parent_id == block.id] for agent in agents: agent_type = agent.metadata.get("agent_type", "default") icon = self._get_mermaid_icon(agent_type) lines.append(f' {agent.id}["{icon} {agent.label}"]') lines.append(" end") lines.append("end") lines.append("") # Generate edges lines.append("%% Edges") for edge in self.blueprint.edges: arrow = self._get_mermaid_arrow(edge.type) label = f"|{edge.label}|" if edge.label else "" lines.append(f"{edge.source} {arrow}{label} {edge.target}") lines.append("") # Apply status classes lines.append("%% Status Classes") for node in self.blueprint.nodes: type_class = node.type.replace("_", "-") lines.append(f"class {node.id} {type_class},{node.status}") return "\n".join(lines) def _get_mermaid_icon(self, agent_type: str) -> str: """Get FontAwesome icon for agent type.""" icons = { "executor": "fa:fa-bolt", "recognizer": "fa:fa-check-circle", "scout": "fa:fa-search", "planner": "fa:fa-sitemap", "upscaler": "fa:fa-arrow-up", } return icons.get(agent_type, "fa:fa-cog") def _get_mermaid_arrow(self, edge_type: str) -> str: """Get Mermaid arrow style for edge type.""" arrows = { "control_flow": "-->", "data_flow": "-->", "dependency": "-.->", "verification": "==>", } return arrows.get(edge_type, "-->") # ========================================================================= # ASCII Generation # ========================================================================= def generate_ascii(self, use_unicode: bool = False, width: int = 80) -> str: """Generate ASCII art diagram.""" chars = UNICODE_CHARS if use_unicode else ASCII_CHARS icon_type = "unicode" if use_unicode else "ascii" lines = [] # Header lines.append(self._box_line("top", width, chars, heavy=True)) lines.append(self._box_text("GRID BLUEPRINT", width, chars, center=True)) lines.append(self._box_text(self.blueprint.mission_name, width, chars, center=True)) lines.append(self._box_text( f"Mode: {self.blueprint.mode} | Complexity: {self.blueprint.complexity}", width, chars, center=True )) lines.append(self._box_line("mid", width, chars, heavy=True)) lines.append(self._box_text("", width, chars)) # Master Control mc_node = next((n for n in self.blueprint.nodes if n.type == "master_control"), None) if mc_node: mc_box = self._create_node_box(mc_node, chars, icon_type, inner_width=24) for line in mc_box: lines.append(self._box_text(line, width, chars, center=True)) lines.append(self._box_text(f" {chars['v_line']}", width, chars, center=True)) # Phases phases = [n for n in self.blueprint.nodes if n.type == "phase_coordinator"] for i, phase in enumerate(phases): # Phase header status_icon = STATUS_ICONS[phase.status][icon_type] phase_box_lines = [] phase_inner_width = 32 phase_box_lines.append( chars["corner_tl"] + chars["h_line"] * (phase_inner_width - 2) + chars["corner_tr"] ) phase_title = f"{status_icon} {phase.label}: {phase.sublabel}"[:phase_inner_width - 4] phase_box_lines.append( chars["v_line"] + f" {phase_title}".ljust(phase_inner_width - 2) + chars["v_line"] ) phase_box_lines.append( chars["t_right"] + chars["h_line"] * (phase_inner_width - 2) + chars["t_left"] ) # Blocks in this phase blocks = [n for n in self.blueprint.nodes if n.parent_id == phase.id] for block in blocks: block_status = STATUS_ICONS[block.status][icon_type] block_text = f" {block_status} {block.label}: {block.sublabel}"[:phase_inner_width - 4] phase_box_lines.append( chars["v_line"] + f"{block_text}".ljust(phase_inner_width - 2) + chars["v_line"] ) # Agents in block agents = [n for n in self.blueprint.nodes if n.parent_id == block.id] for agent in agents: agent_status = STATUS_ICONS[agent.status][icon_type] agent_type = agent.metadata.get("agent_type", "default") agent_icon = AGENT_ICONS.get(agent_type, AGENT_ICONS["default"])[icon_type] agent_text = f" {agent_status} {agent_icon} {agent.label}"[:phase_inner_width - 4] phase_box_lines.append( chars["v_line"] + f"{agent_text}".ljust(phase_inner_width - 2) + chars["v_line"] ) phase_box_lines.append( chars["corner_bl"] + chars["h_line"] * (phase_inner_width - 2) + chars["corner_br"] ) for line in phase_box_lines: lines.append(self._box_text(line, width, chars, center=True)) # Arrow to next phase if i < len(phases) - 1: lines.append(self._box_text(f" {chars['v_line']}", width, chars, center=True)) lines.append(self._box_text(f" {chars['arrow_down']}", width, chars, center=True)) lines.append(self._box_text(f" {chars['v_line']}", width, chars, center=True)) lines.append(self._box_text("", width, chars)) # Progress bar progress = self._calculate_progress() lines.append(self._box_line("mid", width, chars, heavy=True)) bar_width = 40 filled = int((progress["percent"] / 100) * bar_width) if use_unicode: progress_bar = "\u2588" * filled + "\u2591" * (bar_width - filled) else: progress_bar = "#" * filled + "." * (bar_width - filled) lines.append(self._box_text( f" Progress: [{progress_bar}] {progress['percent']}%", width, chars )) lines.append(self._box_text( f" Blocks: {progress['blocks_complete']}/{progress['blocks_total']} | " f"Agents: {progress['agents_complete']}/{progress['agents_total']}", width, chars )) # Legend lines.append(self._box_line("mid", width, chars, heavy=True)) p_icon = STATUS_ICONS["pending"][icon_type] r_icon = STATUS_ICONS["running"][icon_type] c_icon = STATUS_ICONS["complete"][icon_type] f_icon = STATUS_ICONS["failed"][icon_type] lines.append(self._box_text( f" Legend: {p_icon} Pending {r_icon} Running {c_icon} Complete {f_icon} Failed", width, chars )) ex_icon = AGENT_ICONS["executor"][icon_type] rg_icon = AGENT_ICONS["recognizer"][icon_type] sc_icon = AGENT_ICONS["scout"][icon_type] pl_icon = AGENT_ICONS["planner"][icon_type] lines.append(self._box_text( f" Agents: {ex_icon} Executor {rg_icon} Recognizer {sc_icon} Scout {pl_icon} Planner", width, chars )) lines.append(self._box_line("bottom", width, chars, heavy=True)) return "\n".join(lines) def _box_line(self, position: str, width: int, chars: dict, heavy: bool = False) -> str: """Create a horizontal box line.""" h_char = chars["h_heavy"] if heavy else chars["h_line"] if position == "top": return chars["corner_tl"] + h_char * (width - 2) + chars["corner_tr"] elif position == "mid": return chars["t_right"] + h_char * (width - 2) + chars["t_left"] elif position == "bottom": return chars["corner_bl"] + h_char * (width - 2) + chars["corner_br"] else: return h_char * width def _box_text(self, text: str, width: int, chars: dict, center: bool = False) -> str: """Create a text line within a box.""" inner_width = width - 2 if center: padding = max(0, (inner_width - len(text)) // 2) text = " " * padding + text text = text[:inner_width].ljust(inner_width) return chars["v_line"] + text + chars["v_line"] def _create_node_box(self, node: Node, chars: dict, icon_type: str, inner_width: int = 20) -> list: """Create a small box for a node.""" status_icon = STATUS_ICONS[node.status][icon_type] lines = [] lines.append(chars["corner_tl"] + chars["h_line"] * (inner_width - 2) + chars["corner_tr"]) title = f"{status_icon} {node.label}"[:inner_width - 4] lines.append(chars["v_line"] + f" {title}".ljust(inner_width - 2) + chars["v_line"]) if node.sublabel: sublabel = node.sublabel[:inner_width - 4] lines.append(chars["v_line"] + f" {sublabel}".ljust(inner_width - 2) + chars["v_line"]) lines.append(chars["corner_bl"] + chars["h_line"] * (inner_width - 2) + chars["corner_br"]) return lines def _calculate_progress(self) -> dict: """Calculate mission progress statistics.""" blocks = [n for n in self.blueprint.nodes if n.type == "task_group"] agents = [n for n in self.blueprint.nodes if n.type == "agent"] blocks_complete = len([b for b in blocks if b.status == "complete"]) agents_complete = len([a for a in agents if a.status == "complete"]) total_weight = len(blocks) * 2 + len(agents) complete_weight = blocks_complete * 2 + agents_complete percent = int((complete_weight / total_weight) * 100) if total_weight > 0 else 0 return { "blocks_complete": blocks_complete, "blocks_total": len(blocks), "agents_complete": agents_complete, "agents_total": len(agents), "percent": percent, } # ========================================================================= # Output Methods # ========================================================================= def to_json(self) -> str: """Export blueprint as JSON.""" data = { "version": self.blueprint.version, "generated_at": self.blueprint.generated_at, "mission": { "name": self.blueprint.mission_name, "mode": self.blueprint.mode, "complexity": self.blueprint.complexity, }, "nodes": [ { "id": n.id, "type": n.type, "label": n.label, "sublabel": n.sublabel, "status": n.status, "parent_id": n.parent_id, "metadata": n.metadata, } for n in self.blueprint.nodes ], "edges": [ { "source": e.source, "target": e.target, "type": e.type, "label": e.label, "status": e.status, "data_type": e.data_type, } for e in self.blueprint.edges ], "layout": { "direction": self.blueprint.layout_direction, "spacing": { "horizontal": self.blueprint.spacing_h, "vertical": self.blueprint.spacing_v, }, "node_width": self.blueprint.node_width, "node_height": self.blueprint.node_height, }, "progress": self._calculate_progress(), } return json.dumps(data, indent=2) def save_all(self, output_dir: str, use_unicode: bool = False): """Save all output formats to directory.""" output_path = Path(output_dir) output_path.mkdir(parents=True, exist_ok=True) # Save ASCII ascii_output = self.generate_ascii(use_unicode=use_unicode) (output_path / "blueprint.txt").write_text(ascii_output) # Save Mermaid mermaid_output = self.generate_mermaid() (output_path / "blueprint.mmd").write_text(mermaid_output) # Save JSON json_output = self.to_json() (output_path / "blueprint.json").write_text(json_output) print(f"[Blueprint] Saved to {output_path}/") print(f" - blueprint.txt (ASCII diagram)") print(f" - blueprint.mmd (Mermaid diagram)") print(f" - blueprint.json (Blueprint data)") # ============================================================================= # CLI Entry Point # ============================================================================= def main(): """Command-line interface.""" parser = argparse.ArgumentParser( description="Generate Grid Blueprint visualizations from plan YAML" ) parser.add_argument( "plan_file", help="Path to plan YAML file" ) parser.add_argument( "--format", choices=["ascii", "mermaid", "json", "both", "all"], default="both", help="Output format (default: both ascii and mermaid)" ) parser.add_argument( "--output-dir", default=".grid", help="Output directory (default: .grid/)" ) parser.add_argument( "--unicode", action="store_true", help="Use Unicode box-drawing characters" ) parser.add_argument( "--width", type=int, default=80, help="ASCII diagram width (default: 80)" ) args = parser.parse_args() # Load plan file plan_path = Path(args.plan_file) if not plan_path.exists(): print(f"[Blueprint] Error: File not found: {plan_path}") sys.exit(1) # Parse YAML (handle frontmatter) content = plan_path.read_text() if content.startswith("---"): # Extract YAML frontmatter parts = content.split("---", 2) if len(parts) >= 2: yaml_content = parts[1] else: yaml_content = content else: yaml_content = content try: plan_data = yaml.safe_load(yaml_content) except yaml.YAMLError as e: print(f"[Blueprint] Error parsing YAML: {e}") sys.exit(1) if not plan_data: print("[Blueprint] Error: Empty or invalid plan file") sys.exit(1) # Generate blueprint generator = BlueprintGenerator(plan_data) # Output based on format if args.format in ("all", "both"): generator.save_all(args.output_dir, use_unicode=args.unicode) elif args.format == "ascii": output = generator.generate_ascii(use_unicode=args.unicode, width=args.width) print(output) output_path = Path(args.output_dir) output_path.mkdir(parents=True, exist_ok=True) (output_path / "blueprint.txt").write_text(output) elif args.format == "mermaid": output = generator.generate_mermaid() print(output) output_path = Path(args.output_dir) output_path.mkdir(parents=True, exist_ok=True) (output_path / "blueprint.mmd").write_text(output) elif args.format == "json": output = generator.to_json() print(output) output_path = Path(args.output_dir) output_path.mkdir(parents=True, exist_ok=True) (output_path / "blueprint.json").write_text(output) if __name__ == "__main__": main()