From 879ed18f679fc31ed76c03363eb51f20a955b9b4 Mon Sep 17 00:00:00 2001 From: cah Date: Sun, 18 Jan 2026 08:54:31 -0700 Subject: [PATCH] Add comprehensive documentation and comments - Add detailed docstrings and inline comments to all source files - Create README with installation, usage, and Claude recreation guide - Add pyproject.toml for pip installation - Add MIT LICENSE file - Document architecture, design patterns, and MCP server structure Co-Authored-By: Claude Opus 4.5 --- LICENSE | 21 ++ README.md | 265 +++++++++++++++ __init__.py | 68 +++- __main__.py | 18 +- pyproject.toml | 46 +++ server.py | 811 ++++++++++++++++++++++++++++++++++------------ vivado_session.py | 337 ++++++++++++++++--- 7 files changed, 1315 insertions(+), 251 deletions(-) create mode 100644 LICENSE create mode 100644 README.md create mode 100644 pyproject.toml diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..eb6f3fd --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2024 Corey Hahn + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..3a4b63d --- /dev/null +++ b/README.md @@ -0,0 +1,265 @@ +# Vivado MCP Server + +A Model Context Protocol (MCP) server that enables AI assistants like Claude to directly interact with AMD/Xilinx Vivado FPGA development tools. + +## Features + +- **Session Management**: Start/stop persistent Vivado TCL sessions (avoids 30s startup per command) +- **Project Management**: Open/close Vivado projects (.xpr files) +- **Design Flow**: Run synthesis, implementation, and bitstream generation +- **Reports & Analysis**: Get timing summaries, utilization reports, and design analysis +- **Design Queries**: Explore hierarchy, ports, nets, and cells +- **Simulation**: Control Vivado's integrated simulator (xsim) +- **Raw TCL**: Execute arbitrary Vivado TCL commands for advanced operations + +## Requirements + +- Python 3.10+ +- AMD/Xilinx Vivado installed (tested with 2023.2+) +- Vivado must be in your PATH, or specify the full path when starting a session + +## Installation + +### From GitHub + +```bash +git clone https://github.com/coreyhahn/vivado_mcp.git +cd vivado_mcp +pip install -e . +``` + +### Configure Claude Code + +Add to your Claude Code MCP configuration (`~/.claude/claude_desktop_config.json` or project-level `.mcp.json`): + +```json +{ + "mcpServers": { + "vivado": { + "command": "vivado-mcp" + } + } +} +``` + +Or if you want to specify the Python interpreter: + +```json +{ + "mcpServers": { + "vivado": { + "command": "python", + "args": ["-m", "vivado_mcp"] + } + } +} +``` + +## Usage + +Once configured, Claude can interact with Vivado through natural language. Example workflow: + +1. **Start Vivado session**: "Start a Vivado session" +2. **Open project**: "Open my project at /path/to/project.xpr" +3. **Run synthesis**: "Synthesize the design" +4. **Check timing**: "What's the timing summary? Is timing met?" +5. **Check utilization**: "Show me the resource utilization" +6. **Close session**: "Stop the Vivado session" + +## Available Tools + +### Session Management +- `start_session` - Start a persistent Vivado TCL session +- `stop_session` - Stop the Vivado session +- `session_status` - Get session statistics + +### Project Management +- `open_project` - Open a Vivado project (.xpr) +- `close_project` - Close the current project +- `get_project_info` - Get project information (part, directory, etc.) + +### Design Flow +- `run_synthesis` - Run synthesis +- `run_implementation` - Run place and route +- `generate_bitstream` - Generate bitstream + +### Reports & Analysis +- `get_timing_summary` - Get timing summary (WNS, TNS, WHS, THS) +- `get_timing_paths` - Get detailed timing paths for failing/critical paths +- `get_utilization` - Get resource utilization (LUTs, FFs, BRAMs, DSPs) +- `get_clocks` - Get clock information +- `get_messages` - Get synthesis/implementation messages + +### Design Queries +- `get_design_hierarchy` - Get module/instance hierarchy +- `get_ports` - Get top-level ports +- `get_nets` - Search for nets +- `get_cells` - Search for cells/instances + +### Simulation +- `launch_simulation` - Launch behavioral/post-synth/post-impl simulation +- `run_simulation` - Run simulation for specified time +- `restart_simulation` - Restart from time 0 +- `close_simulation` - Close the simulator +- `get_simulation_time` - Get current simulation time +- `get_signal_value` - Get a signal's current value +- `get_signal_values` - Get multiple signal values by pattern +- `add_signals_to_wave` - Add signals to waveform viewer +- `set_simulation_top` - Set the testbench module +- `get_simulation_objects` - List signals in a scope +- `get_scopes` - List hierarchy scopes +- `step_simulation` - Step simulation +- `add_breakpoint` - Add signal breakpoint +- `remove_breakpoints` - Remove all breakpoints + +### Advanced +- `run_tcl` - Execute raw TCL commands +- `generate_full_report` - Generate full reports to file +- `read_report_section` - Read portions of large reports +- `request_feature` - Request new features +- `list_feature_requests` - List submitted requests + +## Architecture + +``` +┌─────────────────┐ MCP Protocol ┌─────────────────┐ +│ Claude Code │◄────(JSON-RPC)────────►│ Vivado MCP │ +│ (AI Client) │ over stdio │ Server │ +└─────────────────┘ └────────┬────────┘ + │ + │ pexpect + │ (TCL commands) + ▼ + ┌─────────────────┐ + │ Vivado Process │ + │ (TCL mode) │ + └─────────────────┘ +``` + +The server maintains a persistent Vivado process in TCL mode. Commands are sent via pexpect and output is captured by waiting for the Vivado prompt. This avoids the ~30 second startup overhead that would occur if Vivado were launched for each command. + +## Recreating This MCP Server with Claude + +This MCP server was created entirely through conversation with Claude. Here's how you can create similar MCP servers: + +### 1. Start with a Clear Goal + +Tell Claude what you want to build: +> "I want to create an MCP server that lets you control Vivado FPGA tools. You should be able to start Vivado, open projects, run synthesis, check timing, etc." + +### 2. Describe the Architecture + +Explain the key technical challenges: +> "Vivado takes 30 seconds to start, so we need a persistent session. Vivado has a TCL interface we can use. We need to parse Vivado's text output into structured data." + +### 3. Iterate on Tools + +Start with basic tools and add more: +1. Session management (start/stop) +2. Project management +3. Design flow commands +4. Reports and queries +5. Simulation control + +### 4. Key Design Patterns Used + +**Singleton Session**: Only one Vivado process runs at a time +```python +_session: Optional[VivadoSession] = None + +def get_session() -> VivadoSession: + global _session + if _session is None: + _session = VivadoSession() + return _session +``` + +**pexpect for Process Management**: Keeps Vivado alive between commands +```python +self.child = pexpect.spawn( + f'{self.vivado_path} -mode tcl -nojournal -nolog', + encoding='utf-8', + timeout=self.timeout +) +self.child.expect('Vivado%', timeout=10) # Wait for prompt +``` + +**Output Parsing**: Convert text reports to structured JSON +```python +def parse_timing_summary(output: str) -> dict: + wns_match = re.search(r"WNS\(ns\)\s*:\s*([-\d.]+)", output) + if wns_match: + result["wns"] = float(wns_match.group(1)) +``` + +**Response Truncation**: Handle large outputs gracefully +```python +def truncate_response(content: str, max_chars: int) -> dict: + if len(content) > max_chars: + return {"content": content[:max_chars], "truncated": True} +``` + +### 5. MCP Server Structure + +Every MCP server needs: + +```python +from mcp.server import Server +from mcp.server.stdio import stdio_server +from mcp.types import Tool, TextContent + +server = Server("your-server-name") + +@server.list_tools() +async def list_tools() -> list[Tool]: + return [Tool(name="...", description="...", inputSchema={...})] + +@server.call_tool() +async def call_tool(name: str, arguments: dict) -> list[TextContent]: + # Handle tool calls + return [TextContent(type="text", text=json.dumps(result))] + +async def main(): + async with stdio_server() as (read_stream, write_stream): + await server.run(read_stream, write_stream, + server.create_initialization_options()) +``` + +### 6. Prompt for Creating Your Own MCP Server + +Use this prompt template with Claude: + +``` +I want to create an MCP server for [YOUR TOOL]. + +Background: +- [Tool] is a [description] that [what it does] +- It has a [CLI/API/etc] interface that accepts [commands/requests] +- Key operations I want to support: [list operations] + +Technical considerations: +- [Startup time, persistent state, output formats, etc.] + +Please help me create an MCP server with: +1. Session/connection management +2. Core operations as tools +3. Proper error handling +4. Structured JSON responses +5. Comprehensive code comments + +Start with the basic structure and we'll iterate from there. +``` + +## Contributing + +Contributions welcome! Please feel free to submit issues and pull requests. + +## License + +MIT License - see LICENSE file for details. + +## Acknowledgments + +- Created with [Claude](https://claude.ai) (Anthropic) +- Uses the [Model Context Protocol](https://modelcontextprotocol.io) specification +- Integrates with [AMD/Xilinx Vivado](https://www.xilinx.com/products/design-tools/vivado.html) diff --git a/__init__.py b/__init__.py index 0951a52..65b143a 100644 --- a/__init__.py +++ b/__init__.py @@ -1,14 +1,76 @@ -"""Vivado MCP Server - Direct integration with AMD/Xilinx Vivado.""" +""" +Vivado MCP Server - Direct integration with AMD/Xilinx Vivado. + +This package provides a Model Context Protocol (MCP) server that allows +AI assistants like Claude to directly interact with AMD/Xilinx Vivado +FPGA development tools. + +Features: + - Session Management: Start/stop persistent Vivado TCL sessions + - Project Management: Open/close Vivado projects (.xpr files) + - Design Flow: Run synthesis, implementation, and bitstream generation + - Reports: Get timing summaries, utilization, and design analysis + - Design Queries: Explore design hierarchy, ports, nets, and cells + - Simulation: Control Vivado's integrated simulator (xsim) + - Raw TCL: Execute arbitrary Vivado TCL commands + +Installation: + pip install -e . + + Or add to your Claude Code MCP configuration: + { + "mcpServers": { + "vivado": { + "command": "python", + "args": ["-m", "vivado_mcp"] + } + } + } + +Usage: + The server is typically launched by an MCP client (like Claude Code). + For manual testing: + + python -m vivado_mcp + +Example workflow (from an AI assistant): + 1. start_session - Launch Vivado + 2. open_project - Open a .xpr file + 3. run_synthesis - Synthesize the design + 4. get_timing_summary - Check if timing is met + 5. get_utilization - Check resource usage + 6. stop_session - Clean up + +Requirements: + - Python 3.10+ + - mcp>=1.0.0 (Model Context Protocol library) + - pexpect (for Vivado process management) + - AMD/Xilinx Vivado installed and in PATH + +Author: Created with Claude (Anthropic) +License: MIT +Version: 0.1.0 +""" import asyncio from .server import main as _async_main +# Package version __version__ = "0.1.0" def main(): - """Entry point for the vivado-mcp console script.""" + """ + Entry point for the vivado-mcp console script. + + This function is called when running: + - vivado-mcp (after pip install) + - python -m vivado_mcp + + It starts the async MCP server event loop. + """ asyncio.run(_async_main()) -__all__ = ["main"] +# Public API - what gets imported with "from vivado_mcp import *" +__all__ = ["main", "__version__"] diff --git a/__main__.py b/__main__.py index 50bb2dc..470244a 100644 --- a/__main__.py +++ b/__main__.py @@ -1,7 +1,23 @@ -"""Entry point for running the Vivado MCP server.""" +""" +Entry point for running the Vivado MCP server as a module. + +This allows running the server with: + python -m vivado_mcp + +Which is equivalent to: + vivado-mcp (after pip install) + +The server communicates via stdin/stdout using the MCP protocol, +so it's typically launched by an MCP client like Claude Code rather +than run directly from the command line. + +For testing, you can run it directly, but you'll need to send +properly formatted JSON-RPC messages to stdin. +""" import asyncio from .server import main +# Run the async main function when this module is executed directly if __name__ == "__main__": asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..50a12a2 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,46 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "vivado-mcp" +version = "0.1.0" +description = "MCP server for AMD/Xilinx Vivado FPGA development - enables AI assistants to control Vivado" +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.10" +authors = [ + {name = "Created with Claude", email = "noreply@anthropic.com"} +] +keywords = ["mcp", "vivado", "fpga", "xilinx", "amd", "claude", "ai"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)", +] +dependencies = [ + "mcp>=1.0.0", + "pexpect>=4.8.0", +] + +[project.scripts] +vivado-mcp = "vivado_mcp:main" + +[project.urls] +Homepage = "https://github.com/coreyhahn/vivado_mcp" +Repository = "https://github.com/coreyhahn/vivado_mcp" +Issues = "https://github.com/coreyhahn/vivado_mcp/issues" + +# Since the repo root is the package itself, we map the package name to "." +[tool.setuptools] +packages = ["vivado_mcp"] +package-dir = {"vivado_mcp" = "."} + +# Include data files +[tool.setuptools.package-data] +vivado_mcp = ["data/*.json"] diff --git a/server.py b/server.py index af5410e..3707dd2 100644 --- a/server.py +++ b/server.py @@ -1,5 +1,48 @@ #!/usr/bin/env python3 -"""Vivado MCP Server - Direct integration with AMD/Xilinx Vivado.""" +""" +Vivado MCP Server - Direct integration with AMD/Xilinx Vivado. + +This module implements a Model Context Protocol (MCP) server that provides +AI assistants (like Claude) with direct access to AMD/Xilinx Vivado FPGA +development tools. It enables: + +- Session management: Start/stop persistent Vivado TCL sessions +- Project management: Open/close Vivado projects (.xpr files) +- Design flow: Run synthesis, implementation, and bitstream generation +- Reports: Get timing, utilization, and other analysis reports +- Design queries: Explore hierarchy, ports, nets, and cells +- Simulation: Control Vivado's behavioral simulator (xsim) +- Raw TCL: Execute arbitrary TCL commands for advanced operations + +Architecture: + The server maintains a singleton VivadoSession that keeps Vivado running + in TCL mode. Commands are sent via pexpect and results are parsed and + returned as structured JSON. This avoids the ~30 second startup time + for each Vivado command. + +MCP Protocol: + The server uses the MCP stdio transport, communicating via stdin/stdout + with JSON-RPC messages. Tools are exposed via the @server.list_tools() + and @server.call_tool() decorators. + +Usage: + # Start the server (typically done by Claude Code or another MCP client) + python -m vivado_mcp + + # Or via the console script (after pip install) + vivado-mcp + +Example workflow (from an AI assistant): + 1. start_session - Start Vivado + 2. open_project - Open your .xpr file + 3. run_synthesis - Synthesize the design + 4. get_timing_summary - Check timing results + 5. get_utilization - Check resource usage + 6. stop_session - Clean up when done + +Author: Created with Claude (Anthropic) +License: MIT +""" import json import os @@ -13,20 +56,46 @@ from mcp.types import Tool, TextContent from .vivado_session import get_session, VivadoSession -# Feature requests storage + +# ============================================================================= +# CONFIGURATION CONSTANTS +# ============================================================================= + +# Feature requests are stored persistently so users can track requested features FEATURE_REQUESTS_FILE = Path(__file__).parent / "data" / "feature_requests.json" -# Report management +# Report management configuration +# Reports are written to temp files when they exceed inline size limits REPORTS_DIR = Path("/tmp/vivado_mcp") -MAX_RESPONSE_CHARS = 50000 # ~50KB limit for inline responses -REPORT_CACHE_HOURS = 1 # Clean up reports older than this -# In-memory cache for report metadata +# Maximum characters to return inline in a response +# Larger reports should use generate_full_report + read_report_section +MAX_RESPONSE_CHARS = 50000 # ~50KB limit for inline responses + +# How long to keep cached report files before cleanup (in hours) +REPORT_CACHE_HOURS = 1 + +# In-memory cache mapping report_id -> metadata (file path, type, etc.) +# This allows quick lookup of previously generated reports _report_cache: dict[str, dict] = {} +# ============================================================================= +# FEATURE REQUEST MANAGEMENT +# ============================================================================= + def load_feature_requests() -> list[dict]: - """Load feature requests from file.""" + """ + Load feature requests from the persistent JSON file. + + Feature requests allow the AI assistant to record when it encounters + limitations or wishes it had a tool that doesn't exist. This helps + guide future development of the MCP server. + + Returns: + List of feature request dictionaries, or empty list if file + doesn't exist or can't be parsed. + """ if FEATURE_REQUESTS_FILE.exists(): try: return json.loads(FEATURE_REQUESTS_FILE.read_text()) @@ -36,26 +105,57 @@ def load_feature_requests() -> list[dict]: def save_feature_request(request: dict) -> None: - """Save a feature request to the file.""" + """ + Save a feature request to the persistent JSON file. + + Args: + request: Dictionary containing the feature request with fields: + - id: Auto-assigned sequential ID + - title: Short description of the feature + - description: Detailed explanation of what's needed + - use_case: The specific task that prompted this request + - priority: low/medium/high + - timestamp: ISO format timestamp + - status: "pending" (could be updated to "implemented" later) + """ requests = load_feature_requests() requests.append(request) + # Ensure the data directory exists FEATURE_REQUESTS_FILE.parent.mkdir(parents=True, exist_ok=True) FEATURE_REQUESTS_FILE.write_text(json.dumps(requests, indent=2)) +# ============================================================================= +# RESPONSE TRUNCATION +# ============================================================================= + def truncate_response(content: str, max_chars: int = MAX_RESPONSE_CHARS) -> dict: """ Truncate response content if it exceeds max_chars. - Returns dict with: - - content: truncated content (if needed) - - truncated: bool indicating if truncation occurred - - total_chars: original content length - - total_lines: original line count + Large Vivado reports can be tens of thousands of lines. Rather than + overwhelming the AI context window, we truncate and provide metadata + about what was cut. The user can then use generate_full_report to + get the complete output to a file. + + Args: + content: The full content string to potentially truncate + max_chars: Maximum characters to return (default: MAX_RESPONSE_CHARS) + + Returns: + Dictionary with: + - content: The (possibly truncated) content + - truncated: Boolean indicating if truncation occurred + - total_chars: Original content length + - total_lines: Original line count + - returned_chars: Characters in truncated content (if truncated) + - returned_lines: Lines in truncated content (if truncated) + - truncation_message: Human-readable message about truncation """ total_chars = len(content) total_lines = content.count('\n') + 1 + # If content fits, return it unchanged if total_chars <= max_chars: return { "content": content, @@ -64,10 +164,14 @@ def truncate_response(content: str, max_chars: int = MAX_RESPONSE_CHARS) -> dict "total_lines": total_lines } - # Truncate at a line boundary if possible + # Truncate to max_chars, but try to end at a line boundary + # This makes the output more readable and avoids cutting mid-line truncated_content = content[:max_chars] last_newline = truncated_content.rfind('\n') - if last_newline > max_chars * 0.8: # Only use newline if we keep >80% of content + + # Only use the newline boundary if we keep >80% of the allowed content + # Otherwise we might lose too much useful data + if last_newline > max_chars * 0.8: truncated_content = truncated_content[:last_newline] truncated_lines = truncated_content.count('\n') + 1 @@ -83,57 +187,121 @@ def truncate_response(content: str, max_chars: int = MAX_RESPONSE_CHARS) -> dict } +# ============================================================================= +# REPORT FILE MANAGEMENT +# ============================================================================= + def ensure_reports_dir() -> Path: - """Ensure the reports directory exists and clean up old reports.""" + """ + Ensure the reports directory exists and clean up old reports. + + This function is called before generating new reports. It: + 1. Creates the reports directory if it doesn't exist + 2. Removes any report files older than REPORT_CACHE_HOURS + 3. Cleans up the in-memory cache for deleted files + + Returns: + Path to the reports directory + """ REPORTS_DIR.mkdir(parents=True, exist_ok=True) - # Clean up old reports (older than REPORT_CACHE_HOURS) + # Calculate cutoff timestamp for old reports cutoff = datetime.now().timestamp() - (REPORT_CACHE_HOURS * 3600) + + # Scan for and remove old report files for report_file in REPORTS_DIR.glob("*.txt"): try: if report_file.stat().st_mtime < cutoff: report_file.unlink() - # Also remove from cache if present + # Also remove from in-memory cache if present report_id = report_file.stem _report_cache.pop(report_id, None) except OSError: - pass + pass # Ignore errors during cleanup return REPORTS_DIR def generate_report_id() -> str: - """Generate a unique report ID.""" + """ + Generate a unique 8-character report ID. + + Uses UUID4 for uniqueness, truncated to 8 chars for readability. + The ID is used to reference reports across tool calls. + + Returns: + 8-character hexadecimal string (e.g., "a1b2c3d4") + """ return str(uuid.uuid4())[:8] def get_hierarchy_depth(path: str) -> int: - """Get the depth of a hierarchical path.""" + """ + Get the depth of a hierarchical path. + + Vivado uses "/" to separate hierarchy levels (e.g., "cpu/alu/adder"). + This function counts the depth to help filter hierarchy queries. + + Args: + path: Hierarchical path string + + Returns: + Depth as integer (0 for top level, 1 for first level children, etc.) + """ return path.count('/') -# Create the MCP server + +# ============================================================================= +# MCP SERVER INSTANCE +# ============================================================================= + +# Create the MCP server instance +# The name "vivado" is used as the server identifier in MCP communications server = Server("vivado") -# ============================================================================ -# Helper functions to parse Vivado output -# ============================================================================ +# ============================================================================= +# VIVADO OUTPUT PARSERS +# ============================================================================= +# These functions parse Vivado's text-based reports into structured data +# that's easier for AI assistants to work with. def parse_timing_summary(output: str) -> dict: - """Parse timing summary report into structured data.""" + """ + Parse a Vivado timing summary report into structured data. + + Timing summary reports contain critical information about whether + the design meets timing requirements. Key metrics: + + - WNS (Worst Negative Slack): Most critical setup timing margin + Positive = timing met, Negative = timing violation + - TNS (Total Negative Slack): Sum of all negative setup slacks + - WHS (Worst Hold Slack): Most critical hold timing margin + - THS (Total Hold Slack): Sum of all negative hold slacks + - WPWS (Worst Pulse Width Slack): For pulse width requirements + - TPWS (Total Pulse Width Slack): Sum of pulse width violations + + Args: + output: Raw text output from report_timing_summary + + Returns: + Dictionary with parsed metrics and "met" boolean indicating + if all timing is met (WNS >= 0 and WHS >= 0) + """ result = { - "wns": None, # Worst Negative Slack - "tns": None, # Total Negative Slack - "whs": None, # Worst Hold Slack - "ths": None, # Total Hold Slack - "wpws": None, # Worst Pulse Width Slack - "tpws": None, # Total Pulse Width Slack + "wns": None, # Worst Negative Slack (setup) + "tns": None, # Total Negative Slack (setup) + "whs": None, # Worst Hold Slack + "ths": None, # Total Hold Slack + "wpws": None, # Worst Pulse Width Slack + "tpws": None, # Total Pulse Width Slack "failing_endpoints": 0, "met": False, - "raw": output + "raw": output # Keep raw output for detailed analysis } - # Parse WNS/TNS + # Parse WNS/TNS (setup timing) using regex + # Format: "WNS(ns) : 1.234" or similar wns_match = re.search(r"WNS\(ns\)\s*:\s*([-\d.]+)", output) tns_match = re.search(r"TNS\(ns\)\s*:\s*([-\d.]+)", output) if wns_match: @@ -141,7 +309,7 @@ def parse_timing_summary(output: str) -> dict: if tns_match: result["tns"] = float(tns_match.group(1)) - # Parse WHS/THS + # Parse WHS/THS (hold timing) whs_match = re.search(r"WHS\(ns\)\s*:\s*([-\d.]+)", output) ths_match = re.search(r"THS\(ns\)\s*:\s*([-\d.]+)", output) if whs_match: @@ -149,12 +317,12 @@ def parse_timing_summary(output: str) -> dict: if ths_match: result["ths"] = float(ths_match.group(1)) - # Parse failing endpoints + # Parse count of failing endpoints fail_match = re.search(r"(\d+)\s+failing\s+endpoint", output, re.IGNORECASE) if fail_match: result["failing_endpoints"] = int(fail_match.group(1)) - # Check if timing is met + # Determine if timing is met: both setup and hold must have non-negative slack if result["wns"] is not None and result["whs"] is not None: result["met"] = result["wns"] >= 0 and result["whs"] >= 0 @@ -162,17 +330,41 @@ def parse_timing_summary(output: str) -> dict: def parse_utilization(output: str) -> dict: - """Parse utilization report into structured data.""" + """ + Parse a Vivado utilization report into structured data. + + Utilization reports show how much of each FPGA resource type is used. + This is critical for understanding if a design will fit and for + optimization decisions. + + Resource types tracked: + - LUT: Look-Up Tables (combinational logic) + - FF: Flip-Flops (registers/sequential logic) + - BRAM: Block RAM (on-chip memory) + - DSP: DSP slices (multipliers, MACs) + - IO: Input/Output pins + + Args: + output: Raw text output from report_utilization + + Returns: + Dictionary with each resource type containing: + - used: Number of resources used + - available: Total resources on the device + - percent: Utilization percentage + """ result = { "lut": {"used": 0, "available": 0, "percent": 0}, "ff": {"used": 0, "available": 0, "percent": 0}, "bram": {"used": 0, "available": 0, "percent": 0}, "dsp": {"used": 0, "available": 0, "percent": 0}, "io": {"used": 0, "available": 0, "percent": 0}, - "raw": output + "raw": output # Keep raw output for detailed analysis } - # Parse different resource types + # Regex patterns for each resource type + # Vivado's table format: "Resource | Used | Fixed | Available | Util%" + # Different device families use slightly different names patterns = { "lut": r"(?:Slice LUTs|CLB LUTs)\s*\|\s*(\d+)\s*\|\s*\d+\s*\|\s*(\d+)\s*\|\s*([\d.]+)", "ff": r"(?:Slice Registers|CLB Registers)\s*\|\s*(\d+)\s*\|\s*\d+\s*\|\s*(\d+)\s*\|\s*([\d.]+)", @@ -181,6 +373,7 @@ def parse_utilization(output: str) -> dict: "io": r"(?:Bonded IOB|Bonded User I/O)\s*\|\s*(\d+)\s*\|\s*\d+\s*\|\s*(\d+)\s*\|\s*([\d.]+)" } + # Apply each pattern and extract values for resource, pattern in patterns.items(): match = re.search(pattern, output, re.IGNORECASE) if match: @@ -192,7 +385,21 @@ def parse_utilization(output: str) -> dict: def parse_messages(output: str) -> dict: - """Parse Vivado messages into categorized lists.""" + """ + Parse Vivado messages into categorized lists. + + Vivado outputs messages with severity prefixes: + - ERROR: Design or tool errors that must be fixed + - CRITICAL WARNING: Serious issues that may cause problems + - WARNING: Potential issues to review + - INFO: Informational messages + + Args: + output: Raw text output containing Vivado messages + + Returns: + Dictionary with lists of messages by category + """ result = { "errors": [], "critical_warnings": [], @@ -201,6 +408,7 @@ def parse_messages(output: str) -> dict: "raw": output } + # Categorize each line by its severity prefix for line in output.split("\n"): line = line.strip() if re.match(r"ERROR:", line): @@ -215,15 +423,41 @@ def parse_messages(output: str) -> dict: return result -# ============================================================================ -# TOOLS -# ============================================================================ +# ============================================================================= +# TOOL DEFINITIONS +# ============================================================================= +# MCP tools are the interface exposed to AI assistants. Each tool has: +# - name: Unique identifier for the tool +# - description: What the tool does (shown to the AI) +# - inputSchema: JSON Schema defining the parameters @server.list_tools() async def list_tools() -> list[Tool]: - """List all available Vivado tools.""" + """ + List all available Vivado tools. + + This function is called by MCP clients to discover available tools. + Tools are organized into categories: + + 1. Session Management: start_session, stop_session, session_status + 2. Project Management: open_project, close_project, get_project_info + 3. Design Flow: run_synthesis, run_implementation, generate_bitstream + 4. Reports/Analysis: get_timing_summary, get_timing_paths, get_utilization, etc. + 5. Design Queries: get_design_hierarchy, get_ports, get_nets, get_cells + 6. Raw TCL: run_tcl for advanced operations + 7. Simulation: launch_simulation, run_simulation, get_signal_value, etc. + 8. Feature Requests: request_feature, list_feature_requests + 9. Report Management: generate_full_report, read_report_section + + Returns: + List of Tool objects with name, description, and inputSchema + """ return [ - # Session management + # ===================================================================== + # SESSION MANAGEMENT TOOLS + # ===================================================================== + # These tools control the Vivado process lifecycle + Tool( name="start_session", description="Start a persistent Vivado TCL session. Must be called before other commands.", @@ -257,7 +491,11 @@ async def list_tools() -> list[Tool]: } ), - # Project management + # ===================================================================== + # PROJECT MANAGEMENT TOOLS + # ===================================================================== + # These tools work with Vivado project files (.xpr) + Tool( name="open_project", description="Open a Vivado project (.xpr file)", @@ -291,7 +529,11 @@ async def list_tools() -> list[Tool]: } ), - # Design flow + # ===================================================================== + # DESIGN FLOW TOOLS + # ===================================================================== + # These tools run the major FPGA design flow steps + Tool( name="run_synthesis", description="Run synthesis on the current project", @@ -330,7 +572,11 @@ async def list_tools() -> list[Tool]: } ), - # Reports and analysis + # ===================================================================== + # REPORTS AND ANALYSIS TOOLS + # ===================================================================== + # These tools generate and parse Vivado's analysis reports + Tool( name="get_timing_summary", description="Get timing summary (WNS, TNS, WHS, THS) - returns structured data", @@ -439,7 +685,11 @@ async def list_tools() -> list[Tool]: } ), - # Design queries + # ===================================================================== + # DESIGN QUERY TOOLS + # ===================================================================== + # These tools explore the elaborated/synthesized design structure + Tool( name="get_design_hierarchy", description="Get the design hierarchy (modules and instances)", @@ -504,7 +754,11 @@ async def list_tools() -> list[Tool]: } ), - # Raw TCL + # ===================================================================== + # RAW TCL TOOL + # ===================================================================== + # Escape hatch for advanced operations not covered by specific tools + Tool( name="run_tcl", description="Execute a raw TCL command in Vivado. Use for advanced operations not covered by other tools.", @@ -520,7 +774,11 @@ async def list_tools() -> list[Tool]: } ), - # Simulation tools + # ===================================================================== + # SIMULATION TOOLS + # ===================================================================== + # These tools control Vivado's integrated simulator (xsim) + Tool( name="launch_simulation", description="Launch behavioral simulation (xsim). Opens the simulator and loads the design.", @@ -739,7 +997,11 @@ async def list_tools() -> list[Tool]: } ), - # Feature requests + # ===================================================================== + # FEATURE REQUEST TOOLS + # ===================================================================== + # Allow AI assistants to request new features + Tool( name="request_feature", description="Request a new feature or capability for the Vivado MCP server. Use this when you encounter a limitation or wish you had a tool that doesn't exist.", @@ -777,7 +1039,11 @@ async def list_tools() -> list[Tool]: } ), - # Report file management + # ===================================================================== + # REPORT FILE MANAGEMENT TOOLS + # ===================================================================== + # Handle large reports that exceed inline response limits + Tool( name="generate_full_report", description="Generate a full Vivado report to a file. Use when inline reports are truncated or you need the complete output.", @@ -834,13 +1100,45 @@ async def list_tools() -> list[Tool]: ] +# ============================================================================= +# TOOL IMPLEMENTATION +# ============================================================================= + @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: - """Handle tool calls.""" + """ + Handle tool calls from MCP clients. + + This is the main dispatcher that routes tool calls to their implementations. + Each tool returns a list containing a single TextContent with JSON-formatted + results. + + Args: + name: The tool name being called + arguments: Dictionary of arguments passed to the tool + + Returns: + List containing one TextContent with JSON response + + Response format: + All tools return JSON with at minimum: + - success: Boolean indicating if the operation succeeded + - Additional fields specific to each tool + + On error: + - error: Error message string + - success: False + """ + # Get the singleton Vivado session session = get_session() - # Session management + # ========================================================================= + # SESSION MANAGEMENT + # ========================================================================= + if name == "start_session": + # Start Vivado TCL session + # This spawns a persistent Vivado process that stays running vivado_path = arguments.get("vivado_path", "vivado") session.vivado_path = vivado_path result = session.start() @@ -851,6 +1149,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "stop_session": + # Stop Vivado session gracefully result = session.stop() return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -858,19 +1157,29 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "session_status": + # Get session statistics (commands run, errors, timing, etc.) stats = session.get_stats() return [TextContent(type="text", text=json.dumps(stats, indent=2))] - # Check session is running for remaining commands + # ========================================================================= + # SESSION CHECK + # ========================================================================= + # All remaining commands require an active Vivado session + if not session.is_running: return [TextContent(type="text", text=json.dumps({ "error": "Vivado session not running. Call start_session first.", "success": False }, indent=2))] - # Project management + # ========================================================================= + # PROJECT MANAGEMENT + # ========================================================================= + if name == "open_project": + # Open a Vivado project file (.xpr) project_path = arguments.get("project_path", "") + # Use braces to handle paths with spaces result = session.run_tcl(f"open_project {{{project_path}}}") if result.success: session.current_project = project_path @@ -881,6 +1190,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "close_project": + # Close the current project result = session.run_tcl("close_project") session.current_project = None return [TextContent(type="text", text=json.dumps({ @@ -889,11 +1199,12 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_project_info": + # Get various project properties commands = [ - "current_project", - "get_property PART [current_project]", - "get_property TARGET_LANGUAGE [current_project]", - "get_property DIRECTORY [current_project]" + "current_project", # Project name + "get_property PART [current_project]", # Target FPGA part + "get_property TARGET_LANGUAGE [current_project]", # Verilog/VHDL + "get_property DIRECTORY [current_project]" # Project directory ] results = {} for cmd in commands: @@ -901,8 +1212,14 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: results[cmd] = r.output return [TextContent(type="text", text=json.dumps(results, indent=2))] - # Design flow + # ========================================================================= + # DESIGN FLOW + # ========================================================================= + elif name == "run_synthesis": + # Run synthesis with optional parallel jobs + # reset_run clears previous results, launch_runs starts synthesis, + # wait_on_run blocks until complete jobs = arguments.get("jobs", 4) result = session.run_tcl(f"reset_run synth_1; launch_runs synth_1 -jobs {jobs}; wait_on_run synth_1") return [TextContent(type="text", text=json.dumps({ @@ -912,6 +1229,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "run_implementation": + # Run place and route jobs = arguments.get("jobs", 4) result = session.run_tcl(f"launch_runs impl_1 -jobs {jobs}; wait_on_run impl_1") return [TextContent(type="text", text=json.dumps({ @@ -921,6 +1239,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "generate_bitstream": + # Generate bitstream (programming file) result = session.run_tcl("launch_runs impl_1 -to_step write_bitstream; wait_on_run impl_1") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -928,126 +1247,29 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "elapsed_ms": result.elapsed_ms }, indent=2))] - # Reports and analysis + # ========================================================================= + # REPORTS AND ANALYSIS + # ========================================================================= + elif name == "get_timing_summary": + # Get timing summary with parsed metrics report_type = arguments.get("report_type", "summary") detail_level = arguments.get("detail_level", "standard") + # Run Vivado timing summary report result = session.run_tcl("report_timing_summary -no_header -return_string") + + # Parse the raw output into structured data parsed = parse_timing_summary(result.output) parsed["success"] = result.success parsed["elapsed_ms"] = result.elapsed_ms - # Control output based on detail_level + # Control output verbosity based on detail_level if detail_level == "summary": - # Remove raw output, keep only parsed metrics + # Only return parsed metrics, no raw output parsed.pop("raw", None) elif detail_level == "standard": - # Truncate raw output if too large - if "raw" in parsed and len(parsed["raw"]) > MAX_RESPONSE_CHARS // 2: - truncated = truncate_response(parsed["raw"], MAX_RESPONSE_CHARS // 2) - parsed["raw"] = truncated["content"] - if truncated["truncated"]: - parsed["raw_truncated"] = True - parsed["raw_total_chars"] = truncated["total_chars"] - # detail_level == "full": keep complete raw output (but still apply safety truncation) - elif detail_level == "full": - if "raw" in parsed: - truncated = truncate_response(parsed["raw"], MAX_RESPONSE_CHARS) - parsed["raw"] = truncated["content"] - if truncated["truncated"]: - parsed["raw_truncated"] = True - parsed["raw_total_chars"] = truncated["total_chars"] - parsed["truncation_message"] = truncated["truncation_message"] - - return [TextContent(type="text", text=json.dumps(parsed, indent=2))] - - elif name == "get_timing_paths": - num_paths = arguments.get("num_paths", 10) - slack_threshold = arguments.get("slack_threshold", 0) - path_type = arguments.get("path_type", "setup") - from_pin = arguments.get("from_pin") - to_pin = arguments.get("to_pin") - through = arguments.get("through") - clock = arguments.get("clock") - - delay_type = "max" if path_type == "setup" else "min" - cmd = f"report_timing -delay_type {delay_type} -max_paths {num_paths} -slack_lesser_than {slack_threshold}" - - # Add optional filters - if from_pin: - cmd += f" -from {{{from_pin}}}" - if to_pin: - cmd += f" -to {{{to_pin}}}" - if through: - cmd += f" -through {{{through}}}" - if clock: - cmd += f" -filter {{CLOCK == {clock}}}" - - cmd += " -return_string" - result = session.run_tcl(cmd) - - # Apply truncation for large outputs - response = { - "success": result.success, - "elapsed_ms": result.elapsed_ms, - "filters_applied": { - "path_type": path_type, - "num_paths": num_paths, - "slack_threshold": slack_threshold - } - } - - if from_pin: - response["filters_applied"]["from_pin"] = from_pin - if to_pin: - response["filters_applied"]["to_pin"] = to_pin - if through: - response["filters_applied"]["through"] = through - if clock: - response["filters_applied"]["clock"] = clock - - if result.success: - truncated = truncate_response(result.output, MAX_RESPONSE_CHARS) - response["paths"] = truncated["content"] - if truncated["truncated"]: - response["truncated"] = True - response["total_chars"] = truncated["total_chars"] - response["truncation_message"] = truncated["truncation_message"] - else: - response["paths"] = result.output - - return [TextContent(type="text", text=json.dumps(response, indent=2))] - - elif name == "get_utilization": - hierarchical = arguments.get("hierarchical", False) - detail_level = arguments.get("detail_level", "standard") - module_filter = arguments.get("module_filter") - threshold_percent = arguments.get("threshold_percent") - - cmd = "report_utilization -return_string" - if hierarchical: - cmd += " -hierarchical" - if module_filter: - cmd += f" -hierarchical_pattern {{{module_filter}}}" - - result = session.run_tcl(cmd) - parsed = parse_utilization(result.output) - parsed["success"] = result.success - parsed["elapsed_ms"] = result.elapsed_ms - - # Apply threshold filter if specified - if threshold_percent is not None: - for resource in ["lut", "ff", "bram", "dsp", "io"]: - if resource in parsed and parsed[resource]["percent"] < threshold_percent: - parsed[resource]["below_threshold"] = True - - # Control output based on detail_level - if detail_level == "summary": - # Remove raw output, keep only parsed metrics - parsed.pop("raw", None) - elif detail_level == "standard": - # Truncate raw output if too large + # Truncate raw output if too large (half of max to leave room for other data) if "raw" in parsed and len(parsed["raw"]) > MAX_RESPONSE_CHARS // 2: truncated = truncate_response(parsed["raw"], MAX_RESPONSE_CHARS // 2) parsed["raw"] = truncated["content"] @@ -1066,7 +1288,118 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: return [TextContent(type="text", text=json.dumps(parsed, indent=2))] + elif name == "get_timing_paths": + # Get detailed timing path information + # Useful for debugging timing violations + num_paths = arguments.get("num_paths", 10) + slack_threshold = arguments.get("slack_threshold", 0) # 0 = failing paths only + path_type = arguments.get("path_type", "setup") + from_pin = arguments.get("from_pin") + to_pin = arguments.get("to_pin") + through = arguments.get("through") + clock = arguments.get("clock") + + # Build the report_timing command + delay_type = "max" if path_type == "setup" else "min" + cmd = f"report_timing -delay_type {delay_type} -max_paths {num_paths} -slack_lesser_than {slack_threshold}" + + # Add optional path filters + if from_pin: + cmd += f" -from {{{from_pin}}}" + if to_pin: + cmd += f" -to {{{to_pin}}}" + if through: + cmd += f" -through {{{through}}}" + if clock: + cmd += f" -filter {{CLOCK == {clock}}}" + + cmd += " -return_string" + result = session.run_tcl(cmd) + + # Build response with filter information + response = { + "success": result.success, + "elapsed_ms": result.elapsed_ms, + "filters_applied": { + "path_type": path_type, + "num_paths": num_paths, + "slack_threshold": slack_threshold + } + } + + # Include any filters that were used + if from_pin: + response["filters_applied"]["from_pin"] = from_pin + if to_pin: + response["filters_applied"]["to_pin"] = to_pin + if through: + response["filters_applied"]["through"] = through + if clock: + response["filters_applied"]["clock"] = clock + + # Handle potentially large output + if result.success: + truncated = truncate_response(result.output, MAX_RESPONSE_CHARS) + response["paths"] = truncated["content"] + if truncated["truncated"]: + response["truncated"] = True + response["total_chars"] = truncated["total_chars"] + response["truncation_message"] = truncated["truncation_message"] + else: + response["paths"] = result.output + + return [TextContent(type="text", text=json.dumps(response, indent=2))] + + elif name == "get_utilization": + # Get resource utilization with parsed metrics + hierarchical = arguments.get("hierarchical", False) + detail_level = arguments.get("detail_level", "standard") + module_filter = arguments.get("module_filter") + threshold_percent = arguments.get("threshold_percent") + + # Build utilization report command + cmd = "report_utilization -return_string" + if hierarchical: + cmd += " -hierarchical" + if module_filter: + cmd += f" -hierarchical_pattern {{{module_filter}}}" + + result = session.run_tcl(cmd) + + # Parse into structured data + parsed = parse_utilization(result.output) + parsed["success"] = result.success + parsed["elapsed_ms"] = result.elapsed_ms + + # Apply threshold filter if specified + if threshold_percent is not None: + for resource in ["lut", "ff", "bram", "dsp", "io"]: + if resource in parsed and parsed[resource]["percent"] < threshold_percent: + parsed[resource]["below_threshold"] = True + + # Control output verbosity + if detail_level == "summary": + parsed.pop("raw", None) + elif detail_level == "standard": + if "raw" in parsed and len(parsed["raw"]) > MAX_RESPONSE_CHARS // 2: + truncated = truncate_response(parsed["raw"], MAX_RESPONSE_CHARS // 2) + parsed["raw"] = truncated["content"] + if truncated["truncated"]: + parsed["raw_truncated"] = True + parsed["raw_total_chars"] = truncated["total_chars"] + elif detail_level == "full": + if "raw" in parsed: + truncated = truncate_response(parsed["raw"], MAX_RESPONSE_CHARS) + parsed["raw"] = truncated["content"] + if truncated["truncated"]: + parsed["raw_truncated"] = True + parsed["raw_total_chars"] = truncated["total_chars"] + parsed["truncation_message"] = truncated["truncation_message"] + + return [TextContent(type="text", text=json.dumps(parsed, indent=2))] + elif name == "get_clocks": + # Get clock information from the design result = session.run_tcl("report_clocks -return_string") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1075,10 +1408,12 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_messages": + # Get Vivado messages filtered by severity severity = arguments.get("severity", "all") - # Get messages from Vivado's message log result = session.run_tcl("get_msg_config -rules") parsed = parse_messages(result.output) + + # Apply severity filter if severity != "all": filtered = { "error": parsed["errors"], @@ -1089,26 +1424,30 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: parsed["success"] = result.success return [TextContent(type="text", text=json.dumps(parsed, indent=2))] - # Design queries + # ========================================================================= + # DESIGN QUERIES + # ========================================================================= + elif name == "get_design_hierarchy": + # Get the design hierarchy (instances and modules) max_depth = arguments.get("max_depth", 3) instance_pattern = arguments.get("instance_pattern", "*") - # Get hierarchical cells with pattern filter + # Get all hierarchical cells matching the pattern cmd = f"get_cells -hierarchical {{{instance_pattern}}}" result = session.run_tcl(cmd) if result.success and result.output.strip(): cells = result.output.strip().split() - # Filter by depth: count '/' separators + # Filter by hierarchy depth (count '/' separators) filtered_cells = [] for cell in cells: depth = get_hierarchy_depth(cell) if depth <= max_depth: filtered_cells.append(cell) - # Build hierarchical structure + # Build a hierarchical structure for easier visualization hierarchy = {} for cell in sorted(filtered_cells): parts = cell.split('/') @@ -1118,9 +1457,9 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: current[part] = {"_children": {}} current = current[part]["_children"] - # Also get module reference for each cell (limited to avoid large output) + # Get module reference for each cell (limited for performance) cell_refs = {} - sample_cells = filtered_cells[:100] # Limit for performance + sample_cells = filtered_cells[:100] for cell in sample_cells: ref_result = session.run_tcl(f"get_property REF_NAME [get_cells {{{cell}}}]") if ref_result.success and ref_result.output.strip(): @@ -1128,7 +1467,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: response = { "success": True, - "cells": filtered_cells[:500], # Limit response size + "cells": filtered_cells[:500], # Limit for response size "cell_count": len(filtered_cells), "cell_modules": cell_refs, "max_depth": max_depth, @@ -1151,6 +1490,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: return [TextContent(type="text", text=json.dumps(response, indent=2))] elif name == "get_ports": + # Get top-level I/O ports result = session.run_tcl("get_ports *") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1159,8 +1499,10 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_nets": + # Search for nets by pattern pattern = arguments.get("pattern", "*") limit = arguments.get("limit", 100) + # Use lrange to limit results result = session.run_tcl(f"lrange [get_nets {{{pattern}}}] 0 {limit-1}") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1169,6 +1511,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_cells": + # Search for cells/instances by pattern pattern = arguments.get("pattern", "*") limit = arguments.get("limit", 100) result = session.run_tcl(f"lrange [get_cells {{{pattern}}}] 0 {limit-1}") @@ -1178,8 +1521,12 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "elapsed_ms": result.elapsed_ms }, indent=2))] - # Raw TCL + # ========================================================================= + # RAW TCL + # ========================================================================= + elif name == "run_tcl": + # Execute arbitrary TCL command (escape hatch for advanced users) command = arguments.get("command", "") result = session.run_tcl(command) return [TextContent(type="text", text=json.dumps({ @@ -1188,15 +1535,21 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "elapsed_ms": result.elapsed_ms }, indent=2))] - # Simulation tools + # ========================================================================= + # SIMULATION TOOLS + # ========================================================================= + elif name == "launch_simulation": + # Launch Vivado's integrated simulator (xsim) mode = arguments.get("mode", "behavioral") + + # Map friendly names to Vivado's mode strings mode_map = { - "behavioral": "behav", - "post_synth_func": "synth -type func", - "post_synth_timing": "synth -type timing", - "post_impl_func": "impl -type func", - "post_impl_timing": "impl -type timing" + "behavioral": "behav", # RTL simulation + "post_synth_func": "synth -type func", # Post-synthesis functional + "post_synth_timing": "synth -type timing", # Post-synthesis with timing + "post_impl_func": "impl -type func", # Post-implementation functional + "post_impl_timing": "impl -type timing" # Post-implementation with timing } sim_mode = mode_map.get(mode, "behav") result = session.run_tcl(f"launch_simulation -mode {sim_mode}") @@ -1207,8 +1560,10 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "run_simulation": + # Advance simulation time time_val = arguments.get("time", "100ns") if time_val.lower() == "all": + # Run until all events processed (testbench completes) result = session.run_tcl("run -all") else: result = session.run_tcl(f"run {time_val}") @@ -1219,6 +1574,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "restart_simulation": + # Reset simulation to time 0 result = session.run_tcl("restart") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1227,6 +1583,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "close_simulation": + # Close the simulator result = session.run_tcl("close_sim") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1235,6 +1592,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_simulation_time": + # Get current simulation time result = session.run_tcl("current_time") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1243,6 +1601,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_signal_value": + # Get current value of a single signal signal = arguments.get("signal", "") radix = arguments.get("radix", "hex") result = session.run_tcl(f"get_value -radix {radix} {{{signal}}}") @@ -1255,14 +1614,17 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_signal_values": + # Get values of multiple signals matching a pattern pattern = arguments.get("pattern", "/*") radix = arguments.get("radix", "hex") - # Get list of signals matching pattern + + # First get list of signals matching pattern signals_result = session.run_tcl(f"get_objects -filter {{TYPE == signal || TYPE == port}} {{{pattern}}}") if signals_result.success and signals_result.output.strip(): signals = signals_result.output.strip().split() values = {} - for sig in signals[:50]: # Limit to 50 signals + # Limit to 50 signals to avoid overwhelming response + for sig in signals[:50]: val_result = session.run_tcl(f"get_value -radix {radix} {{{sig}}}") if val_result.success: values[sig] = val_result.output.strip() @@ -1279,6 +1641,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "add_signals_to_wave": + # Add signals to waveform viewer signals = arguments.get("signals", []) if isinstance(signals, str): signals = [signals] @@ -1292,6 +1655,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "set_simulation_top": + # Set the top-level testbench module top_module = arguments.get("top_module", "") fileset = arguments.get("fileset", "sim_1") result = session.run_tcl(f"set_property top {top_module} [get_filesets {fileset}]") @@ -1302,9 +1666,11 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_simulation_objects": + # List simulation objects (signals, ports, variables) in a scope scope = arguments.get("scope", "/") obj_filter = arguments.get("filter", "all") + # Map filter names to Vivado filter expressions filter_map = { "all": "", "signals": "-filter {TYPE == signal}", @@ -1323,6 +1689,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_scopes": + # List child scopes (hierarchy levels) in simulation parent = arguments.get("parent", "/") result = session.run_tcl(f"get_scopes {{{parent}/*}}") scopes = result.output.strip().split() if result.success and result.output.strip() else [] @@ -1335,6 +1702,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "step_simulation": + # Step simulation by delta cycles count = arguments.get("count", 1) result = session.run_tcl(f"step {count}") return [TextContent(type="text", text=json.dumps({ @@ -1344,12 +1712,15 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "add_breakpoint": + # Add a breakpoint on signal edge or change signal = arguments.get("signal", "") condition = arguments.get("condition", "change") + + # Map condition names to Vivado flags cond_map = { - "posedge": "-posedge", - "negedge": "-negedge", - "change": "" + "posedge": "-posedge", # Rising edge + "negedge": "-negedge", # Falling edge + "change": "" # Any change } cond_str = cond_map.get(condition, "") result = session.run_tcl(f"add_bp {cond_str} {{{signal}}}") @@ -1362,6 +1733,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "remove_breakpoints": + # Remove all breakpoints result = session.run_tcl("remove_bps -all") return [TextContent(type="text", text=json.dumps({ "success": result.success, @@ -1370,6 +1742,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "get_simulation_messages": + # Get simulation log messages severity = arguments.get("severity", "all") if severity == "all": result = session.run_tcl("get_msg_config -count") @@ -1381,8 +1754,12 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "elapsed_ms": result.elapsed_ms }, indent=2))] - # Feature requests + # ========================================================================= + # FEATURE REQUESTS + # ========================================================================= + elif name == "request_feature": + # Submit a feature request for future development title = arguments.get("title", "") description = arguments.get("description", "") use_case = arguments.get("use_case", "") @@ -1406,6 +1783,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "list_feature_requests": + # List all submitted feature requests requests = load_feature_requests() return [TextContent(type="text", text=json.dumps({ "success": True, @@ -1413,8 +1791,12 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "requests": requests }, indent=2))] - # Report file management tools + # ========================================================================= + # REPORT FILE MANAGEMENT + # ========================================================================= + elif name == "generate_full_report": + # Generate a complete report to a file (for large reports) report_type = arguments.get("report_type", "timing") options = arguments.get("options", {}) output_file = arguments.get("output_file") @@ -1422,14 +1804,14 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: # Ensure reports directory exists and clean up old files ensure_reports_dir() - # Generate report ID and file path + # Generate unique report ID and file path report_id = generate_report_id() if output_file: file_path = Path(output_file) else: file_path = REPORTS_DIR / f"{report_type}_{report_id}.txt" - # Build the report command based on type + # Map report types to Vivado commands report_commands = { "timing": "report_timing -max_paths 100", "timing_summary": "report_timing_summary", @@ -1437,28 +1819,28 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "hierarchy": "report_hierarchy", "clocks": "report_clocks", "power": "report_power", - "drc": "report_drc" + "drc": "report_drc" # Design Rule Check } base_cmd = report_commands.get(report_type, f"report_{report_type}") - # Add options for specific report types + # Apply report-specific options if report_type == "utilization" and options.get("hierarchical"): base_cmd += " -hierarchical" if report_type == "timing" and options.get("num_paths"): base_cmd = base_cmd.replace("-max_paths 100", f"-max_paths {options['num_paths']}") - # Use -file option to write directly to file + # Write directly to file using Vivado's -file option cmd = f"{base_cmd} -file {{{file_path}}}" result = session.run_tcl(cmd) if result.success: - # Get file info try: + # Get file statistics file_stat = file_path.stat() line_count = sum(1 for _ in open(file_path)) - # Store in cache + # Cache report metadata for later lookup _report_cache[report_id] = { "file_path": str(file_path), "report_type": report_type, @@ -1492,18 +1874,19 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: }, indent=2))] elif name == "read_report_section": + # Read a portion of a previously generated report report_id = arguments.get("report_id") file_path = arguments.get("file_path") start_line = arguments.get("start_line", 1) num_lines = arguments.get("num_lines", 100) search_pattern = arguments.get("search_pattern") - # Resolve file path + # Resolve file path from report_id if provided if report_id: if report_id in _report_cache: file_path = _report_cache[report_id]["file_path"] else: - # Try to find file in reports directory + # Try to find file in reports directory by ID possible_files = list(REPORTS_DIR.glob(f"*_{report_id}.txt")) if possible_files: file_path = str(possible_files[0]) @@ -1527,12 +1910,13 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "error": f"File not found: {file_path}" }, indent=2))] + # Read all lines from file with open(file_path, 'r') as f: all_lines = f.readlines() total_lines = len(all_lines) - # Handle search pattern + # Handle search pattern - find and return context around match if search_pattern: pattern = re.compile(search_pattern, re.IGNORECASE) for i, line in enumerate(all_lines): @@ -1550,7 +1934,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "file_path": str(file_path) }, indent=2))] - # Extract requested lines (1-indexed) + # Extract requested line range (1-indexed to 0-indexed) start_idx = max(0, start_line - 1) end_idx = min(total_lines, start_idx + num_lines) selected_lines = all_lines[start_idx:end_idx] @@ -1573,11 +1957,27 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]: "error": f"Error reading file: {e}" }, indent=2))] + # ========================================================================= + # UNKNOWN TOOL + # ========================================================================= + return [TextContent(type="text", text=json.dumps({"error": f"Unknown tool: {name}"}, indent=2))] +# ============================================================================= +# SERVER ENTRY POINT +# ============================================================================= + async def main(): - """Run the MCP server.""" + """ + Run the MCP server. + + This function starts the MCP server using stdio transport (stdin/stdout). + It's designed to be launched by an MCP client like Claude Code. + + The server runs until the client closes the connection or sends an + exit signal. + """ async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, @@ -1586,6 +1986,7 @@ async def main(): ) +# Allow running directly with: python server.py if __name__ == "__main__": import asyncio asyncio.run(main()) diff --git a/vivado_session.py b/vivado_session.py index 93823e1..d28c549 100644 --- a/vivado_session.py +++ b/vivado_session.py @@ -1,4 +1,51 @@ -"""Vivado TCL session manager - maintains persistent Vivado process using pexpect.""" +""" +Vivado TCL Session Manager - Maintains a persistent Vivado process using pexpect. + +This module provides the core Vivado interaction layer for the MCP server. +It manages a persistent Vivado TCL session, avoiding the ~30 second startup +overhead that would occur if Vivado were launched for each command. + +Architecture: + The VivadoSession class spawns Vivado in TCL mode (-mode tcl) using pexpect. + Commands are sent via sendline() and output is captured by waiting for the + Vivado prompt (Vivado%). The session stays alive between commands, maintaining + state (open projects, synthesized designs, etc.). + +Key Design Decisions: + 1. Singleton Pattern: A global _session instance is used to ensure only one + Vivado process runs at a time. Use get_session() to access it. + + 2. Thread Safety: A threading lock protects command execution to prevent + interleaved commands if multiple async tasks try to use Vivado. + + 3. Prompt-Based Parsing: We wait for "Vivado%" prompt to know when a command + completes. Output between command send and prompt is captured. + + 4. Error Detection: Success/failure is determined by checking for error + keywords in the output (ERROR:, invalid command, etc.). + + 5. Statistics Tracking: Command count, timing, and error counts are tracked + for debugging and performance analysis. + +Usage: + from vivado_session import get_session + + session = get_session() + session.start() # Launch Vivado + + result = session.run_tcl("open_project /path/to/project.xpr") + if result.success: + print(f"Project opened in {result.elapsed_ms}ms") + + session.stop() # Clean shutdown + +Dependencies: + - pexpect: For spawning and interacting with Vivado process + - Vivado: Must be installed and in PATH (or specify path explicitly) + +Author: Created with Claude (Anthropic) +License: MIT +""" import pexpect import time @@ -9,9 +56,32 @@ from datetime import datetime import threading +# ============================================================================= +# DATA CLASSES +# ============================================================================= + @dataclass class CommandResult: - """Result from a Vivado TCL command.""" + """ + Result from executing a Vivado TCL command. + + This dataclass encapsulates all information about a command execution, + making it easy to check success, access output, and measure performance. + + Attributes: + command: The TCL command that was executed + output: The captured output from Vivado (excluding prompts) + return_value: "0" for success, "1" for failure (string for JSON compat) + success: Boolean indicating if the command succeeded + elapsed_ms: Time taken to execute the command in milliseconds + timestamp: ISO format timestamp of when the command completed + + Example: + result = session.run_tcl("get_property PART [current_project]") + if result.success: + print(f"Target part: {result.output}") + print(f"Took {result.elapsed_ms:.1f}ms") + """ command: str output: str return_value: str @@ -20,36 +90,96 @@ class CommandResult: timestamp: str = field(default_factory=lambda: datetime.now().isoformat()) +# ============================================================================= +# VIVADO SESSION CLASS +# ============================================================================= + class VivadoSession: """ Manages a persistent Vivado TCL session using pexpect. Vivado is started once and kept running. Commands are sent and output - is captured using pexpect's expect/sendline interface. + is captured using pexpect's expect/sendline interface. This avoids the + ~30 second startup time that would be incurred for each command. + + The session maintains state between commands, so you can open a project, + run synthesis, and then query results - all using the same Vivado instance. + + Attributes: + vivado_path: Path to the Vivado executable + timeout: Maximum time to wait for command completion (seconds) + child: The pexpect spawn object (Vivado process) + is_running: Whether Vivado is currently running + current_project: Path to currently open project (if any) + stats: Dictionary of session statistics + + Thread Safety: + A lock (_lock) protects command execution. Multiple threads can + safely call run_tcl(), though commands will be serialized. + + Example: + with VivadoSession() as session: + session.run_tcl("open_project /path/to/project.xpr") + result = session.run_tcl("report_timing_summary -return_string") + print(result.output) + # Vivado is automatically stopped when exiting the context """ - # Unique marker to detect end of command output (use something that won't appear in normal output) + # Unique marker that won't appear in normal Vivado output + # Used internally for sentinel-based parsing (not currently used but reserved) SENTINEL = "XYZZY_MCP_9f8e7d6c_DONE" def __init__(self, vivado_path: str = "vivado", timeout: float = 300.0): + """ + Initialize the Vivado session manager. + + Args: + vivado_path: Path to Vivado executable. Defaults to "vivado" which + assumes it's in the system PATH. Can be an absolute path + like "/tools/Xilinx/Vivado/2023.2/bin/vivado". + timeout: Maximum time in seconds to wait for any command to complete. + Defaults to 300s (5 minutes) to handle long operations like + synthesis and implementation. + """ self.vivado_path = vivado_path self.timeout = timeout self.child: Optional[pexpect.spawn] = None self.is_running = False self.current_project: Optional[str] = None + + # Thread lock for command execution + # Ensures only one command runs at a time even with async callers self._lock = threading.Lock() - # Statistics + # Statistics tracking for debugging and performance analysis self.stats = { - "session_start": None, - "commands_run": 0, - "total_command_time_ms": 0, - "errors": 0, - "command_history": [] + "session_start": None, # ISO timestamp when session started + "commands_run": 0, # Total commands executed + "total_command_time_ms": 0, # Sum of all command times + "errors": 0, # Count of failed commands + "command_history": [] # Last 100 commands (for debugging) } def start(self) -> CommandResult: - """Start the Vivado TCL session.""" + """ + Start the Vivado TCL session. + + This spawns a new Vivado process in TCL mode with: + - No journal file (-nojournal): Avoids cluttering directory + - No log file (-nolog): Output goes to pexpect instead + + The function waits for Vivado's startup banner ("Start of session") + and then confirms readiness by waiting for the "Vivado%" prompt. + + Returns: + CommandResult with success=True if Vivado started successfully, + or success=False with error message if startup failed. + + Note: + If already running, returns success immediately without restarting. + Vivado startup typically takes 20-30 seconds. + """ + # Don't restart if already running if self.is_running: return CommandResult( command="start", @@ -62,30 +192,36 @@ class VivadoSession: start_time = time.time() try: - # Start Vivado with pexpect + # Spawn Vivado in TCL mode + # -mode tcl: Interactive TCL shell (no GUI) + # -nojournal: Don't create vivado.jou files + # -nolog: Don't create vivado.log files self.child = pexpect.spawn( f'{self.vivado_path} -mode tcl -nojournal -nolog', encoding='utf-8', timeout=self.timeout, - echo=False # Don't echo commands back + echo=False # Don't echo commands back to us ) - # Wait for Vivado to start (look for startup banner) + # Wait for Vivado to display its startup banner + # This indicates Vivado has loaded and is ready to accept commands self.child.expect('Start of session', timeout=120) - # Give it a moment to fully initialize + # Brief pause to let Vivado fully initialize time.sleep(1) - # Drain any remaining startup output + # Drain any remaining startup output to clear the buffer try: self.child.read_nonblocking(size=100000, timeout=1) except (pexpect.TIMEOUT, pexpect.EOF): - pass + pass # Expected - no more data to read - # Wait for prompt to confirm ready - self.child.sendline("") # Empty command to get prompt + # Send empty command to confirm we get a prompt back + # This validates that Vivado is responsive + self.child.sendline("") self.child.expect('Vivado%', timeout=10) + # Mark session as running and record start time self.is_running = True self.stats["session_start"] = datetime.now().isoformat() @@ -100,6 +236,7 @@ class VivadoSession: ) except pexpect.TIMEOUT: + # Vivado didn't respond in time self.is_running = False elapsed = (time.time() - start_time) * 1000 return CommandResult( @@ -110,6 +247,7 @@ class VivadoSession: elapsed_ms=elapsed ) except Exception as e: + # Other errors (file not found, permissions, etc.) self.is_running = False elapsed = (time.time() - start_time) * 1000 return CommandResult( @@ -124,12 +262,39 @@ class VivadoSession: """ Execute a TCL command and return the result. + This is the primary interface for interacting with Vivado. The command + is sent to the Vivado TCL shell, and output is captured by waiting for + the next "Vivado%" prompt. + Args: - command: TCL command to execute + command: TCL command to execute. Can be any valid Vivado TCL command. + Examples: + - "open_project /path/to/project.xpr" + - "report_timing_summary -return_string" + - "get_property PART [current_project]" Returns: - CommandResult with output and status + CommandResult containing: + - output: The command's output (stdout from Vivado) + - success: True if no error keywords were found in output + - elapsed_ms: Execution time in milliseconds + + Thread Safety: + This method is thread-safe. A lock ensures only one command + executes at a time. + + Output Parsing: + The raw pexpect output includes the echoed command and prompts. + This method strips those to return only the meaningful output. + + Error Detection: + Success is determined by checking for error keywords in output: + - "error:" - Vivado error messages + - "invalid command" - TCL syntax errors + - "can't read" - Variable/file access errors + - "wrong # args" - Argument count errors """ + # Check session is running if not self.is_running: return CommandResult( command=command, @@ -139,47 +304,52 @@ class VivadoSession: elapsed_ms=0 ) + # Serialize command execution with a lock with self._lock: start_time = time.time() try: - # Clear any pending output first + # Clear any pending output from previous commands + # This ensures we only capture this command's output try: self.child.read_nonblocking(size=100000, timeout=0.1) except (pexpect.TIMEOUT, pexpect.EOF): - pass + pass # Expected - buffer was empty - # Send the command + # Send the command to Vivado self.child.sendline(command) - # Wait for Vivado prompt (indicates command completed) + # Wait for the Vivado prompt indicating command completion + # The prompt appears after Vivado finishes processing self.child.expect('Vivado%', timeout=self.timeout) - # Get the output (everything before the prompt) + # Get everything that was output before the prompt raw_output = self.child.before - # Parse output: extract content after command echo + # Parse the output to extract meaningful content + # Raw output includes: command echo, actual output, whitespace lines = raw_output.replace('\r', '').split('\n') clean_lines = [] found_command = False - # Normalize command for matching + # Normalize command for matching (handle whitespace differences) cmd_normalized = command.strip() for line in lines: stripped = line.strip() - # Look for the command echo + # Skip lines until we find the echoed command + # Everything before is leftover from previous operations if not found_command: if cmd_normalized in stripped: found_command = True continue - # Skip Vivado prompts + # Skip Vivado prompts in output if stripped == 'Vivado%' or stripped.startswith('Vivado%'): continue - # Skip empty lines + # Skip empty lines for cleaner output if not stripped: continue @@ -189,11 +359,12 @@ class VivadoSession: elapsed = (time.time() - start_time) * 1000 - # Check for errors in output + # Determine success by checking for error indicators + # Vivado prefixes errors with specific keywords success = not any(err in output.lower() for err in ["error:", "invalid command", "can't read", "wrong # args"]) - # Update stats + # Update statistics self.stats["commands_run"] += 1 self.stats["total_command_time_ms"] += elapsed if not success: @@ -207,7 +378,7 @@ class VivadoSession: elapsed_ms=elapsed ) - # Keep last 100 commands in history + # Add to command history (keep last 100 for debugging) self.stats["command_history"].append({ "command": command, "success": success, @@ -220,6 +391,7 @@ class VivadoSession: return result except pexpect.TIMEOUT: + # Command took too long - might be hung or very long operation elapsed = (time.time() - start_time) * 1000 self.stats["errors"] += 1 return CommandResult( @@ -230,6 +402,7 @@ class VivadoSession: elapsed_ms=elapsed ) except Exception as e: + # Unexpected error during command execution elapsed = (time.time() - start_time) * 1000 self.stats["errors"] += 1 return CommandResult( @@ -241,7 +414,19 @@ class VivadoSession: ) def stop(self) -> CommandResult: - """Stop the Vivado session.""" + """ + Stop the Vivado session gracefully. + + Sends the "exit" command to Vivado and waits for the process to + terminate. If graceful exit fails, force-closes the process. + + Returns: + CommandResult with success=True (stopping always "succeeds" + even if we had to force-close) + + Note: + Safe to call even if session is not running. + """ if not self.is_running: return CommandResult( command="stop", @@ -254,15 +439,18 @@ class VivadoSession: start_time = time.time() try: + # Send exit command for graceful shutdown self.child.sendline('exit') + # Wait for process to terminate (EOF on stdout) self.child.expect(pexpect.EOF, timeout=30) except Exception: - # Force close if graceful exit fails + # If graceful exit fails, force-terminate the process try: self.child.close(force=True) except: - pass + pass # Best effort - process might already be dead + # Update state self.is_running = False self.current_project = None elapsed = (time.time() - start_time) * 1000 @@ -276,28 +464,82 @@ class VivadoSession: ) def get_stats(self) -> dict: - """Get session statistics.""" + """ + Get session statistics for monitoring and debugging. + + Returns: + Dictionary containing: + - is_running: Whether session is active + - current_project: Path to open project (or None) + - session_start: ISO timestamp when session started + - commands_run: Total commands executed + - total_command_time_ms: Sum of all command times + - errors: Count of failed commands + - avg_command_time_ms: Average command time (if commands > 0) + - command_history: Last 100 commands with timing info + """ stats = self.stats.copy() stats["is_running"] = self.is_running stats["current_project"] = self.current_project + + # Calculate average command time if we have data if self.stats["commands_run"] > 0: - stats["avg_command_time_ms"] = self.stats["total_command_time_ms"] / self.stats["commands_run"] + stats["avg_command_time_ms"] = ( + self.stats["total_command_time_ms"] / self.stats["commands_run"] + ) + return stats def __enter__(self): + """ + Context manager entry - start the session. + + Example: + with VivadoSession() as session: + session.run_tcl("...") + """ self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): + """ + Context manager exit - stop the session. + + Ensures Vivado is properly shut down even if an exception occurred. + """ self.stop() -# Singleton session instance +# ============================================================================= +# SINGLETON SESSION MANAGEMENT +# ============================================================================= + +# Global singleton session instance +# Using a singleton ensures only one Vivado process runs at a time _session: Optional[VivadoSession] = None def get_session() -> VivadoSession: - """Get or create the global Vivado session.""" + """ + Get or create the global Vivado session. + + This function implements the singleton pattern for VivadoSession. + The first call creates a new session; subsequent calls return the + same instance. + + Returns: + The global VivadoSession instance + + Example: + session = get_session() + session.start() + # ... use session ... + session.stop() + + Note: + The session is created lazily (on first access) and is NOT + automatically started. Call session.start() explicitly. + """ global _session if _session is None: _session = VivadoSession() @@ -305,7 +547,18 @@ def get_session() -> VivadoSession: def reset_session(): - """Reset the global session (stop if running).""" + """ + Reset the global session (stop if running and clear instance). + + Use this to force a fresh Vivado session, for example after + recovering from an error or when changing Vivado versions. + + This function: + 1. Stops the current session if running + 2. Clears the singleton instance + + The next call to get_session() will create a fresh instance. + """ global _session if _session is not None and _session.is_running: _session.stop()