Ryan Malloy
4ae38fed59
Migrate from FastMCP 2.14.5 to 3.1.0 with complete architectural overhaul. Adopt src-layout packaging, lazy config functions to eliminate .env race condition, and decorator-based tool registration. Consolidate 14 tool modules into 8 focused modules (33 tools total). Add 9 new schematic tools via kicad-sch-api for creating and manipulating .kicad_sch files. Drop pandas dependency (BOM uses stdlib csv). Remove ~17k lines of stubs, over-engineering, and dead code. All checks pass: ruff clean, mypy 0 errors, 17/17 tests green.
370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""
|
|
FreeRouting integration tools for automated PCB routing.
|
|
|
|
Wraps the FreeRouting autorouter engine and KiCad IPC API to provide
|
|
automated routing, routing quality analysis, and capability checking
|
|
through MCP tool interfaces.
|
|
"""
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from mckicad.server import mcp
|
|
from mckicad.utils.file_utils import get_project_files
|
|
from mckicad.utils.freerouting import FreeRoutingEngine, check_routing_prerequisites
|
|
from mckicad.utils.ipc_client import check_kicad_availability, kicad_ipc_session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@mcp.tool()
|
|
def check_routing_capability() -> dict[str, Any]:
|
|
"""Check whether automated PCB routing is available on this system.
|
|
|
|
Verifies that all required components are installed and properly
|
|
configured: KiCad IPC API (for real-time board access), FreeRouting
|
|
JAR (for autorouting), and KiCad CLI (for DSN/SES file conversion).
|
|
|
|
Call this before attempting any routing operations to confirm the
|
|
toolchain is ready.
|
|
|
|
Returns:
|
|
Dictionary with overall readiness status, per-component status,
|
|
and a summary of available capabilities.
|
|
"""
|
|
try:
|
|
status = check_routing_prerequisites()
|
|
|
|
return {
|
|
"success": True,
|
|
"routing_available": status["overall_ready"],
|
|
"message": status["message"],
|
|
"component_status": status["components"],
|
|
"capabilities": {
|
|
"automated_routing": status["overall_ready"],
|
|
"interactive_placement": status["components"]
|
|
.get("kicad_ipc", {})
|
|
.get("available", False),
|
|
"optimization": status["overall_ready"],
|
|
"real_time_updates": status["components"]
|
|
.get("kicad_ipc", {})
|
|
.get("available", False),
|
|
},
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error checking routing capability: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"routing_available": False,
|
|
}
|
|
|
|
|
|
@mcp.tool()
|
|
def route_pcb_automatically(
|
|
project_path: str,
|
|
routing_strategy: str = "balanced",
|
|
preserve_existing: bool = False,
|
|
optimization_level: str = "standard",
|
|
) -> dict[str, Any]:
|
|
"""Run the FreeRouting autorouter on a KiCad PCB.
|
|
|
|
Takes a board with placed components and automatically routes all
|
|
(or remaining) copper connections. The workflow is: export DSN from
|
|
KiCad CLI, run FreeRouting, import the routed SES file back.
|
|
|
|
Args:
|
|
project_path: Path to the KiCad project (.kicad_pro) or directory
|
|
containing one.
|
|
routing_strategy: Controls via cost and iteration depth.
|
|
"conservative" minimises vias and iterations (quick, safe).
|
|
"balanced" is a good default for most 2-layer boards.
|
|
"aggressive" allows more vias and iterations for dense boards.
|
|
preserve_existing: When True, existing routed traces are kept and
|
|
only unrouted nets are processed.
|
|
optimization_level: Post-routing cleanup pass intensity.
|
|
"none" skips optimisation.
|
|
"standard" runs a single cleanup pass.
|
|
"aggressive" doubles iteration count and tightens the
|
|
improvement threshold.
|
|
|
|
Returns:
|
|
Dictionary with routing results including pre/post statistics,
|
|
routing report, and the configuration that was used.
|
|
"""
|
|
try:
|
|
files = get_project_files(project_path)
|
|
if "pcb" not in files:
|
|
return {
|
|
"success": False,
|
|
"error": "PCB file not found in project",
|
|
}
|
|
|
|
board_path = files["pcb"]
|
|
|
|
# Map strategy name to FreeRouting parameter set
|
|
routing_configs: dict[str, dict[str, int | float | bool]] = {
|
|
"conservative": {
|
|
"via_costs": 30,
|
|
"start_ripup_costs": 50,
|
|
"max_iterations": 500,
|
|
"automatic_neckdown": False,
|
|
"postroute_optimization": optimization_level != "none",
|
|
},
|
|
"balanced": {
|
|
"via_costs": 50,
|
|
"start_ripup_costs": 100,
|
|
"max_iterations": 1000,
|
|
"automatic_neckdown": True,
|
|
"postroute_optimization": optimization_level != "none",
|
|
},
|
|
"aggressive": {
|
|
"via_costs": 80,
|
|
"start_ripup_costs": 200,
|
|
"max_iterations": 2000,
|
|
"automatic_neckdown": True,
|
|
"postroute_optimization": True,
|
|
},
|
|
}
|
|
|
|
config = routing_configs.get(routing_strategy, routing_configs["balanced"])
|
|
|
|
if optimization_level == "aggressive":
|
|
config.update(
|
|
{
|
|
"improvement_threshold": 0.005,
|
|
"max_iterations": config["max_iterations"] * 2,
|
|
}
|
|
)
|
|
|
|
engine = FreeRoutingEngine()
|
|
|
|
availability = engine.check_freerouting_availability()
|
|
if not availability["available"]:
|
|
return {
|
|
"success": False,
|
|
"error": f"FreeRouting not available: {availability['message']}",
|
|
"routing_strategy": routing_strategy,
|
|
}
|
|
|
|
result = engine.route_board_complete(
|
|
board_path,
|
|
routing_config=config,
|
|
preserve_existing=preserve_existing,
|
|
)
|
|
|
|
result.update(
|
|
{
|
|
"routing_strategy": routing_strategy,
|
|
"optimization_level": optimization_level,
|
|
"project_path": project_path,
|
|
"board_path": board_path,
|
|
}
|
|
)
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in automated routing: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"project_path": project_path,
|
|
"routing_strategy": routing_strategy,
|
|
}
|
|
|
|
|
|
@mcp.tool()
|
|
def analyze_routing_quality(project_path: str) -> dict[str, Any]:
|
|
"""Analyse the current PCB routing for quality and potential issues.
|
|
|
|
Connects to a running KiCad instance via IPC and inspects tracks,
|
|
vias, nets, and footprints to evaluate signal integrity risk,
|
|
routing density, via usage, thermal considerations, and
|
|
manufacturability.
|
|
|
|
Returns a numeric quality score (0-100) together with per-category
|
|
breakdowns and actionable recommendations.
|
|
|
|
Args:
|
|
project_path: Path to the KiCad project (.kicad_pro) or directory
|
|
containing one.
|
|
|
|
Returns:
|
|
Dictionary with quality score, category analyses, and
|
|
improvement recommendations.
|
|
"""
|
|
try:
|
|
files = get_project_files(project_path)
|
|
if "pcb" not in files:
|
|
return {
|
|
"success": False,
|
|
"error": "PCB file not found in project",
|
|
}
|
|
|
|
board_path = files["pcb"]
|
|
|
|
ipc_status = check_kicad_availability()
|
|
if not ipc_status["available"]:
|
|
return {
|
|
"success": False,
|
|
"error": f"KiCad IPC not available: {ipc_status['message']}",
|
|
"project_path": project_path,
|
|
}
|
|
|
|
with kicad_ipc_session(board_path=board_path) as client:
|
|
tracks = client.get_tracks()
|
|
nets = client.get_nets()
|
|
footprints = client.get_footprints()
|
|
connectivity = client.check_connectivity()
|
|
|
|
# --- per-category analysis ---
|
|
|
|
routing_density = _analyze_routing_density(tracks, footprints)
|
|
via_analysis = _analyze_via_usage(tracks)
|
|
trace_analysis = _analyze_trace_characteristics(tracks)
|
|
signal_integrity = _analyze_signal_integrity(tracks, nets)
|
|
thermal_analysis = _analyze_thermal_aspects(tracks, footprints)
|
|
manufacturability = _analyze_manufacturability(tracks)
|
|
|
|
quality_analysis = {
|
|
"connectivity_analysis": connectivity,
|
|
"routing_density": routing_density,
|
|
"via_analysis": via_analysis,
|
|
"trace_analysis": trace_analysis,
|
|
"signal_integrity": signal_integrity,
|
|
"thermal_analysis": thermal_analysis,
|
|
"manufacturability": manufacturability,
|
|
}
|
|
|
|
quality_score = _calculate_quality_score(quality_analysis)
|
|
recommendations = _generate_routing_recommendations(quality_analysis)
|
|
|
|
return {
|
|
"success": True,
|
|
"project_path": project_path,
|
|
"quality_score": quality_score,
|
|
"analysis": quality_analysis,
|
|
"recommendations": recommendations,
|
|
"summary": f"Routing quality score: {quality_score}/100",
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in routing quality analysis: {e}")
|
|
return {
|
|
"success": False,
|
|
"error": str(e),
|
|
"project_path": project_path,
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Private helpers for routing quality analysis
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _analyze_routing_density(tracks: list, footprints: list) -> dict[str, Any]:
|
|
"""Compute track-to-component density ratio."""
|
|
ratio = len(tracks) / max(len(footprints), 1)
|
|
if ratio > 4.0:
|
|
rating = "high"
|
|
elif ratio > 1.5:
|
|
rating = "medium"
|
|
else:
|
|
rating = "low"
|
|
|
|
return {
|
|
"total_tracks": len(tracks),
|
|
"total_footprints": len(footprints),
|
|
"track_per_component": round(ratio, 2),
|
|
"density_rating": rating,
|
|
}
|
|
|
|
|
|
def _analyze_via_usage(tracks: list) -> dict[str, Any]:
|
|
"""Count vias and assess usage density."""
|
|
via_count = sum(1 for t in tracks if hasattr(t, "drill"))
|
|
track_count = len(tracks) - via_count
|
|
via_ratio = via_count / max(track_count, 1)
|
|
|
|
return {
|
|
"total_vias": via_count,
|
|
"total_traces": track_count,
|
|
"via_to_trace_ratio": round(via_ratio, 3),
|
|
"via_density": "high" if via_ratio > 0.3 else "normal",
|
|
}
|
|
|
|
|
|
def _analyze_trace_characteristics(tracks: list) -> dict[str, Any]:
|
|
"""Summarise trace count and basic statistics."""
|
|
trace_count = sum(1 for t in tracks if not hasattr(t, "drill"))
|
|
return {
|
|
"total_traces": trace_count,
|
|
"width_distribution": {"standard": trace_count},
|
|
}
|
|
|
|
|
|
def _analyze_signal_integrity(tracks: list, nets: list) -> dict[str, Any]:
|
|
"""Flag nets whose names suggest high-speed or clock signals."""
|
|
clock_nets = sum(
|
|
1
|
|
for n in nets
|
|
if n.name and any(kw in n.name.lower() for kw in ("clk", "clock", "mclk"))
|
|
)
|
|
return {
|
|
"critical_nets": clock_nets,
|
|
"high_speed_traces": 0,
|
|
"impedance_controlled": False,
|
|
}
|
|
|
|
|
|
def _analyze_thermal_aspects(tracks: list, footprints: list) -> dict[str, Any]:
|
|
"""Basic thermal heuristic (placeholder for deeper analysis)."""
|
|
return {
|
|
"thermal_vias": 0,
|
|
"power_trace_width": "adequate",
|
|
"heat_dissipation": "good",
|
|
}
|
|
|
|
|
|
def _analyze_manufacturability(tracks: list) -> dict[str, Any]:
|
|
"""Placeholder manufacturability assessment."""
|
|
return {
|
|
"minimum_trace_width_mm": 0.1,
|
|
"minimum_spacing_mm": 0.1,
|
|
"manufacturability_rating": "good",
|
|
}
|
|
|
|
|
|
def _calculate_quality_score(analysis: dict[str, Any]) -> int:
|
|
"""Derive a 0-100 quality score from the sub-analyses."""
|
|
base = 75
|
|
connectivity = analysis.get("connectivity_analysis", {})
|
|
completion = connectivity.get("routing_completion", 0)
|
|
# Completion contributes up to 25 points
|
|
return min(int(base + completion * 0.25), 100)
|
|
|
|
|
|
def _generate_routing_recommendations(analysis: dict[str, Any]) -> list[str]:
|
|
"""Produce a list of human-readable improvement suggestions."""
|
|
recs: list[str] = []
|
|
|
|
connectivity = analysis.get("connectivity_analysis", {})
|
|
unrouted = connectivity.get("unrouted_nets", 0)
|
|
if unrouted > 0:
|
|
recs.append(f"Complete routing for {unrouted} unrouted net(s)")
|
|
|
|
via_info = analysis.get("via_analysis", {})
|
|
if via_info.get("via_density") == "high":
|
|
recs.append("Consider reducing via count for improved signal integrity")
|
|
|
|
density = analysis.get("routing_density", {})
|
|
if density.get("density_rating") == "high":
|
|
recs.append("High routing density detected -- verify clearance rules")
|
|
|
|
recs.append("Run DRC check to validate design rules after routing changes")
|
|
recs.append("Verify impedance control for high-speed signals")
|
|
|
|
return recs
|