portugalfuturista/vivado_mcp
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Improve reliability with smart error detection and health checks

Browse source 
- Add ErrorClassification to distinguish real errors from report content
  containing error-like strings (e.g., "Timing ERROR: 0")
- Verify synthesis/implementation success via Vivado run properties
  instead of parsing output text
- Add session health checking (is_healthy, ensure_healthy) with
  auto-recovery for unresponsive sessions
- Add per-command timeout_override for long operations (synth: 30min,
  impl: 60min defaults)
- Add check_session_health tool for manual health verification

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
cah 2026-02-04 19:15:40 -07:00
parent 879ed18f67
commit c019e32ed8
2 changed files with 448 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

302
server.py
View file

@ -70,7 +70,7 @@ REPORTS_DIR = Path("/tmp/vivado_mcp")
# 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
MAX_RESPONSE_CHARS = 8000 # ~8KB limit for inline responses
# How long to keep cached report files before cleanup (in hours)
REPORT_CACHE_HOURS = 1
@ -187,6 +187,45 @@ def truncate_response(content: str, max_chars: int = MAX_RESPONSE_CHARS) -> dict
}
def verify_run_status(session, run_name: str) -> dict:
"""
Verify actual Vivado run status instead of relying on output parsing.
Vivado run status is stored as properties on the run object. This function
queries those properties directly, which is more reliable than parsing
text output that may contain misleading strings.
Args:
session: VivadoSession instance
run_name: Name of the run to check (e.g., "synth_1", "impl_1")
Returns:
Dictionary with:
- run_name: The run that was checked
- status: Vivado's STATUS property (e.g., "synth_design Complete!")
- progress: Vivado's PROGRESS property (e.g., "100%")
- actually_succeeded: True if run completed successfully
- actually_failed: True if run failed
"""
status_result = session.run_tcl(f"get_property STATUS [get_runs {run_name}]")
progress_result = session.run_tcl(f"get_property PROGRESS [get_runs {run_name}]")
status = status_result.output.strip() if status_result.success else "unknown"
progress = progress_result.output.strip() if progress_result.success else "unknown"
# Determine actual success/failure from status string
# Successful runs have "Complete!" in status
# Failed runs have "ERROR" in status
status_lower = status.lower()
return {
"run_name": run_name,
"status": status,
"progress": progress,
"actually_succeeded": "complete" in status_lower,
"actually_failed": "error" in status_lower,
}
# =============================================================================
# REPORT FILE MANAGEMENT
# =============================================================================
@ -423,6 +462,90 @@ def parse_messages(output: str) -> dict:
return result
def parse_timing_paths_summary(output: str, max_paths: int = 5) -> list[dict]:
"""
Extract structured summary of timing paths from report_timing output.
Parses Vivado's timing path reports to extract key information about
each path without the verbose detailed breakdown.
Args:
output: Raw text output from report_timing command
max_paths: Maximum number of paths to return (default: 5)
Returns:
List of dictionaries, each containing:
- slack: Path slack in ns (negative = failing)
- source: Source register/port name
- destination: Destination register/port name
- source_clock: Source clock domain (if applicable)
- dest_clock: Destination clock domain (if applicable)
- requirement: Timing requirement in ns
- data_path_delay: Data path delay in ns
- logic_levels: Number of logic levels
"""
paths = []
# Split output into individual path blocks
# Each path starts with "Slack" line
path_blocks = re.split(r'\n(?=Slack\s*(?:\([A-Z]+\))?\s*:)', output)
for block in path_blocks:
if not block.strip() or 'Slack' not in block:
continue
path_info = {}
# Extract slack value
slack_match = re.search(r'Slack\s*(?:\([A-Z]+\))?\s*:\s*([-\d.]+)\s*ns', block)
if slack_match:
path_info['slack'] = float(slack_match.group(1))
# Extract source (startpoint)
source_match = re.search(r'Source:\s*(\S+)', block)
if source_match:
path_info['source'] = source_match.group(1)
# Extract destination (endpoint)
dest_match = re.search(r'Destination:\s*(\S+)', block)
if dest_match:
path_info['destination'] = dest_match.group(1)
# Extract source clock
src_clk_match = re.search(r'Source Clock:\s*(\S+)', block)
if src_clk_match:
path_info['source_clock'] = src_clk_match.group(1)
# Extract destination clock
dst_clk_match = re.search(r'Destination Clock:\s*(\S+)', block)
if dst_clk_match:
path_info['dest_clock'] = dst_clk_match.group(1)
# Extract requirement
req_match = re.search(r'Requirement:\s*([-\d.]+)\s*ns', block)
if req_match:
path_info['requirement'] = float(req_match.group(1))
# Extract data path delay
data_delay_match = re.search(r'Data Path Delay:\s*([-\d.]+)\s*ns', block)
if data_delay_match:
path_info['data_path_delay'] = float(data_delay_match.group(1))
# Extract logic levels
levels_match = re.search(r'Logic Levels:\s*(\d+)', block)
if levels_match:
path_info['logic_levels'] = int(levels_match.group(1))
# Only add if we got meaningful data
if 'slack' in path_info:
paths.append(path_info)
if len(paths) >= max_paths:
break
return paths
# =============================================================================
# TOOL DEFINITIONS
# =============================================================================
@ -490,6 +613,20 @@ async def list_tools() -> list[Tool]:
"required": []
}
),
Tool(
name="check_session_health",
description="Check if Vivado session is responsive and recover if needed. Use this if commands are timing out or behaving unexpectedly.",
inputSchema={
"type": "object",
"properties": {
"auto_recover": {
"type": "boolean",
"description": "Restart session if unhealthy (default: true)"
}
},
"required": []
}
),
# =====================================================================
# PROJECT MANAGEMENT TOOLS
@ -543,6 +680,10 @@ async def list_tools() -> list[Tool]:
"jobs": {
"type": "integer",
"description": "Number of parallel jobs (default: 4)"
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default: 1800 = 30 minutes). Increase for large designs."
}
},
"required": []
@ -557,6 +698,10 @@ async def list_tools() -> list[Tool]:
"jobs": {
"type": "integer",
"description": "Number of parallel jobs (default: 4)"
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default: 3600 = 60 minutes). Increase for large designs."
}
},
"required": []
@ -579,7 +724,7 @@ async def list_tools() -> list[Tool]:
Tool(
name="get_timing_summary",
description="Get timing summary (WNS, TNS, WHS, THS) - returns structured data",
description="Get timing summary (WNS, TNS, WHS, THS). Returns parsed metrics only by default. Use generate_full_report for raw output.",
inputSchema={
"type": "object",
"properties": {
@ -590,7 +735,7 @@ async def list_tools() -> list[Tool]:
"detail_level": {
"type": "string",
"enum": ["summary", "standard", "full"],
"description": "Detail level: 'summary' (parsed metrics only), 'standard' (default), 'full' (include raw report)"
"description": "Detail level: 'summary' (default, parsed metrics only), 'standard' (+ truncated raw), 'full' (+ complete raw)"
}
},
"required": []
@ -598,7 +743,7 @@ async def list_tools() -> list[Tool]:
),
Tool(
name="get_timing_paths",
description="Get detailed timing paths for failing or critical paths",
description="Get timing paths for failing or critical paths. Returns structured summary (slack, source, dest, clocks) by default. Use generate_full_report for verbose path details.",
inputSchema={
"type": "object",
"properties": {
@ -629,6 +774,11 @@ async def list_tools() -> list[Tool]:
"clock": {
"type": "string",
"description": "Filter paths by clock domain name"
},
"detail_level": {
"type": "string",
"enum": ["summary", "standard", "full"],
"description": "Detail level: 'summary' (default, structured only), 'standard' (+ truncated raw), 'full' (+ complete raw)"
}
},
"required": []
@ -636,7 +786,7 @@ async def list_tools() -> list[Tool]:
),
Tool(
name="get_utilization",
description="Get resource utilization report - returns structured data",
description="Get resource utilization (LUT, FF, BRAM, DSP, IO). Returns parsed metrics only by default. Use generate_full_report for hierarchical details.",
inputSchema={
"type": "object",
"properties": {
@ -647,7 +797,7 @@ async def list_tools() -> list[Tool]:
"detail_level": {
"type": "string",
"enum": ["summary", "standard", "full"],
"description": "Detail level: 'summary' (parsed only), 'standard' (default, + top consumers), 'full' (+ raw report)"
"description": "Detail level: 'summary' (default, parsed only), 'standard' (+ truncated raw), 'full' (+ complete raw)"
},
"module_filter": {
"type": "string",
@ -1161,6 +1311,52 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
stats = session.get_stats()
return [TextContent(type="text", text=json.dumps(stats, indent=2))]
elif name == "check_session_health":
# Check if session is responsive and optionally recover
auto_recover = arguments.get("auto_recover", True)
if not session.is_running:
if auto_recover:
result = session.start()
return [TextContent(type="text", text=json.dumps({
"healthy": result.success,
"action": "started",
"message": "Session was not running, started new session",
"elapsed_ms": result.elapsed_ms
}, indent=2))]
else:
return [TextContent(type="text", text=json.dumps({
"healthy": False,
"action": "none",
"message": "Session not running (auto_recover=false)"
}, indent=2))]
# Session thinks it's running, check if actually responsive
is_healthy = session.is_healthy()
if is_healthy:
return [TextContent(type="text", text=json.dumps({
"healthy": True,
"action": "none",
"message": "Session is healthy and responsive"
}, indent=2))]
# Session is unresponsive
if auto_recover:
result = session.ensure_healthy()
return [TextContent(type="text", text=json.dumps({
"healthy": result.success,
"action": "restarted",
"message": "Session was unresponsive, restarted",
"elapsed_ms": result.elapsed_ms
}, indent=2))]
else:
return [TextContent(type="text", text=json.dumps({
"healthy": False,
"action": "none",
"message": "Session is unresponsive (auto_recover=false)"
}, indent=2))]
# =========================================================================
# SESSION CHECK
# =========================================================================
@ -1221,22 +1417,58 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
# 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({
"success": result.success,
timeout = arguments.get("timeout", 1800) # 30 min default
result = session.run_tcl(
f"reset_run synth_1; launch_runs synth_1 -jobs {jobs}; wait_on_run synth_1",
timeout_override=timeout
)
# Verify actual run status (more reliable than output parsing)
verification = verify_run_status(session, "synth_1")
actual_success = verification["actually_succeeded"]
response = {
"success": actual_success,
"output": result.output,
"elapsed_ms": result.elapsed_ms
}, indent=2))]
"elapsed_ms": result.elapsed_ms,
"run_status": verification["status"],
"run_progress": verification["progress"],
}
# Note if there was a mismatch between output parsing and actual status
if not result.success and actual_success:
response["note"] = "Output contained error-like strings but run completed successfully"
return [TextContent(type="text", text=json.dumps(response, 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({
"success": result.success,
timeout = arguments.get("timeout", 3600) # 60 min default
result = session.run_tcl(
f"launch_runs impl_1 -jobs {jobs}; wait_on_run impl_1",
timeout_override=timeout
)
# Verify actual run status (more reliable than output parsing)
verification = verify_run_status(session, "impl_1")
actual_success = verification["actually_succeeded"]
response = {
"success": actual_success,
"output": result.output,
"elapsed_ms": result.elapsed_ms
}, indent=2))]
"elapsed_ms": result.elapsed_ms,
"run_status": verification["status"],
"run_progress": verification["progress"],
}
# Note if there was a mismatch between output parsing and actual status
if not result.success and actual_success:
response["note"] = "Output contained error-like strings but run completed successfully"
return [TextContent(type="text", text=json.dumps(response, indent=2))]
elif name == "generate_bitstream":
# Generate bitstream (programming file)
@ -1254,7 +1486,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
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")
detail_level = arguments.get("detail_level", "summary")
# Run Vivado timing summary report
result = session.run_tcl("report_timing_summary -no_header -return_string")
@ -1298,6 +1530,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
to_pin = arguments.get("to_pin")
through = arguments.get("through")
clock = arguments.get("clock")
detail_level = arguments.get("detail_level", "summary")
# Build the report_timing command
delay_type = "max" if path_type == "setup" else "min"
@ -1337,23 +1570,40 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if clock:
response["filters_applied"]["clock"] = clock
# Handle potentially large output
# Handle output based on detail level
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"]
# Always parse paths into structured format
parsed_paths = parse_timing_paths_summary(result.output, max_paths=num_paths)
response["paths"] = parsed_paths
response["path_count"] = len(parsed_paths)
if detail_level == "summary":
# Only return structured data, no raw output
pass
elif detail_level == "standard":
# Include truncated raw for reference
truncated = truncate_response(result.output, MAX_RESPONSE_CHARS // 2)
response["raw"] = truncated["content"]
if truncated["truncated"]:
response["raw_truncated"] = True
response["raw_total_chars"] = truncated["total_chars"]
elif detail_level == "full":
# Include complete raw output
truncated = truncate_response(result.output, MAX_RESPONSE_CHARS)
response["raw"] = truncated["content"]
if truncated["truncated"]:
response["raw_truncated"] = True
response["raw_total_chars"] = truncated["total_chars"]
response["truncation_message"] = truncated["truncation_message"]
else:
response["paths"] = result.output
response["error"] = 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")
detail_level = arguments.get("detail_level", "summary")
module_filter = arguments.get("module_filter")
threshold_percent = arguments.get("threshold_percent")

184
vivado_session.py
View file

@ -90,6 +90,107 @@ class CommandResult:
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
@dataclass
class ErrorClassification:
"""
Classification of Vivado output for smart error detection.
Distinguishes between actual errors (TCL syntax/runtime errors, Vivado tool
errors) and false positives (error strings that appear in report output like
"Timing ERROR: 0" or utilization tables).
Attributes:
is_tcl_error: True if TCL syntax or runtime error detected
is_vivado_error: True if Vivado tool error (lines starting with ERROR:)
is_report_content: True if output appears to be report/table data
error_messages: List of actual error message strings found
"""
is_tcl_error: bool = False
is_vivado_error: bool = False
is_report_content: bool = False
error_messages: list = field(default_factory=list)
@property
def is_actual_failure(self) -> bool:
"""Return True only if this is a real error, not report content."""
return self.is_tcl_error or self.is_vivado_error
def classify_output_errors(output: str, command: str) -> ErrorClassification:
"""
Classify errors based on context - distinguishes real failures from
report content that happens to contain 'error' strings.
This function performs smart error detection by:
1. Checking for TCL syntax errors at the START of output
2. Looking for Vivado errors that START with "ERROR:" (not just contain it)
3. Detecting report context (tables, summaries) where "error" is just data
Args:
output: The raw output from Vivado
command: The command that was executed (for context)
Returns:
ErrorClassification with details about any errors found
Example:
# This is a real error:
# "ERROR: [Synth 8-87] can't read file..."
# This is NOT an error (report content):
# "| Timing ERROR | 0 |"
# "WNS(ns): -0.5 TNS ERROR: 0"
"""
classification = ErrorClassification()
lines = output.strip().split('\n')
# TCL syntax errors - appear at START of output (first few lines)
tcl_error_patterns = [
r'^invalid command name',
r'^wrong # args:',
r'^can\'t read ".*": no such variable',
r'^expected .* but got',
r'^couldn\'t open',
r'^no files matched',
]
for line in lines[:5]:
stripped = line.strip()
for pattern in tcl_error_patterns:
if re.match(pattern, stripped, re.IGNORECASE):
classification.is_tcl_error = True
classification.error_messages.append(stripped)
# Vivado errors - lines STARTING with "ERROR:" followed by bracket
# Real errors look like: "ERROR: [Synth 8-87] description"
# False positives look like: "| Timing ERROR | 0 |" or "error: 0"
for line in lines:
stripped = line.strip()
# Match lines that START with ERROR: followed by a bracket (Vivado error code)
if re.match(r'^ERROR:\s*\[', stripped):
classification.is_vivado_error = True
classification.error_messages.append(stripped)
# Detect report context - error strings in tables/summaries don't count as errors
# These indicators suggest we're looking at report output, not error messages
report_indicators = [
'WNS(ns)', # Timing summary
'TNS(ns)', # Timing summary
'WHS(ns)', # Timing summary
'+---------', # Table borders
'|------', # Table borders
'| Site Type', # Utilization report
'| Resource', # Utilization report
'Utilization', # Utilization report header
'Design Timing Summary',
'Clock Summary',
]
if any(ind in output for ind in report_indicators):
classification.is_report_content = True
return classification
# =============================================================================
# VIVADO SESSION CLASS
# =============================================================================
@ -258,7 +359,7 @@ class VivadoSession:
elapsed_ms=elapsed
)
def run_tcl(self, command: str) -> CommandResult:
def run_tcl(self, command: str, timeout_override: float = None) -> CommandResult:
"""
Execute a TCL command and return the result.
@ -272,11 +373,14 @@ class VivadoSession:
- "open_project /path/to/project.xpr"
- "report_timing_summary -return_string"
- "get_property PART [current_project]"
timeout_override: Optional timeout in seconds for this specific command.
Useful for long-running operations like synthesis (30+ min)
or implementation (60+ min). If None, uses self.timeout.
Returns:
CommandResult containing:
- output: The command's output (stdout from Vivado)
- success: True if no error keywords were found in output
- success: True if no actual error was detected
- elapsed_ms: Execution time in milliseconds
Thread Safety:
@ -288,11 +392,11 @@ class VivadoSession:
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
Uses smart error classification to distinguish real errors from
report content that contains error-like strings. Real errors are:
- TCL syntax errors at start of output
- Vivado errors (lines starting with "ERROR: [code]")
Report content like "Timing ERROR: 0" is NOT treated as an error.
"""
# Check session is running
if not self.is_running:
@ -321,7 +425,9 @@ class VivadoSession:
# Wait for the Vivado prompt indicating command completion
# The prompt appears after Vivado finishes processing
self.child.expect('Vivado%', timeout=self.timeout)
# Use timeout_override if provided (for long operations like synthesis)
effective_timeout = timeout_override if timeout_override is not None else self.timeout
self.child.expect('Vivado%', timeout=effective_timeout)
# Get everything that was output before the prompt
raw_output = self.child.before
@ -359,10 +465,10 @@ class VivadoSession:
elapsed = (time.time() - start_time) * 1000
# 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"])
# Use smart error classification to detect real errors
# This avoids false positives from report content like "Timing ERROR: 0"
classification = classify_output_errors(output, command)
success = not classification.is_actual_failure
# Update statistics
self.stats["commands_run"] += 1
@ -490,6 +596,60 @@ class VivadoSession:
return stats
def is_healthy(self) -> bool:
"""
Check if the Vivado session is responsive.
Sends a simple command to Vivado and checks if it responds within
a short timeout. This is useful for detecting hung or dead sessions.
Returns:
True if session responds, False if unresponsive or not running
Note:
This is a quick check (5 second timeout). Use ensure_healthy()
if you want to automatically recover from unhealthy sessions.
"""
if not self.is_running or not self.child:
return False
try:
# Send a simple command that produces predictable output
self.child.sendline("puts {HEALTH_OK}")
self.child.expect("HEALTH_OK", timeout=5)
self.child.expect("Vivado%", timeout=5)
return True
except (pexpect.TIMEOUT, pexpect.EOF):
return False
def ensure_healthy(self) -> CommandResult:
"""
Check session health and restart if needed.
This is the recommended way to recover from session failures.
It checks if the session is responsive and automatically restarts
it if not.
Returns:
CommandResult indicating health status or restart result
Example:
result = session.ensure_healthy()
if result.success:
# Session is ready to use
session.run_tcl("...")
"""
if self.is_healthy():
return CommandResult(
command="health_check",
output="Session healthy",
return_value="0",
success=True,
elapsed_ms=0
)
# Session is unhealthy, try to restart
self.stop()
return self.start()
def __enter__(self):
"""
Context manager entry - start the session.