From 76503b144c59bb464f5e1c0e3ca815745511c24e Mon Sep 17 00:00:00 2001 From: Tom Date: Sat, 28 Feb 2026 01:23:36 +0100 Subject: [PATCH] fix: generate_netlist schematic_path, PinLocator cache, server.ts Python detection - connection_schematic.py: generate_netlist() now accepts schematic_path param, threaded through to get_net_connections() so PinLocator is actually invoked (previously only 1 connection per component was returned due to fallback break) - kicad_interface.py: pass schematic_path to generate_netlist() - pin_locator.py: add _schematic_cache to avoid loading Schematic() once per pin (was causing timeout: O(nets x components x pins) Schematic() calls) - server.ts: remove fragile PYTHONPATH?.includes('KiCad') condition, always prefer KiCAD bundled Python on Windows when executable exists - CHANGELOG.md: document fixes under v2.2.0-alpha --- CHANGELOG.md | 21 + python/commands/connection_schematic.py | 192 ++++-- python/commands/pin_locator.py | 150 +++-- python/kicad_interface.py | 812 +++++++++++++----------- src/server.ts | 442 ++++++++----- 5 files changed, 967 insertions(+), 650 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index abd8562..c355aec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,27 @@ All notable changes to the KiCAD MCP Server project are documented here. - `library.py`: Fix loop variable shadowing `Path` object (mypy) - `design_rules.py`: Add type annotation for `violation_counts` (mypy) +### Pending fixes (not yet committed) + +- `connection_schematic.py` / `kicad_interface.py`: Fix `generate_netlist` missing + `schematic_path` parameter – without it `get_net_connections` always fell back to + proximity matching which only returns one connection per component (first wire hit, + then `break`). PinLocator was never invoked. Fix: added `schematic_path: Optional[Path]` + to `generate_netlist` signature and threaded it through to `get_net_connections`, + and updated `_handle_generate_netlist` in `kicad_interface.py` to pass `schematic_path`. +- `server.ts`: Fix KiCAD bundled Python (3.11.5) not being selected on Windows – the + detection condition `process.env.PYTHONPATH?.includes("KiCad")` was fragile and failed + in some environments, causing System Python 3.12 to be used instead. Since `pcbnew.pyd` + is compiled for KiCAD's Python 3.11.5, this resulted in `No module named 'pcbnew'`. + Fix: removed the condition, KiCAD bundled Python is now always preferred on Windows + when it exists at `C:\Program Files\KiCad\9.0\bin\python.exe`. + Also added `KICAD_PYTHON` to `claude_desktop_config.json` as explicit override. +- `pin_locator.py`: Fix `generate_netlist` timeout – `get_pin_location` and + `get_all_symbol_pins` called `Schematic(schematic_path)` on every single pin lookup, + causing O(nets × components × pins) schematic file loads (e.g. 400+ loads for a + medium schematic). Fix: added `_schematic_cache` dict to `PinLocator.__init__`, + schematic is now loaded once per path and reused. + --- ## [2.1.0-alpha] - 2026-01-10 diff --git a/python/commands/connection_schematic.py b/python/commands/connection_schematic.py index 15b552d..286b67f 100644 --- a/python/commands/connection_schematic.py +++ b/python/commands/connection_schematic.py @@ -10,11 +10,13 @@ logger = logging.getLogger(__name__) try: from commands.wire_manager import WireManager from commands.pin_locator import PinLocator + WIRE_MANAGER_AVAILABLE = True except ImportError: logger.warning("WireManager/PinLocator not available") WIRE_MANAGER_AVAILABLE = False + class ConnectionManager: """Manage connections between components in schematics""" @@ -29,7 +31,12 @@ class ConnectionManager: return cls._pin_locator @staticmethod - def add_wire(schematic_path: Path, start_point: list, end_point: list, properties: dict = None): + def add_wire( + schematic_path: Path, + start_point: list, + end_point: list, + properties: dict = None, + ): """ Add a wire between two points using WireManager @@ -47,11 +54,18 @@ class ConnectionManager: logger.error("WireManager not available") return False - stroke_width = properties.get('stroke_width', 0) if properties else 0 - stroke_type = properties.get('stroke_type', 'default') if properties else 'default' + stroke_width = properties.get("stroke_width", 0) if properties else 0 + stroke_type = ( + properties.get("stroke_type", "default") if properties else "default" + ) - success = WireManager.add_wire(schematic_path, start_point, end_point, - stroke_width=stroke_width, stroke_type=stroke_type) + success = WireManager.add_wire( + schematic_path, + start_point, + end_point, + stroke_width=stroke_width, + stroke_type=stroke_type, + ) return success except Exception as e: logger.error(f"Error adding wire: {e}") @@ -70,7 +84,7 @@ class ConnectionManager: [x, y] coordinates or None if pin not found """ try: - if not hasattr(symbol, 'pin'): + if not hasattr(symbol, "pin"): logger.warning(f"Symbol {symbol.property.Reference.value} has no pins") return None @@ -82,7 +96,9 @@ class ConnectionManager: break if not target_pin: - logger.warning(f"Pin '{pin_name}' not found on {symbol.property.Reference.value}") + logger.warning( + f"Pin '{pin_name}' not found on {symbol.property.Reference.value}" + ) return None # Get pin location relative to symbol @@ -101,8 +117,14 @@ class ConnectionManager: return None @staticmethod - def add_connection(schematic_path: Path, source_ref: str, source_pin: str, - target_ref: str, target_pin: str, routing: str = 'direct'): + def add_connection( + schematic_path: Path, + source_ref: str, + source_pin: str, + target_ref: str, + target_pin: str, + routing: str = "direct", + ): """ Add a wire connection between two component pins @@ -128,31 +150,41 @@ class ConnectionManager: return False # Get pin locations - source_loc = locator.get_pin_location(schematic_path, source_ref, source_pin) - target_loc = locator.get_pin_location(schematic_path, target_ref, target_pin) + source_loc = locator.get_pin_location( + schematic_path, source_ref, source_pin + ) + target_loc = locator.get_pin_location( + schematic_path, target_ref, target_pin + ) if not source_loc or not target_loc: logger.error("Could not determine pin locations") return False # Create wire based on routing style - if routing == 'direct': + if routing == "direct": # Simple direct wire success = WireManager.add_wire(schematic_path, source_loc, target_loc) - elif routing == 'orthogonal_h': + elif routing == "orthogonal_h": # Orthogonal routing (horizontal first) - path = WireManager.create_orthogonal_path(source_loc, target_loc, prefer_horizontal_first=True) + path = WireManager.create_orthogonal_path( + source_loc, target_loc, prefer_horizontal_first=True + ) success = WireManager.add_polyline_wire(schematic_path, path) - elif routing == 'orthogonal_v': + elif routing == "orthogonal_v": # Orthogonal routing (vertical first) - path = WireManager.create_orthogonal_path(source_loc, target_loc, prefer_horizontal_first=False) + path = WireManager.create_orthogonal_path( + source_loc, target_loc, prefer_horizontal_first=False + ) success = WireManager.add_polyline_wire(schematic_path, path) else: logger.error(f"Unknown routing style: {routing}") return False if success: - logger.info(f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})") + logger.info( + f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})" + ) return True else: return False @@ -160,6 +192,7 @@ class ConnectionManager: except Exception as e: logger.error(f"Error adding connection: {e}") import traceback + logger.error(traceback.format_exc()) return False @@ -177,13 +210,12 @@ class ConnectionManager: Label object or None on error """ try: - if not hasattr(schematic, 'label'): + if not hasattr(schematic, "label"): logger.error("Schematic does not have label collection") return None label = schematic.label.append( - text=net_name, - at={'x': position[0], 'y': position[1]} + text=net_name, at={"x": position[0], "y": position[1]} ) logger.info(f"Added net label '{net_name}' at {position}") return label @@ -192,7 +224,9 @@ class ConnectionManager: return None @staticmethod - def connect_to_net(schematic_path: Path, component_ref: str, pin_name: str, net_name: str): + def connect_to_net( + schematic_path: Path, component_ref: str, pin_name: str, net_name: str + ): """ Connect a component pin to a named net using a wire stub and label @@ -231,7 +265,9 @@ class ConnectionManager: return False # Add label at the end of the stub using WireManager - label_success = WireManager.add_label(schematic_path, net_name, stub_end, label_type='label') + label_success = WireManager.add_label( + schematic_path, net_name, stub_end, label_type="label" + ) if not label_success: logger.error(f"Failed to add net label '{net_name}'") return False @@ -242,11 +278,14 @@ class ConnectionManager: except Exception as e: logger.error(f"Error connecting to net: {e}") import traceback + logger.error(traceback.format_exc()) return False @staticmethod - def get_net_connections(schematic: Schematic, net_name: str, schematic_path: Optional[Path] = None): + def get_net_connections( + schematic: Schematic, net_name: str, schematic_path: Optional[Path] = None + ): """ Get all connections for a named net using wire graph analysis @@ -273,14 +312,14 @@ class ConnectionManager: return dx < tolerance and dy < tolerance # 1. Find all labels with this net name - if not hasattr(schematic, 'label'): + if not hasattr(schematic, "label"): logger.warning("Schematic has no labels") return connections net_label_positions = [] for label in schematic.label: - if hasattr(label, 'value') and label.value == net_name: - if hasattr(label, 'at') and hasattr(label.at, 'value'): + if hasattr(label, "value") and label.value == net_name: + if hasattr(label, "at") and hasattr(label.at, "value"): pos = label.at.value net_label_positions.append([float(pos[0]), float(pos[1])]) @@ -288,21 +327,25 @@ class ConnectionManager: logger.info(f"No labels found for net '{net_name}'") return connections - logger.debug(f"Found {len(net_label_positions)} labels for net '{net_name}'") + logger.debug( + f"Found {len(net_label_positions)} labels for net '{net_name}'" + ) # 2. Find all wires connected to these label positions - if not hasattr(schematic, 'wire'): + if not hasattr(schematic, "wire"): logger.warning("Schematic has no wires") return connections connected_wire_points = set() for wire in schematic.wire: - if hasattr(wire, 'pts') and hasattr(wire.pts, 'xy'): + if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): # Get all points in this wire (polyline) wire_points = [] for point in wire.pts.xy: - if hasattr(point, 'value'): - wire_points.append([float(point.value[0]), float(point.value[1])]) + if hasattr(point, "value"): + wire_points.append( + [float(point.value[0]), float(point.value[1])] + ) # Check if any wire point touches a label wire_connected = False @@ -323,10 +366,12 @@ class ConnectionManager: logger.debug(f"No wires connected to net '{net_name}' labels") return connections - logger.debug(f"Found {len(connected_wire_points)} wire connection points for net '{net_name}'") + logger.debug( + f"Found {len(connected_wire_points)} wire connection points for net '{net_name}'" + ) # 3. Find component pins at wire endpoints - if not hasattr(schematic, 'symbol'): + if not hasattr(schematic, "symbol"): logger.warning("Schematic has no symbols") return connections @@ -337,15 +382,15 @@ class ConnectionManager: for symbol in schematic.symbol: # Skip template symbols - if not hasattr(symbol.property, 'Reference'): + if not hasattr(symbol.property, "Reference"): continue ref = symbol.property.Reference.value - if ref.startswith('_TEMPLATE'): + if ref.startswith("_TEMPLATE"): continue # Get lib_id for pin location lookup - lib_id = symbol.lib_id.value if hasattr(symbol, 'lib_id') else None + lib_id = symbol.lib_id.value if hasattr(symbol, "lib_id") else None if not lib_id: continue @@ -360,17 +405,18 @@ class ConnectionManager: # Check each pin for pin_num, pin_data in pins.items(): # Get pin location - pin_loc = locator.get_pin_location(schematic_path, ref, pin_num) + pin_loc = locator.get_pin_location( + schematic_path, ref, pin_num + ) if not pin_loc: continue # Check if pin coincides with any wire point for wire_pt in connected_wire_points: if points_coincide(pin_loc, list(wire_pt)): - connections.append({ - "component": ref, - "pin": pin_num - }) + connections.append( + {"component": ref, "pin": pin_num} + ) break # Pin found, no need to check more wire points except Exception as e: @@ -380,7 +426,7 @@ class ConnectionManager: # Fallback: proximity-based matching if no PinLocator if not locator or not schematic_path: - symbol_pos = symbol.at.value if hasattr(symbol, 'at') else None + symbol_pos = symbol.at.value if hasattr(symbol, "at") else None if not symbol_pos: continue @@ -389,12 +435,11 @@ class ConnectionManager: # Check if symbol is near any wire point (within 10mm) for wire_pt in connected_wire_points: - dist = ((symbol_x - wire_pt[0])**2 + (symbol_y - wire_pt[1])**2)**0.5 + dist = ( + (symbol_x - wire_pt[0]) ** 2 + (symbol_y - wire_pt[1]) ** 2 + ) ** 0.5 if dist < 10.0: # 10mm proximity threshold - connections.append({ - "component": ref, - "pin": "unknown" - }) + connections.append({"component": ref, "pin": "unknown"}) break # Only add once per component logger.info(f"Found {len(connections)} connections for net '{net_name}'") @@ -403,14 +448,20 @@ class ConnectionManager: except Exception as e: logger.error(f"Error getting net connections: {e}") import traceback + logger.error(traceback.format_exc()) return [] @staticmethod - def generate_netlist(schematic: Schematic): + def generate_netlist(schematic: Schematic, schematic_path: Optional[Path] = None): """ Generate a netlist from the schematic + Args: + schematic: Schematic object + schematic_path: Optional path to schematic file (enables accurate pin matching + via PinLocator; without it, only one connection per component is found) + Returns: Dictionary with net information: { @@ -431,47 +482,58 @@ class ConnectionManager: } """ try: - netlist = { - "nets": [], - "components": [] - } + netlist = {"nets": [], "components": []} # Gather all components - if hasattr(schematic, 'symbol'): + if hasattr(schematic, "symbol"): for symbol in schematic.symbol: component_info = { "reference": symbol.property.Reference.value, - "value": symbol.property.Value.value if hasattr(symbol.property, 'Value') else "", - "footprint": symbol.property.Footprint.value if hasattr(symbol.property, 'Footprint') else "" + "value": ( + symbol.property.Value.value + if hasattr(symbol.property, "Value") + else "" + ), + "footprint": ( + symbol.property.Footprint.value + if hasattr(symbol.property, "Footprint") + else "" + ), } netlist["components"].append(component_info) # Gather all nets from labels - if hasattr(schematic, 'label'): + if hasattr(schematic, "label"): net_names = set() for label in schematic.label: - if hasattr(label, 'value'): + if hasattr(label, "value"): net_names.add(label.value) # For each net, get connections for net_name in net_names: - connections = ConnectionManager.get_net_connections(schematic, net_name) + connections = ConnectionManager.get_net_connections( + schematic, net_name, schematic_path + ) if connections: - netlist["nets"].append({ - "name": net_name, - "connections": connections - }) + netlist["nets"].append( + {"name": net_name, "connections": connections} + ) - logger.info(f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components") + logger.info( + f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components" + ) return netlist except Exception as e: logger.error(f"Error generating netlist: {e}") return {"nets": [], "components": []} -if __name__ == '__main__': + +if __name__ == "__main__": # Example Usage (for testing) - from schematic import SchematicManager # Assuming schematic.py is in the same directory + from schematic import ( + SchematicManager, + ) # Assuming schematic.py is in the same directory # Create a new schematic test_sch = SchematicManager.create_schematic("ConnectionTestSchematic") diff --git a/python/commands/pin_locator.py b/python/commands/pin_locator.py index 1e91bb0..bf4367d 100644 --- a/python/commands/pin_locator.py +++ b/python/commands/pin_locator.py @@ -14,7 +14,7 @@ import sexpdata from sexpdata import Symbol from skip import Schematic -logger = logging.getLogger('kicad_interface') +logger = logging.getLogger("kicad_interface") class PinLocator: @@ -23,6 +23,7 @@ class PinLocator: def __init__(self): """Initialize pin locator with empty cache""" self.pin_definition_cache = {} # Cache: "lib_id:symbol_name" -> pin_data + self._schematic_cache: Dict[str, object] = {} # Cache: path -> loaded Schematic @staticmethod def parse_symbol_definition(symbol_def: list) -> Dict[str, Dict]: @@ -47,39 +48,39 @@ class PinLocator: return # Check if this is a pin definition - if len(sexp) > 0 and sexp[0] == Symbol('pin'): + if len(sexp) > 0 and sexp[0] == Symbol("pin"): # Pin format: (pin type shape (at x y angle) (length len) (name "name") (number "num")) pin_data = { - 'x': 0, - 'y': 0, - 'angle': 0, - 'length': 0, - 'name': '', - 'number': '', - 'type': str(sexp[1]) if len(sexp) > 1 else 'passive' + "x": 0, + "y": 0, + "angle": 0, + "length": 0, + "name": "", + "number": "", + "type": str(sexp[1]) if len(sexp) > 1 else "passive", } # Extract pin attributes for item in sexp: if isinstance(item, list) and len(item) > 0: - if item[0] == Symbol('at') and len(item) >= 3: - pin_data['x'] = float(item[1]) - pin_data['y'] = float(item[2]) + if item[0] == Symbol("at") and len(item) >= 3: + pin_data["x"] = float(item[1]) + pin_data["y"] = float(item[2]) if len(item) >= 4: - pin_data['angle'] = float(item[3]) + pin_data["angle"] = float(item[3]) - elif item[0] == Symbol('length') and len(item) >= 2: - pin_data['length'] = float(item[1]) + elif item[0] == Symbol("length") and len(item) >= 2: + pin_data["length"] = float(item[1]) - elif item[0] == Symbol('name') and len(item) >= 2: - pin_data['name'] = str(item[1]).strip('"') + elif item[0] == Symbol("name") and len(item) >= 2: + pin_data["name"] = str(item[1]).strip('"') - elif item[0] == Symbol('number') and len(item) >= 2: - pin_data['number'] = str(item[1]).strip('"') + elif item[0] == Symbol("number") and len(item) >= 2: + pin_data["number"] = str(item[1]).strip('"') # Store by pin number - if pin_data['number']: - pins[pin_data['number']] = pin_data + if pin_data["number"]: + pins[pin_data["number"]] = pin_data # Recurse into sublists for item in sexp: @@ -108,7 +109,7 @@ class PinLocator: try: # Read schematic - with open(schematic_path, 'r', encoding='utf-8') as f: + with open(schematic_path, "r", encoding="utf-8") as f: sch_content = f.read() sch_data = sexpdata.loads(sch_content) @@ -116,7 +117,11 @@ class PinLocator: # Find lib_symbols section lib_symbols = None for item in sch_data: - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('lib_symbols'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("lib_symbols") + ): lib_symbols = item break @@ -126,7 +131,11 @@ class PinLocator: # Find the specific symbol definition for item in lib_symbols[1:]: # Skip 'lib_symbols' itself - if isinstance(item, list) and len(item) > 1 and item[0] == Symbol('symbol'): + if ( + isinstance(item, list) + and len(item) > 1 + and item[0] == Symbol("symbol") + ): symbol_name = str(item[1]).strip('"') if symbol_name == lib_id: # Found the symbol, parse pins @@ -141,6 +150,7 @@ class PinLocator: except Exception as e: logger.error(f"Error getting symbol pins: {e}") import traceback + logger.error(traceback.format_exc()) return {} @@ -169,8 +179,9 @@ class PinLocator: return (rotated_x, rotated_y) - def get_pin_location(self, schematic_path: Path, symbol_reference: str, - pin_number: str) -> Optional[List[float]]: + def get_pin_location( + self, schematic_path: Path, symbol_reference: str, pin_number: str + ) -> Optional[List[float]]: """ Get the absolute location of a pin on a symbol instance @@ -184,7 +195,11 @@ class PinLocator: """ try: # Load schematic with kicad-skip to get symbol instance - sch = Schematic(str(schematic_path)) + # Use cache to avoid reloading the file for every pin lookup + sch_key = str(schematic_path) + if sch_key not in self._schematic_cache: + self._schematic_cache[sch_key] = Schematic(sch_key) + sch = self._schematic_cache[sch_key] # Find the symbol instance target_symbol = None @@ -205,12 +220,16 @@ class PinLocator: symbol_rotation = float(symbol_at[2]) if len(symbol_at) > 2 else 0.0 # Get symbol lib_id - lib_id = target_symbol.lib_id.value if hasattr(target_symbol, 'lib_id') else None + lib_id = ( + target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None + ) if not lib_id: logger.error(f"Symbol {symbol_reference} has no lib_id") return None - logger.debug(f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, lib_id={lib_id}") + logger.debug( + f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, lib_id={lib_id}" + ) # Get pin definitions for this symbol pins = self.get_symbol_pins(schematic_path, lib_id) @@ -220,36 +239,49 @@ class PinLocator: # Find the requested pin if pin_number not in pins: - logger.error(f"Pin {pin_number} not found on {symbol_reference}. Available pins: {list(pins.keys())}") + logger.error( + f"Pin {pin_number} not found on {symbol_reference}. Available pins: {list(pins.keys())}" + ) return None pin_data = pins[pin_number] # Get pin position relative to symbol origin - pin_rel_x = pin_data['x'] - pin_rel_y = pin_data['y'] + pin_rel_x = pin_data["x"] + pin_rel_y = pin_data["y"] - logger.debug(f"Pin {pin_number} relative position: ({pin_rel_x}, {pin_rel_y})") + logger.debug( + f"Pin {pin_number} relative position: ({pin_rel_x}, {pin_rel_y})" + ) # Apply symbol rotation to pin position if symbol_rotation != 0: - pin_rel_x, pin_rel_y = self.rotate_point(pin_rel_x, pin_rel_y, symbol_rotation) - logger.debug(f"After rotation {symbol_rotation}°: ({pin_rel_x}, {pin_rel_y})") + pin_rel_x, pin_rel_y = self.rotate_point( + pin_rel_x, pin_rel_y, symbol_rotation + ) + logger.debug( + f"After rotation {symbol_rotation}°: ({pin_rel_x}, {pin_rel_y})" + ) # Calculate absolute position abs_x = symbol_x + pin_rel_x abs_y = symbol_y + pin_rel_y - logger.info(f"Pin {symbol_reference}/{pin_number} located at ({abs_x}, {abs_y})") + logger.info( + f"Pin {symbol_reference}/{pin_number} located at ({abs_x}, {abs_y})" + ) return [abs_x, abs_y] except Exception as e: logger.error(f"Error getting pin location: {e}") import traceback + logger.error(traceback.format_exc()) return None - def get_all_symbol_pins(self, schematic_path: Path, symbol_reference: str) -> Dict[str, List[float]]: + def get_all_symbol_pins( + self, schematic_path: Path, symbol_reference: str + ) -> Dict[str, List[float]]: """ Get locations of all pins on a symbol instance @@ -261,8 +293,11 @@ class PinLocator: Dictionary mapping pin number -> [x, y] coordinates """ try: - # Load schematic - sch = Schematic(str(schematic_path)) + # Load schematic (use cache) + sch_key = str(schematic_path) + if sch_key not in self._schematic_cache: + self._schematic_cache[sch_key] = Schematic(sch_key) + sch = self._schematic_cache[sch_key] # Find symbol target_symbol = None @@ -276,7 +311,9 @@ class PinLocator: return {} # Get lib_id - lib_id = target_symbol.lib_id.value if hasattr(target_symbol, 'lib_id') else None + lib_id = ( + target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None + ) if not lib_id: logger.error(f"Symbol {symbol_reference} has no lib_id") return {} @@ -289,7 +326,9 @@ class PinLocator: # Calculate location for each pin result = {} for pin_num in pins.keys(): - location = self.get_pin_location(schematic_path, symbol_reference, pin_num) + location = self.get_pin_location( + schematic_path, symbol_reference, pin_num + ) if location: result[pin_num] = location @@ -301,10 +340,11 @@ class PinLocator: return {} -if __name__ == '__main__': +if __name__ == "__main__": # Test pin location discovery import sys - sys.path.insert(0, '/home/chris/MCP/KiCAD-MCP-Server/python') + + sys.path.insert(0, "/home/chris/MCP/KiCAD-MCP-Server/python") from pathlib import Path from commands.component_schematic import ComponentManager @@ -316,8 +356,10 @@ if __name__ == '__main__': print("=" * 80) # Create test schematic with components (cross-platform temp directory) - test_path = Path(tempfile.gettempdir()) / 'test_pin_locator.kicad_sch' - template_path = Path('/home/chris/MCP/KiCAD-MCP-Server/python/templates/template_with_symbols_expanded.kicad_sch') + test_path = Path(tempfile.gettempdir()) / "test_pin_locator.kicad_sch" + template_path = Path( + "/home/chris/MCP/KiCAD-MCP-Server/python/templates/template_with_symbols_expanded.kicad_sch" + ) shutil.copy(template_path, test_path) print(f"\n✓ Created test schematic: {test_path}") @@ -327,11 +369,25 @@ if __name__ == '__main__': sch = SchematicManager.load_schematic(str(test_path)) # Add resistor at (100, 100), rotation 0 - r1_def = {'type': 'R', 'reference': 'R1', 'value': '10k', 'x': 100, 'y': 100, 'rotation': 0} + r1_def = { + "type": "R", + "reference": "R1", + "value": "10k", + "x": 100, + "y": 100, + "rotation": 0, + } ComponentManager.add_component(sch, r1_def, test_path) # Add capacitor at (150, 100), rotation 90 - c1_def = {'type': 'C', 'reference': 'C1', 'value': '100nF', 'x': 150, 'y': 100, 'rotation': 90} + c1_def = { + "type": "C", + "reference": "C1", + "value": "100nF", + "x": 150, + "y": 100, + "rotation": 90, + } ComponentManager.add_component(sch, c1_def, test_path) SchematicManager.save_schematic(sch, str(test_path)) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 2beb33b..2587f61 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -19,19 +19,16 @@ from schemas.tool_schemas import TOOL_SCHEMAS from resources.resource_definitions import RESOURCE_DEFINITIONS, handle_resource_read # Configure logging -log_dir = os.path.join(os.path.expanduser('~'), '.kicad-mcp', 'logs') +log_dir = os.path.join(os.path.expanduser("~"), ".kicad-mcp", "logs") os.makedirs(log_dir, exist_ok=True) -log_file = os.path.join(log_dir, 'kicad_interface.log') +log_file = os.path.join(log_dir, "kicad_interface.log") logging.basicConfig( level=logging.DEBUG, - format='%(asctime)s [%(levelname)s] %(message)s', - handlers=[ - logging.FileHandler(log_file), - logging.StreamHandler(sys.stderr) - ] + format="%(asctime)s [%(levelname)s] %(message)s", + handlers=[logging.FileHandler(log_file), logging.StreamHandler(sys.stderr)], ) -logger = logging.getLogger('kicad_interface') +logger = logging.getLogger("kicad_interface") # Log Python environment details logger.info(f"Python version: {sys.version}") @@ -40,16 +37,13 @@ logger.info(f"Platform: {sys.platform}") logger.info(f"Working directory: {os.getcwd()}") # Windows-specific diagnostics -if sys.platform == 'win32': +if sys.platform == "win32": logger.info("=== Windows Environment Diagnostics ===") logger.info(f"PYTHONPATH: {os.environ.get('PYTHONPATH', 'NOT SET')}") logger.info(f"PATH: {os.environ.get('PATH', 'NOT SET')[:200]}...") # Truncate PATH # Check for common KiCAD installations - common_kicad_paths = [ - r"C:\Program Files\KiCad", - r"C:\Program Files (x86)\KiCad" - ] + common_kicad_paths = [r"C:\Program Files\KiCad", r"C:\Program Files (x86)\KiCad"] found_kicad = False for base_path in common_kicad_paths: @@ -57,10 +51,16 @@ if sys.platform == 'win32': logger.info(f"Found KiCAD installation at: {base_path}") # List versions try: - versions = [d for d in os.listdir(base_path) if os.path.isdir(os.path.join(base_path, d))] + versions = [ + d + for d in os.listdir(base_path) + if os.path.isdir(os.path.join(base_path, d)) + ] logger.info(f" Versions found: {', '.join(versions)}") for version in versions: - python_path = os.path.join(base_path, version, 'lib', 'python3', 'dist-packages') + python_path = os.path.join( + base_path, version, "lib", "python3", "dist-packages" + ) if os.path.exists(python_path): logger.info(f" ✓ Python path exists: {python_path}") found_kicad = True @@ -71,7 +71,9 @@ if sys.platform == 'win32': if not found_kicad: logger.warning("No KiCAD installations found in standard locations!") - logger.warning("Please ensure KiCAD 9.0+ is installed from https://www.kicad.org/download/windows/") + logger.warning( + "Please ensure KiCAD 9.0+ is installed from https://www.kicad.org/download/windows/" + ) logger.info("========================================") @@ -90,7 +92,9 @@ paths_added = PlatformHelper.add_kicad_to_python_path() if paths_added: logger.info("Successfully added KiCAD Python paths to sys.path") else: - logger.warning("No KiCAD Python paths found - attempting to import pcbnew from system path") + logger.warning( + "No KiCAD Python paths found - attempting to import pcbnew from system path" + ) logger.info(f"Current Python path: {sys.path}") @@ -108,7 +112,7 @@ logger.info(f"KiCAD backend preference: {KICAD_BACKEND}") USE_IPC_BACKEND = False ipc_backend = None -if KICAD_BACKEND in ('auto', 'ipc'): +if KICAD_BACKEND in ("auto", "ipc"): try: logger.info("Checking IPC backend availability...") from kicad_api.ipc_backend import IPCBackend @@ -129,11 +133,12 @@ if KICAD_BACKEND in ('auto', 'ipc'): ipc_backend = None # Fall back to SWIG backend if IPC not available -if not USE_IPC_BACKEND and KICAD_BACKEND != 'ipc': +if not USE_IPC_BACKEND and KICAD_BACKEND != "ipc": # Import KiCAD's Python API (SWIG) try: logger.info("Attempting to import pcbnew module (SWIG backend)...") import pcbnew # type: ignore + logger.info(f"Successfully imported pcbnew module from: {pcbnew.__file__}") logger.info(f"pcbnew version: {pcbnew.GetBuildVersion()}") logger.warning("Using SWIG backend - changes require manual reload in KiCAD UI") @@ -143,7 +148,7 @@ if not USE_IPC_BACKEND and KICAD_BACKEND != 'ipc': # Platform-specific help message help_message = "" - if sys.platform == 'win32': + if sys.platform == "win32": help_message = """ Windows Troubleshooting: 1. Verify KiCAD is installed: C:\\Program Files\\KiCad\\9.0 @@ -153,7 +158,7 @@ Windows Troubleshooting: 4. Log file location: %USERPROFILE%\\.kicad-mcp\\logs\\kicad_interface.log 5. Run setup-windows.ps1 for automatic configuration """ - elif sys.platform == 'darwin': + elif sys.platform == "darwin": help_message = """ macOS Troubleshooting: 1. Verify KiCAD is installed: /Applications/KiCad/KiCad.app @@ -173,7 +178,7 @@ Linux Troubleshooting: error_response = { "success": False, "message": "Failed to import pcbnew module - KiCAD Python API not found", - "errorDetails": f"Error: {str(e)}\n\n{help_message}\n\nPython sys.path:\n{chr(10).join(sys.path)}" + "errorDetails": f"Error: {str(e)}\n\n{help_message}\n\nPython sys.path:\n{chr(10).join(sys.path)}", } print(json.dumps(error_response)) sys.exit(1) @@ -183,17 +188,17 @@ Linux Troubleshooting: error_response = { "success": False, "message": "Error importing pcbnew module", - "errorDetails": str(e) + "errorDetails": str(e), } print(json.dumps(error_response)) sys.exit(1) # If IPC-only mode requested but not available, exit with error -elif KICAD_BACKEND == 'ipc' and not USE_IPC_BACKEND: +elif KICAD_BACKEND == "ipc" and not USE_IPC_BACKEND: error_response = { "success": False, "message": "IPC backend requested but not available", - "errorDetails": "KiCAD must be running with IPC API enabled. Enable at: Preferences > Plugins > Enable IPC API Server" + "errorDetails": "KiCAD must be running with IPC API enabled. Enable at: Preferences > Plugins > Enable IPC API Server", } print(json.dumps(error_response)) sys.exit(1) @@ -211,21 +216,26 @@ try: from commands.component_schematic import ComponentManager from commands.connection_schematic import ConnectionManager from commands.library_schematic import LibraryManager as SchematicLibraryManager - from commands.library import LibraryManager as FootprintLibraryManager, LibraryCommands + from commands.library import ( + LibraryManager as FootprintLibraryManager, + LibraryCommands, + ) from commands.library_symbol import SymbolLibraryManager, SymbolLibraryCommands from commands.jlcpcb import JLCPCBClient, test_jlcpcb_connection from commands.jlcpcb_parts import JLCPCBPartsManager + logger.info("Successfully imported all command handlers") except ImportError as e: logger.error(f"Failed to import command handlers: {e}") error_response = { "success": False, "message": "Failed to import command handlers", - "errorDetails": str(e) + "errorDetails": str(e), } print(json.dumps(error_response)) sys.exit(1) + class KiCADInterface: """Main interface class to handle KiCAD operations""" @@ -267,12 +277,13 @@ class KiCADInterface: # Initialize JLCPCB API integration self.jlcpcb_client = JLCPCBClient() # Official API (requires auth) from commands.jlcsearch import JLCSearchClient + self.jlcsearch_client = JLCSearchClient() # Public API (no auth required) self.jlcpcb_parts = JLCPCBPartsManager() # Schematic-related classes don't need board reference # as they operate directly on schematic files - + # Command routing dictionary self.command_routes = { # Project commands @@ -280,7 +291,6 @@ class KiCADInterface: "open_project": self.project_commands.open_project, "save_project": self.project_commands.save_project, "get_project_info": self.project_commands.get_project_info, - # Board commands "set_board_size": self.board_commands.set_board_size, "add_layer": self.board_commands.add_layer, @@ -293,7 +303,6 @@ class KiCADInterface: "add_mounting_hole": self.board_commands.add_mounting_hole, "add_text": self.board_commands.add_text, "add_board_text": self.board_commands.add_text, # Alias for TypeScript tool - # Component commands "place_component": self.component_commands.place_component, "move_component": self.component_commands.move_component, @@ -308,7 +317,6 @@ class KiCADInterface: "place_component_array": self.component_commands.place_component_array, "align_components": self.component_commands.align_components, "duplicate_component": self.component_commands.duplicate_component, - # Routing commands "add_net": self.routing_commands.add_net, "route_trace": self.routing_commands.route_trace, @@ -322,39 +330,33 @@ class KiCADInterface: "add_copper_pour": self.routing_commands.add_copper_pour, "route_differential_pair": self.routing_commands.route_differential_pair, "refill_zones": self._handle_refill_zones, - # Design rule commands "set_design_rules": self.design_rule_commands.set_design_rules, "get_design_rules": self.design_rule_commands.get_design_rules, "run_drc": self.design_rule_commands.run_drc, "get_drc_violations": self.design_rule_commands.get_drc_violations, - # Export commands "export_gerber": self.export_commands.export_gerber, "export_pdf": self.export_commands.export_pdf, "export_svg": self.export_commands.export_svg, "export_3d": self.export_commands.export_3d, "export_bom": self.export_commands.export_bom, - # Library commands (footprint management) "list_libraries": self.library_commands.list_libraries, "search_footprints": self.library_commands.search_footprints, "list_library_footprints": self.library_commands.list_library_footprints, "get_footprint_info": self.library_commands.get_footprint_info, - # Symbol library commands (local KiCad symbol library search) "list_symbol_libraries": self.symbol_library_commands.list_symbol_libraries, "search_symbols": self.symbol_library_commands.search_symbols, "list_library_symbols": self.symbol_library_commands.list_library_symbols, "get_symbol_info": self.symbol_library_commands.get_symbol_info, - # JLCPCB API commands (complete parts catalog via API) "download_jlcpcb_database": self._handle_download_jlcpcb_database, "search_jlcpcb_parts": self._handle_search_jlcpcb_parts, "get_jlcpcb_part": self._handle_get_jlcpcb_part, "get_jlcpcb_database_stats": self._handle_get_jlcpcb_database_stats, "suggest_jlcpcb_alternatives": self._handle_suggest_jlcpcb_alternatives, - # Schematic commands "create_schematic": self._handle_create_schematic, "load_schematic": self._handle_load_schematic, @@ -367,11 +369,9 @@ class KiCADInterface: "generate_netlist": self._handle_generate_netlist, "list_schematic_libraries": self._handle_list_schematic_libraries, "export_schematic_pdf": self._handle_export_schematic_pdf, - # UI/Process management commands "check_kicad_ui": self._handle_check_kicad_ui, "launch_kicad_ui": self._handle_launch_kicad_ui, - # IPC-specific commands (real-time operations) "get_backend_info": self._handle_get_backend_info, "ipc_add_track": self._handle_ipc_add_track, @@ -380,10 +380,12 @@ class KiCADInterface: "ipc_list_components": self._handle_ipc_list_components, "ipc_get_tracks": self._handle_ipc_get_tracks, "ipc_get_vias": self._handle_ipc_get_vias, - "ipc_save_board": self._handle_ipc_save_board + "ipc_save_board": self._handle_ipc_save_board, } - logger.info(f"KiCAD interface initialized (backend: {'IPC' if self.use_ipc else 'SWIG'})") + logger.info( + f"KiCAD interface initialized (backend: {'IPC' if self.use_ipc else 'SWIG'})" + ) # Commands that can be handled via IPC for real-time updates IPC_CAPABLE_COMMANDS = { @@ -422,7 +424,11 @@ class KiCADInterface: try: # Check if we can use IPC for this command (real-time UI sync) - if self.use_ipc and self.ipc_board_api and command in self.IPC_CAPABLE_COMMANDS: + if ( + self.use_ipc + and self.ipc_board_api + and command in self.IPC_CAPABLE_COMMANDS + ): ipc_handler_name = self.IPC_CAPABLE_COMMANDS[command] ipc_handler = getattr(self, ipc_handler_name, None) @@ -440,7 +446,9 @@ class KiCADInterface: # Fall back to SWIG-based handler if self.use_ipc and command in self.IPC_CAPABLE_COMMANDS: - logger.warning(f"IPC handler not available for {command}, falling back to SWIG (deprecated)") + logger.warning( + f"IPC handler not available for {command}, falling back to SWIG (deprecated)" + ) # Get the handler for the command handler = self.command_routes.get(command) @@ -469,7 +477,7 @@ class KiCADInterface: return { "success": False, "message": f"Unknown command: {command}", - "errorDetails": "The specified command is not supported" + "errorDetails": "The specified command is not supported", } except Exception as e: @@ -479,7 +487,7 @@ class KiCADInterface: return { "success": False, "message": f"Error handling command: {command}", - "errorDetails": f"{str(e)}\n{traceback_str}" + "errorDetails": f"{str(e)}\n{traceback_str}", } def _update_command_handlers(self): @@ -491,7 +499,7 @@ class KiCADInterface: self.routing_commands.board = self.board self.design_rule_commands.board = self.board self.export_commands.board = self.board - + # Schematic command handlers def _handle_create_schematic(self, params): """Create a new schematic""" @@ -502,16 +510,14 @@ class KiCADInterface: # - Python schema uses: filename, title # - Legacy uses: projectName, path, metadata project_name = ( - params.get("projectName") or - params.get("name") or - params.get("title") + params.get("projectName") or params.get("name") or params.get("title") ) # Handle filename parameter - it may contain full path filename = params.get("filename") if filename: # If filename provided, extract name and path from it - if filename.endswith('.kicad_sch'): + if filename.endswith(".kicad_sch"): filename = filename[:-10] # Remove .kicad_sch extension path = os.path.dirname(filename) or "." project_name = project_name or os.path.basename(filename) @@ -522,7 +528,7 @@ class KiCADInterface: if not project_name: return { "success": False, - "message": "Schematic name is required. Provide 'name', 'projectName', or 'filename' parameter." + "message": "Schematic name is required. Provide 'name', 'projectName', or 'filename' parameter.", } schematic = SchematicManager.create_schematic(project_name, metadata) @@ -533,19 +539,19 @@ class KiCADInterface: except Exception as e: logger.error(f"Error creating schematic: {str(e)}") return {"success": False, "message": str(e)} - + def _handle_load_schematic(self, params): """Load an existing schematic""" logger.info("Loading schematic") try: filename = params.get("filename") - + if not filename: return {"success": False, "message": "Filename is required"} - + schematic = SchematicManager.load_schematic(filename) success = schematic is not None - + if success: metadata = SchematicManager.get_schematic_metadata(schematic) return {"success": success, "metadata": metadata} @@ -554,7 +560,7 @@ class KiCADInterface: except Exception as e: logger.error(f"Error loading schematic: {str(e)}") return {"success": False, "message": str(e)} - + def _handle_add_schematic_component(self, params): """Add a component to a schematic using text-based injection (no sexpdata)""" logger.info("Adding component to schematic") @@ -570,27 +576,33 @@ class KiCADInterface: if not component: return {"success": False, "message": "Component definition is required"} - comp_type = component.get('type', 'R') - library = component.get('library', 'Device') - reference = component.get('reference', 'X?') - value = component.get('value', comp_type) - x = component.get('x', 0) - y = component.get('y', 0) + comp_type = component.get("type", "R") + library = component.get("library", "Device") + reference = component.get("reference", "X?") + value = component.get("value", comp_type) + x = component.get("x", 0) + y = component.get("y", 0) loader = DynamicSymbolLoader() loader.add_component( - Path(schematic_path), library, comp_type, - reference=reference, value=value, x=x, y=y + Path(schematic_path), + library, + comp_type, + reference=reference, + value=value, + x=x, + y=y, ) return { "success": True, "component_reference": reference, - "symbol_source": f"{library}:{comp_type}" + "symbol_source": f"{library}:{comp_type}", } except Exception as e: logger.error(f"Error adding component to schematic: {str(e)}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} @@ -609,11 +621,14 @@ class KiCADInterface: if not schematic_path: return {"success": False, "message": "Schematic path is required"} if not start_point or not end_point: - return {"success": False, "message": "Start and end points are required"} + return { + "success": False, + "message": "Start and end points are required", + } # Extract wire properties - stroke_width = properties.get('stroke_width', 0) - stroke_type = properties.get('stroke_type', 'default') + stroke_width = properties.get("stroke_width", 0) + stroke_type = properties.get("stroke_type", "default") # Use WireManager for S-expression manipulation success = WireManager.add_wire( @@ -621,7 +636,7 @@ class KiCADInterface: start_point, end_point, stroke_width=stroke_width, - stroke_type=stroke_type + stroke_type=stroke_type, ) if success: @@ -631,43 +646,57 @@ class KiCADInterface: except Exception as e: logger.error(f"Error adding wire to schematic: {str(e)}") import traceback + logger.error(traceback.format_exc()) - return {"success": False, "message": str(e), "errorDetails": traceback.format_exc()} - + return { + "success": False, + "message": str(e), + "errorDetails": traceback.format_exc(), + } + def _handle_list_schematic_libraries(self, params): """List available symbol libraries""" logger.info("Listing schematic libraries") try: search_paths = params.get("searchPaths") - + libraries = LibraryManager.list_available_libraries(search_paths) return {"success": True, "libraries": libraries} except Exception as e: logger.error(f"Error listing schematic libraries: {str(e)}") return {"success": False, "message": str(e)} - + def _handle_export_schematic_pdf(self, params): """Export schematic to PDF""" logger.info("Exporting schematic to PDF") try: schematic_path = params.get("schematicPath") output_path = params.get("outputPath") - + if not schematic_path: return {"success": False, "message": "Schematic path is required"} if not output_path: return {"success": False, "message": "Output path is required"} - + import subprocess + result = subprocess.run( - ["kicad-cli", "sch", "export", "pdf", "--output", output_path, schematic_path], - capture_output=True, - text=True + [ + "kicad-cli", + "sch", + "export", + "pdf", + "--output", + output_path, + schematic_path, + ], + capture_output=True, + text=True, ) - + success = result.returncode == 0 message = result.stderr if not success else "" - + return {"success": success, "message": message} except Exception as e: logger.error(f"Error exporting schematic to PDF: {str(e)}") @@ -684,9 +713,13 @@ class KiCADInterface: source_pin = params.get("sourcePin") target_ref = params.get("targetRef") target_pin = params.get("targetPin") - routing = params.get("routing", "direct") # 'direct', 'orthogonal_h', 'orthogonal_v' + routing = params.get( + "routing", "direct" + ) # 'direct', 'orthogonal_h', 'orthogonal_v' - if not all([schematic_path, source_ref, source_pin, target_ref, target_pin]): + if not all( + [schematic_path, source_ref, source_pin, target_ref, target_pin] + ): return {"success": False, "message": "Missing required parameters"} # Use ConnectionManager with new PinLocator and WireManager integration @@ -696,21 +729,26 @@ class KiCADInterface: source_pin, target_ref, target_pin, - routing=routing + routing=routing, ) if success: return { "success": True, - "message": f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})" + "message": f"Connected {source_ref}/{source_pin} to {target_ref}/{target_pin} (routing: {routing})", } else: return {"success": False, "message": "Failed to add connection"} except Exception as e: logger.error(f"Error adding schematic connection: {str(e)}") import traceback + logger.error(traceback.format_exc()) - return {"success": False, "message": str(e), "errorDetails": traceback.format_exc()} + return { + "success": False, + "message": str(e), + "errorDetails": traceback.format_exc(), + } def _handle_add_schematic_net_label(self, params): """Add a net label to schematic using WireManager""" @@ -722,7 +760,9 @@ class KiCADInterface: schematic_path = params.get("schematicPath") net_name = params.get("netName") position = params.get("position") - label_type = params.get("labelType", "label") # 'label', 'global_label', 'hierarchical_label' + label_type = params.get( + "labelType", "label" + ) # 'label', 'global_label', 'hierarchical_label' orientation = params.get("orientation", 0) # 0, 90, 180, 270 if not all([schematic_path, net_name, position]): @@ -734,18 +774,26 @@ class KiCADInterface: net_name, position, label_type=label_type, - orientation=orientation + orientation=orientation, ) if success: - return {"success": True, "message": f"Added net label '{net_name}' at {position}"} + return { + "success": True, + "message": f"Added net label '{net_name}' at {position}", + } else: return {"success": False, "message": "Failed to add net label"} except Exception as e: logger.error(f"Error adding net label: {str(e)}") import traceback + logger.error(traceback.format_exc()) - return {"success": False, "message": str(e), "errorDetails": traceback.format_exc()} + return { + "success": False, + "message": str(e), + "errorDetails": traceback.format_exc(), + } def _handle_connect_to_net(self, params): """Connect a component pin to a named net using wire stub and label""" @@ -763,24 +811,26 @@ class KiCADInterface: # Use ConnectionManager with new WireManager integration success = ConnectionManager.connect_to_net( - Path(schematic_path), - component_ref, - pin_name, - net_name + Path(schematic_path), component_ref, pin_name, net_name ) if success: return { "success": True, - "message": f"Connected {component_ref}/{pin_name} to net '{net_name}'" + "message": f"Connected {component_ref}/{pin_name} to net '{net_name}'", } else: return {"success": False, "message": "Failed to connect to net"} except Exception as e: logger.error(f"Error connecting to net: {str(e)}") import traceback + logger.error(traceback.format_exc()) - return {"success": False, "message": str(e), "errorDetails": traceback.format_exc()} + return { + "success": False, + "message": str(e), + "errorDetails": traceback.format_exc(), + } def _handle_get_net_connections(self, params): """Get all connections for a named net""" @@ -815,7 +865,9 @@ class KiCADInterface: if not schematic: return {"success": False, "message": "Failed to load schematic"} - netlist = ConnectionManager.generate_netlist(schematic) + netlist = ConnectionManager.generate_netlist( + schematic, schematic_path=schematic_path + ) return {"success": True, "netlist": netlist} except Exception as e: logger.error(f"Error generating netlist: {str(e)}") @@ -833,7 +885,7 @@ class KiCADInterface: "success": True, "running": is_running, "processes": processes, - "message": "KiCAD is running" if is_running else "KiCAD is not running" + "message": "KiCAD is running" if is_running else "KiCAD is not running", } except Exception as e: logger.error(f"Error checking KiCAD UI status: {str(e)}") @@ -848,14 +900,12 @@ class KiCADInterface: # Convert project path to Path object if provided from pathlib import Path + path_obj = Path(project_path) if project_path else None result = check_and_launch_kicad(path_obj, auto_launch) - return { - "success": True, - **result - } + return {"success": True, **result} except Exception as e: logger.error(f"Error launching KiCAD UI: {str(e)}") return {"success": False, "message": str(e)} @@ -868,7 +918,7 @@ class KiCADInterface: return { "success": False, "message": "No board is loaded", - "errorDetails": "Load or create a board first" + "errorDetails": "Load or create a board first", } # Use pcbnew's zone filler for SWIG backend @@ -879,7 +929,9 @@ class KiCADInterface: return { "success": True, "message": "Zones refilled successfully", - "zoneCount": zones.size() if hasattr(zones, 'size') else len(list(zones)) + "zoneCount": ( + zones.size() if hasattr(zones, "size") else len(list(zones)) + ), } except Exception as e: logger.error(f"Error refilling zones: {str(e)}") @@ -901,8 +953,16 @@ class KiCADInterface: net = params.get("net") # Handle both dict format and direct x/y - start_x = start.get("x", 0) if isinstance(start, dict) else params.get("startX", 0) - start_y = start.get("y", 0) if isinstance(start, dict) else params.get("startY", 0) + start_x = ( + start.get("x", 0) + if isinstance(start, dict) + else params.get("startX", 0) + ) + start_y = ( + start.get("y", 0) + if isinstance(start, dict) + else params.get("startY", 0) + ) end_x = end.get("x", 0) if isinstance(end, dict) else params.get("endX", 0) end_y = end.get("y", 0) if isinstance(end, dict) else params.get("endY", 0) @@ -913,19 +973,23 @@ class KiCADInterface: end_y=end_y, width=width, layer=layer, - net_name=net + net_name=net, ) return { "success": success, - "message": "Added trace (visible in KiCAD UI)" if success else "Failed to add trace", + "message": ( + "Added trace (visible in KiCAD UI)" + if success + else "Failed to add trace" + ), "trace": { "start": {"x": start_x, "y": start_y, "unit": "mm"}, "end": {"x": end_x, "y": end_y, "unit": "mm"}, "layer": layer, "width": width, - "net": net - } + "net": net, + }, } except Exception as e: logger.error(f"IPC route_trace error: {e}") @@ -935,8 +999,16 @@ class KiCADInterface: """IPC handler for add_via - adds via with real-time UI update""" try: position = params.get("position", {}) - x = position.get("x", 0) if isinstance(position, dict) else params.get("x", 0) - y = position.get("y", 0) if isinstance(position, dict) else params.get("y", 0) + x = ( + position.get("x", 0) + if isinstance(position, dict) + else params.get("x", 0) + ) + y = ( + position.get("y", 0) + if isinstance(position, dict) + else params.get("y", 0) + ) size = params.get("size", 0.8) drill = params.get("drill", 0.4) @@ -945,25 +1017,24 @@ class KiCADInterface: to_layer = params.get("to_layer", "B.Cu") success = self.ipc_board_api.add_via( - x=x, - y=y, - diameter=size, - drill=drill, - net_name=net, - via_type="through" + x=x, y=y, diameter=size, drill=drill, net_name=net, via_type="through" ) return { "success": success, - "message": "Added via (visible in KiCAD UI)" if success else "Failed to add via", + "message": ( + "Added via (visible in KiCAD UI)" + if success + else "Failed to add via" + ), "via": { "position": {"x": x, "y": y, "unit": "mm"}, "size": size, "drill": drill, "from_layer": from_layer, "to_layer": to_layer, - "net": net - } + "net": net, + }, } except Exception as e: logger.error(f"IPC add_via error: {e}") @@ -978,7 +1049,7 @@ class KiCADInterface: return { "success": True, "message": f"Net '{name}' will be created when components are connected", - "net": {"name": name} + "net": {"name": name}, } def _ipc_add_copper_pour(self, params): @@ -996,16 +1067,15 @@ class KiCADInterface: if not points or len(points) < 3: return { "success": False, - "message": "At least 3 points are required for copper pour outline" + "message": "At least 3 points are required for copper pour outline", } # Convert points format if needed (handle both {x, y} and {x, y, unit}) formatted_points = [] for point in points: - formatted_points.append({ - "x": point.get("x", 0), - "y": point.get("y", 0) - }) + formatted_points.append( + {"x": point.get("x", 0), "y": point.get("y", 0)} + ) success = self.ipc_board_api.add_zone( points=formatted_points, @@ -1015,12 +1085,16 @@ class KiCADInterface: min_thickness=min_width, priority=priority, fill_mode=fill_type, - name=name + name=name, ) return { "success": success, - "message": "Added copper pour (visible in KiCAD UI)" if success else "Failed to add copper pour", + "message": ( + "Added copper pour (visible in KiCAD UI)" + if success + else "Failed to add copper pour" + ), "pour": { "layer": layer, "net": net, @@ -1028,8 +1102,8 @@ class KiCADInterface: "minWidth": min_width, "priority": priority, "fillType": fill_type, - "pointCount": len(points) - } + "pointCount": len(points), + }, } except Exception as e: logger.error(f"IPC add_copper_pour error: {e}") @@ -1042,7 +1116,11 @@ class KiCADInterface: return { "success": success, - "message": "Zones refilled (visible in KiCAD UI)" if success else "Failed to refill zones" + "message": ( + "Zones refilled (visible in KiCAD UI)" + if success + else "Failed to refill zones" + ), } except Exception as e: logger.error(f"IPC refill_zones error: {e}") @@ -1053,24 +1131,31 @@ class KiCADInterface: try: text = params.get("text", "") position = params.get("position", {}) - x = position.get("x", 0) if isinstance(position, dict) else params.get("x", 0) - y = position.get("y", 0) if isinstance(position, dict) else params.get("y", 0) + x = ( + position.get("x", 0) + if isinstance(position, dict) + else params.get("x", 0) + ) + y = ( + position.get("y", 0) + if isinstance(position, dict) + else params.get("y", 0) + ) layer = params.get("layer", "F.SilkS") size = params.get("size", 1.0) rotation = params.get("rotation", 0) success = self.ipc_board_api.add_text( - text=text, - x=x, - y=y, - layer=layer, - size=size, - rotation=rotation + text=text, x=x, y=y, layer=layer, size=size, rotation=rotation ) return { "success": success, - "message": f"Added text '{text}' (visible in KiCAD UI)" if success else "Failed to add text" + "message": ( + f"Added text '{text}' (visible in KiCAD UI)" + if success + else "Failed to add text" + ), } except Exception as e: logger.error(f"IPC add_text error: {e}") @@ -1087,8 +1172,12 @@ class KiCADInterface: return { "success": success, - "message": f"Board size set to {width}x{height} {unit} (visible in KiCAD UI)" if success else "Failed to set board size", - "boardSize": {"width": width, "height": height, "unit": unit} + "message": ( + f"Board size set to {width}x{height} {unit} (visible in KiCAD UI)" + if success + else "Failed to set board size" + ), + "boardSize": {"width": width, "height": height, "unit": unit}, } except Exception as e: logger.error(f"IPC set_board_size error: {e}") @@ -1112,8 +1201,8 @@ class KiCADInterface: "viaCount": len(vias), "netCount": len(nets), "backend": "ipc", - "realtime": True - } + "realtime": True, + }, } except Exception as e: logger.error(f"IPC get_board_info error: {e}") @@ -1125,8 +1214,16 @@ class KiCADInterface: reference = params.get("reference", params.get("componentId", "")) footprint = params.get("footprint", "") position = params.get("position", {}) - x = position.get("x", 0) if isinstance(position, dict) else params.get("x", 0) - y = position.get("y", 0) if isinstance(position, dict) else params.get("y", 0) + x = ( + position.get("x", 0) + if isinstance(position, dict) + else params.get("x", 0) + ) + y = ( + position.get("y", 0) + if isinstance(position, dict) + else params.get("y", 0) + ) rotation = params.get("rotation", 0) layer = params.get("layer", "F.Cu") value = params.get("value", "") @@ -1138,19 +1235,23 @@ class KiCADInterface: y=y, rotation=rotation, layer=layer, - value=value + value=value, ) return { "success": success, - "message": f"Placed component {reference} (visible in KiCAD UI)" if success else "Failed to place component", + "message": ( + f"Placed component {reference} (visible in KiCAD UI)" + if success + else "Failed to place component" + ), "component": { "reference": reference, "footprint": footprint, "position": {"x": x, "y": y, "unit": "mm"}, "rotation": rotation, - "layer": layer - } + "layer": layer, + }, } except Exception as e: logger.error(f"IPC place_component error: {e}") @@ -1161,20 +1262,29 @@ class KiCADInterface: try: reference = params.get("reference", params.get("componentId", "")) position = params.get("position", {}) - x = position.get("x", 0) if isinstance(position, dict) else params.get("x", 0) - y = position.get("y", 0) if isinstance(position, dict) else params.get("y", 0) + x = ( + position.get("x", 0) + if isinstance(position, dict) + else params.get("x", 0) + ) + y = ( + position.get("y", 0) + if isinstance(position, dict) + else params.get("y", 0) + ) rotation = params.get("rotation") success = self.ipc_board_api.move_component( - reference=reference, - x=x, - y=y, - rotation=rotation + reference=reference, x=x, y=y, rotation=rotation ) return { "success": success, - "message": f"Moved component {reference} (visible in KiCAD UI)" if success else "Failed to move component" + "message": ( + f"Moved component {reference} (visible in KiCAD UI)" + if success + else "Failed to move component" + ), } except Exception as e: logger.error(f"IPC move_component error: {e}") @@ -1189,7 +1299,11 @@ class KiCADInterface: return { "success": success, - "message": f"Deleted component {reference} (visible in KiCAD UI)" if success else "Failed to delete component" + "message": ( + f"Deleted component {reference} (visible in KiCAD UI)" + if success + else "Failed to delete component" + ), } except Exception as e: logger.error(f"IPC delete_component error: {e}") @@ -1200,11 +1314,7 @@ class KiCADInterface: try: components = self.ipc_board_api.list_components() - return { - "success": True, - "components": components, - "count": len(components) - } + return {"success": True, "components": components, "count": len(components)} except Exception as e: logger.error(f"IPC get_component_list error: {e}") return {"success": False, "message": str(e)} @@ -1216,7 +1326,7 @@ class KiCADInterface: return { "success": success, - "message": "Project saved" if success else "Failed to save project" + "message": "Project saved" if success else "Failed to save project", } except Exception as e: logger.error(f"IPC save_project error: {e}") @@ -1226,7 +1336,9 @@ class KiCADInterface: """IPC handler for delete_trace - Note: IPC doesn't support direct trace deletion yet""" # IPC API doesn't have a direct delete track method # Fall back to SWIG for this operation - logger.info("delete_trace: Falling back to SWIG (IPC doesn't support trace deletion)") + logger.info( + "delete_trace: Falling back to SWIG (IPC doesn't support trace deletion)" + ) return self.routing_commands.delete_trace(params) def _ipc_get_nets_list(self, params): @@ -1234,11 +1346,7 @@ class KiCADInterface: try: nets = self.ipc_board_api.get_nets() - return { - "success": True, - "nets": nets, - "count": len(nets) - } + return {"success": True, "nets": nets, "count": len(nets)} except Exception as e: logger.error(f"IPC get_nets_list error: {e}") return {"success": False, "message": str(e)} @@ -1257,7 +1365,10 @@ class KiCADInterface: width = params.get("width", 0.1) if len(points) < 2: - return {"success": False, "message": "At least 2 points required for board outline"} + return { + "success": False, + "message": "At least 2 points required for board outline", + } commit = board.begin_commit() lines_created = 0 @@ -1268,8 +1379,12 @@ class KiCADInterface: end = points[(i + 1) % len(points)] # Wrap around to close the outline segment = BoardSegment() - segment.start = Vector2.from_xy(from_mm(start.get("x", 0)), from_mm(start.get("y", 0))) - segment.end = Vector2.from_xy(from_mm(end.get("x", 0)), from_mm(end.get("y", 0))) + segment.start = Vector2.from_xy( + from_mm(start.get("x", 0)), from_mm(start.get("y", 0)) + ) + segment.end = Vector2.from_xy( + from_mm(end.get("x", 0)), from_mm(end.get("y", 0)) + ) segment.layer = BoardLayer.BL_Edge_Cuts segment.attributes.stroke.width = from_mm(width) @@ -1281,7 +1396,7 @@ class KiCADInterface: return { "success": True, "message": f"Added board outline with {lines_created} segments (visible in KiCAD UI)", - "segments": lines_created + "segments": lines_created, } except Exception as e: logger.error(f"IPC add_board_outline error: {e}") @@ -1316,10 +1431,7 @@ class KiCADInterface: return { "success": True, "message": f"Added mounting hole at ({x}, {y}) mm (visible in KiCAD UI)", - "hole": { - "position": {"x": x, "y": y}, - "diameter": diameter - } + "hole": {"position": {"x": x, "y": y}, "diameter": diameter}, } except Exception as e: logger.error(f"IPC add_mounting_hole error: {e}") @@ -1330,11 +1442,7 @@ class KiCADInterface: try: layers = self.ipc_board_api.get_enabled_layers() - return { - "success": True, - "layers": layers, - "count": len(layers) - } + return {"success": True, "layers": layers, "count": len(layers)} except Exception as e: logger.error(f"IPC get_layer_list error: {e}") return {"success": False, "message": str(e)} @@ -1365,13 +1473,17 @@ class KiCADInterface: reference=reference, x=target.get("position", {}).get("x", 0), y=target.get("position", {}).get("y", 0), - rotation=new_rotation + rotation=new_rotation, ) return { "success": success, - "message": f"Rotated component {reference} by {angle}° (visible in KiCAD UI)" if success else "Failed to rotate component", - "newRotation": new_rotation + "message": ( + f"Rotated component {reference} by {angle}° (visible in KiCAD UI)" + if success + else "Failed to rotate component" + ), + "newRotation": new_rotation, } except Exception as e: logger.error(f"IPC rotate_component error: {e}") @@ -1392,10 +1504,7 @@ class KiCADInterface: if not target: return {"success": False, "message": f"Component {reference} not found"} - return { - "success": True, - "component": target - } + return {"success": True, "component": target} except Exception as e: logger.error(f"IPC get_component_properties error: {e}") return {"success": False, "message": str(e)} @@ -1410,9 +1519,15 @@ class KiCADInterface: "success": True, "backend": "ipc" if self.use_ipc else "swig", "realtime_sync": self.use_ipc, - "ipc_connected": self.ipc_backend.is_connected() if self.ipc_backend else False, + "ipc_connected": ( + self.ipc_backend.is_connected() if self.ipc_backend else False + ), "version": self.ipc_backend.get_version() if self.ipc_backend else "N/A", - "message": "Using IPC backend with real-time UI sync" if self.use_ipc else "Using SWIG backend (requires manual reload)" + "message": ( + "Using IPC backend with real-time UI sync" + if self.use_ipc + else "Using SWIG backend (requires manual reload)" + ), } def _handle_ipc_add_track(self, params): @@ -1428,12 +1543,16 @@ class KiCADInterface: end_y=params.get("endY", 0), width=params.get("width", 0.25), layer=params.get("layer", "F.Cu"), - net_name=params.get("net") + net_name=params.get("net"), ) return { "success": success, - "message": "Track added (visible in KiCAD UI)" if success else "Failed to add track", - "realtime": True + "message": ( + "Track added (visible in KiCAD UI)" + if success + else "Failed to add track" + ), + "realtime": True, } except Exception as e: logger.error(f"Error adding track via IPC: {e}") @@ -1451,12 +1570,16 @@ class KiCADInterface: diameter=params.get("diameter", 0.8), drill=params.get("drill", 0.4), net_name=params.get("net"), - via_type=params.get("type", "through") + via_type=params.get("type", "through"), ) return { "success": success, - "message": "Via added (visible in KiCAD UI)" if success else "Failed to add via", - "realtime": True + "message": ( + "Via added (visible in KiCAD UI)" + if success + else "Failed to add via" + ), + "realtime": True, } except Exception as e: logger.error(f"Error adding via via IPC: {e}") @@ -1474,12 +1597,16 @@ class KiCADInterface: y=params.get("y", 0), layer=params.get("layer", "F.SilkS"), size=params.get("size", 1.0), - rotation=params.get("rotation", 0) + rotation=params.get("rotation", 0), ) return { "success": success, - "message": "Text added (visible in KiCAD UI)" if success else "Failed to add text", - "realtime": True + "message": ( + "Text added (visible in KiCAD UI)" + if success + else "Failed to add text" + ), + "realtime": True, } except Exception as e: logger.error(f"Error adding text via IPC: {e}") @@ -1492,11 +1619,7 @@ class KiCADInterface: try: components = self.ipc_board_api.list_components() - return { - "success": True, - "components": components, - "count": len(components) - } + return {"success": True, "components": components, "count": len(components)} except Exception as e: logger.error(f"Error listing components via IPC: {e}") return {"success": False, "message": str(e)} @@ -1508,11 +1631,7 @@ class KiCADInterface: try: tracks = self.ipc_board_api.get_tracks() - return { - "success": True, - "tracks": tracks, - "count": len(tracks) - } + return {"success": True, "tracks": tracks, "count": len(tracks)} except Exception as e: logger.error(f"Error getting tracks via IPC: {e}") return {"success": False, "message": str(e)} @@ -1524,11 +1643,7 @@ class KiCADInterface: try: vias = self.ipc_board_api.get_vias() - return { - "success": True, - "vias": vias, - "count": len(vias) - } + return {"success": True, "vias": vias, "count": len(vias)} except Exception as e: logger.error(f"Error getting vias via IPC: {e}") return {"success": False, "message": str(e)} @@ -1542,7 +1657,7 @@ class KiCADInterface: success = self.ipc_board_api.save() return { "success": success, - "message": "Board saved" if success else "Failed to save board" + "message": "Board saved" if success else "Failed to save board", } except Exception as e: logger.error(f"Error saving board via IPC: {e}") @@ -1553,16 +1668,17 @@ class KiCADInterface: def _handle_download_jlcpcb_database(self, params): """Download JLCPCB parts database from JLCSearch API""" try: - force = params.get('force', False) + force = params.get("force", False) # Check if database exists import os + stats = self.jlcpcb_parts.get_database_stats() - if stats['total_parts'] > 0 and not force: + if stats["total_parts"] > 0 and not force: return { "success": False, "message": "Database already exists. Use force=true to re-download.", - "stats": stats + "stats": stats, } logger.info("Downloading JLCPCB parts database from JLCSearch...") @@ -1575,8 +1691,7 @@ class KiCADInterface: # Import into database logger.info(f"Importing {len(parts)} parts into database...") self.jlcpcb_parts.import_jlcsearch_parts( - parts, - progress_callback=lambda curr, total, msg: logger.info(msg) + parts, progress_callback=lambda curr, total, msg: logger.info(msg) ) # Get final stats @@ -1587,33 +1702,33 @@ class KiCADInterface: return { "success": True, - "total_parts": stats['total_parts'], - "basic_parts": stats['basic_parts'], - "extended_parts": stats['extended_parts'], + "total_parts": stats["total_parts"], + "basic_parts": stats["basic_parts"], + "extended_parts": stats["extended_parts"], "db_size_mb": round(db_size_mb, 2), - "db_path": stats['db_path'] + "db_path": stats["db_path"], } except Exception as e: logger.error(f"Error downloading JLCPCB database: {e}", exc_info=True) return { "success": False, - "message": f"Failed to download database: {str(e)}" + "message": f"Failed to download database: {str(e)}", } def _handle_search_jlcpcb_parts(self, params): """Search JLCPCB parts database""" try: - query = params.get('query') - category = params.get('category') - package = params.get('package') - library_type = params.get('library_type', 'All') - manufacturer = params.get('manufacturer') - in_stock = params.get('in_stock', True) - limit = params.get('limit', 20) + query = params.get("query") + category = params.get("category") + package = params.get("package") + library_type = params.get("library_type", "All") + manufacturer = params.get("manufacturer") + in_stock = params.get("in_stock", True) + limit = params.get("limit", 20) # Adjust library_type filter - if library_type == 'All': + if library_type == "All": library_type = None parts = self.jlcpcb_parts.search_parts( @@ -1623,97 +1738,72 @@ class KiCADInterface: library_type=library_type, manufacturer=manufacturer, in_stock=in_stock, - limit=limit + limit=limit, ) # Add price breaks and footprints to each part for part in parts: - if part.get('price_json'): + if part.get("price_json"): try: - part['price_breaks'] = json.loads(part['price_json']) + part["price_breaks"] = json.loads(part["price_json"]) except: - part['price_breaks'] = [] + part["price_breaks"] = [] - return { - "success": True, - "parts": parts, - "count": len(parts) - } + return {"success": True, "parts": parts, "count": len(parts)} except Exception as e: logger.error(f"Error searching JLCPCB parts: {e}", exc_info=True) - return { - "success": False, - "message": f"Search failed: {str(e)}" - } + return {"success": False, "message": f"Search failed: {str(e)}"} def _handle_get_jlcpcb_part(self, params): """Get detailed information for a specific JLCPCB part""" try: - lcsc_number = params.get('lcsc_number') + lcsc_number = params.get("lcsc_number") if not lcsc_number: - return { - "success": False, - "message": "Missing lcsc_number parameter" - } + return {"success": False, "message": "Missing lcsc_number parameter"} part = self.jlcpcb_parts.get_part_info(lcsc_number) if not part: - return { - "success": False, - "message": f"Part not found: {lcsc_number}" - } + return {"success": False, "message": f"Part not found: {lcsc_number}"} # Get suggested KiCAD footprints - footprints = self.jlcpcb_parts.map_package_to_footprint(part.get('package', '')) + footprints = self.jlcpcb_parts.map_package_to_footprint( + part.get("package", "") + ) - return { - "success": True, - "part": part, - "footprints": footprints - } + return {"success": True, "part": part, "footprints": footprints} except Exception as e: logger.error(f"Error getting JLCPCB part: {e}", exc_info=True) - return { - "success": False, - "message": f"Failed to get part info: {str(e)}" - } + return {"success": False, "message": f"Failed to get part info: {str(e)}"} def _handle_get_jlcpcb_database_stats(self, params): """Get statistics about JLCPCB database""" try: stats = self.jlcpcb_parts.get_database_stats() - return { - "success": True, - "stats": stats - } + return {"success": True, "stats": stats} except Exception as e: logger.error(f"Error getting database stats: {e}", exc_info=True) - return { - "success": False, - "message": f"Failed to get stats: {str(e)}" - } + return {"success": False, "message": f"Failed to get stats: {str(e)}"} def _handle_suggest_jlcpcb_alternatives(self, params): """Suggest alternative JLCPCB parts""" try: - lcsc_number = params.get('lcsc_number') - limit = params.get('limit', 5) + lcsc_number = params.get("lcsc_number") + limit = params.get("limit", 5) if not lcsc_number: - return { - "success": False, - "message": "Missing lcsc_number parameter" - } + return {"success": False, "message": "Missing lcsc_number parameter"} # Get original part for price comparison original_part = self.jlcpcb_parts.get_part_info(lcsc_number) reference_price = None - if original_part and original_part.get('price_breaks'): + if original_part and original_part.get("price_breaks"): try: - reference_price = float(original_part['price_breaks'][0].get('price', 0)) + reference_price = float( + original_part["price_breaks"][0].get("price", 0) + ) except: pass @@ -1721,23 +1811,23 @@ class KiCADInterface: # Add price breaks to alternatives for part in alternatives: - if part.get('price_json'): + if part.get("price_json"): try: - part['price_breaks'] = json.loads(part['price_json']) + part["price_breaks"] = json.loads(part["price_json"]) except: - part['price_breaks'] = [] + part["price_breaks"] = [] return { "success": True, "alternatives": alternatives, - "reference_price": reference_price + "reference_price": reference_price, } except Exception as e: logger.error(f"Error suggesting alternatives: {e}", exc_info=True) return { "success": False, - "message": f"Failed to suggest alternatives: {str(e)}" + "message": f"Failed to suggest alternatives: {str(e)}", } @@ -1756,38 +1846,36 @@ def main(): command_data = json.loads(line) # Check if this is JSON-RPC 2.0 format - if 'jsonrpc' in command_data and command_data['jsonrpc'] == '2.0': + if "jsonrpc" in command_data and command_data["jsonrpc"] == "2.0": logger.info("Detected JSON-RPC 2.0 format message") - method = command_data.get('method') - params = command_data.get('params', {}) - request_id = command_data.get('id') + method = command_data.get("method") + params = command_data.get("params", {}) + request_id = command_data.get("id") # Handle MCP protocol methods - if method == 'initialize': + if method == "initialize": logger.info("Handling MCP initialize") response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'result': { - 'protocolVersion': '2025-06-18', - 'capabilities': { - 'tools': { - 'listChanged': True + "jsonrpc": "2.0", + "id": request_id, + "result": { + "protocolVersion": "2025-06-18", + "capabilities": { + "tools": {"listChanged": True}, + "resources": { + "subscribe": False, + "listChanged": True, }, - 'resources': { - 'subscribe': False, - 'listChanged': True - } }, - 'serverInfo': { - 'name': 'kicad-mcp-server', - 'title': 'KiCAD PCB Design Assistant', - 'version': '2.1.0-alpha' + "serverInfo": { + "name": "kicad-mcp-server", + "title": "KiCAD PCB Design Assistant", + "version": "2.1.0-alpha", }, - 'instructions': 'AI-assisted PCB design with KiCAD. Use tools to create projects, design boards, place components, route traces, and export manufacturing files.' - } + "instructions": "AI-assisted PCB design with KiCAD. Use tools to create projects, design boards, place components, route traces, and export manufacturing files.", + }, } - elif method == 'tools/list': + elif method == "tools/list": logger.info("Handling MCP tools/list") # Return list of available tools with proper schemas tools = [] @@ -1798,85 +1886,84 @@ def main(): tools.append(tool_def) else: # Fallback for tools without schemas - logger.warning(f"No schema defined for tool: {cmd_name}") - tools.append({ - 'name': cmd_name, - 'description': f'KiCAD command: {cmd_name}', - 'inputSchema': { - 'type': 'object', - 'properties': {} + logger.warning( + f"No schema defined for tool: {cmd_name}" + ) + tools.append( + { + "name": cmd_name, + "description": f"KiCAD command: {cmd_name}", + "inputSchema": { + "type": "object", + "properties": {}, + }, } - }) + ) logger.info(f"Returning {len(tools)} tools") response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'result': { - 'tools': tools - } + "jsonrpc": "2.0", + "id": request_id, + "result": {"tools": tools}, } - elif method == 'tools/call': + elif method == "tools/call": logger.info("Handling MCP tools/call") - tool_name = params.get('name') - tool_params = params.get('arguments', {}) + tool_name = params.get("name") + tool_params = params.get("arguments", {}) # Execute the command result = interface.handle_command(tool_name, tool_params) response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'result': { - 'content': [ - { - 'type': 'text', - 'text': json.dumps(result) - } + "jsonrpc": "2.0", + "id": request_id, + "result": { + "content": [ + {"type": "text", "text": json.dumps(result)} ] - } + }, } - elif method == 'resources/list': + elif method == "resources/list": logger.info("Handling MCP resources/list") # Return list of available resources response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'result': { - 'resources': RESOURCE_DEFINITIONS - } + "jsonrpc": "2.0", + "id": request_id, + "result": {"resources": RESOURCE_DEFINITIONS}, } - elif method == 'resources/read': + elif method == "resources/read": logger.info("Handling MCP resources/read") - resource_uri = params.get('uri') + resource_uri = params.get("uri") if not resource_uri: response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'error': { - 'code': -32602, - 'message': 'Missing required parameter: uri' - } + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": -32602, + "message": "Missing required parameter: uri", + }, } else: # Read the resource - resource_data = handle_resource_read(resource_uri, interface) + resource_data = handle_resource_read( + resource_uri, interface + ) response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'result': resource_data + "jsonrpc": "2.0", + "id": request_id, + "result": resource_data, } else: logger.error(f"Unknown JSON-RPC method: {method}") response = { - 'jsonrpc': '2.0', - 'id': request_id, - 'error': { - 'code': -32601, - 'message': f'Method not found: {method}' - } + "jsonrpc": "2.0", + "id": request_id, + "error": { + "code": -32601, + "message": f"Method not found: {method}", + }, } else: # Handle legacy custom format @@ -1889,7 +1976,7 @@ def main(): response = { "success": False, "message": "Missing command", - "errorDetails": "The command field is required" + "errorDetails": "The command field is required", } else: # Handle command @@ -1905,7 +1992,7 @@ def main(): response = { "success": False, "message": "Invalid JSON input", - "errorDetails": str(e) + "errorDetails": str(e), } print(json.dumps(response)) sys.stdout.flush() @@ -1918,5 +2005,6 @@ def main(): logger.error(f"Unexpected error: {str(e)}\n{traceback.format_exc()}") sys.exit(1) + if __name__ == "__main__": main() diff --git a/src/server.ts b/src/server.ts index af1db1b..3a83ba5 100644 --- a/src/server.ts +++ b/src/server.ts @@ -2,46 +2,46 @@ * KiCAD MCP Server implementation */ -import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; -import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'; -import express from 'express'; -import { spawn, exec, execSync, ChildProcess } from 'child_process'; -import { existsSync } from 'fs'; -import { join, dirname } from 'path'; -import { logger } from './logger.js'; +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; +import express from "express"; +import { spawn, exec, execSync, ChildProcess } from "child_process"; +import { existsSync } from "fs"; +import { join, dirname } from "path"; +import { logger } from "./logger.js"; // Import tool registration functions -import { registerProjectTools } from './tools/project.js'; -import { registerBoardTools } from './tools/board.js'; -import { registerComponentTools } from './tools/component.js'; -import { registerRoutingTools } from './tools/routing.js'; -import { registerDesignRuleTools } from './tools/design-rules.js'; -import { registerExportTools } from './tools/export.js'; -import { registerSchematicTools } from './tools/schematic.js'; -import { registerLibraryTools } from './tools/library.js'; -import { registerSymbolLibraryTools } from './tools/library-symbol.js'; -import { registerJLCPCBApiTools } from './tools/jlcpcb-api.js'; -import { registerUITools } from './tools/ui.js'; -import { registerRouterTools } from './tools/router.js'; +import { registerProjectTools } from "./tools/project.js"; +import { registerBoardTools } from "./tools/board.js"; +import { registerComponentTools } from "./tools/component.js"; +import { registerRoutingTools } from "./tools/routing.js"; +import { registerDesignRuleTools } from "./tools/design-rules.js"; +import { registerExportTools } from "./tools/export.js"; +import { registerSchematicTools } from "./tools/schematic.js"; +import { registerLibraryTools } from "./tools/library.js"; +import { registerSymbolLibraryTools } from "./tools/library-symbol.js"; +import { registerJLCPCBApiTools } from "./tools/jlcpcb-api.js"; +import { registerUITools } from "./tools/ui.js"; +import { registerRouterTools } from "./tools/router.js"; // Import resource registration functions -import { registerProjectResources } from './resources/project.js'; -import { registerBoardResources } from './resources/board.js'; -import { registerComponentResources } from './resources/component.js'; -import { registerLibraryResources } from './resources/library.js'; +import { registerProjectResources } from "./resources/project.js"; +import { registerBoardResources } from "./resources/board.js"; +import { registerComponentResources } from "./resources/component.js"; +import { registerLibraryResources } from "./resources/library.js"; // Import prompt registration functions -import { registerComponentPrompts } from './prompts/component.js'; -import { registerRoutingPrompts } from './prompts/routing.js'; -import { registerDesignPrompts } from './prompts/design.js'; +import { registerComponentPrompts } from "./prompts/component.js"; +import { registerRoutingPrompts } from "./prompts/routing.js"; +import { registerDesignPrompts } from "./prompts/design.js"; /** * Find the Python executable to use * Prioritizes virtual environment if available, falls back to system Python */ function findPythonExecutable(scriptPath: string): string { - const isWindows = process.platform === 'win32'; - const isMac = process.platform === 'darwin'; + const isWindows = process.platform === "win32"; + const isMac = process.platform === "darwin"; const isLinux = !isWindows && !isMac; // Get the project root (parent of the python/ directory) @@ -49,8 +49,18 @@ function findPythonExecutable(scriptPath: string): string { // Check for virtual environment const venvPaths = [ - join(projectRoot, 'venv', isWindows ? 'Scripts' : 'bin', isWindows ? 'python.exe' : 'python'), - join(projectRoot, '.venv', isWindows ? 'Scripts' : 'bin', isWindows ? 'python.exe' : 'python'), + join( + projectRoot, + "venv", + isWindows ? "Scripts" : "bin", + isWindows ? "python.exe" : "python", + ), + join( + projectRoot, + ".venv", + isWindows ? "Scripts" : "bin", + isWindows ? "python.exe" : "python", + ), ]; for (const venvPath of venvPaths) { @@ -62,27 +72,29 @@ function findPythonExecutable(scriptPath: string): string { // Allow override via KICAD_PYTHON environment variable (any platform) if (process.env.KICAD_PYTHON) { - logger.info(`Using KICAD_PYTHON environment variable: ${process.env.KICAD_PYTHON}`); + logger.info( + `Using KICAD_PYTHON environment variable: ${process.env.KICAD_PYTHON}`, + ); return process.env.KICAD_PYTHON; } // Platform-specific KiCAD bundled Python detection - if (isWindows && process.env.PYTHONPATH?.includes('KiCad')) { - // Windows: Try KiCAD's bundled Python - const kicadPython = 'C:\\Program Files\\KiCad\\9.0\\bin\\python.exe'; + if (isWindows) { + // Windows: Always prefer KiCAD's bundled Python (pcbnew.pyd is compiled for it) + const kicadPython = "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe"; if (existsSync(kicadPython)) { logger.info(`Found KiCAD bundled Python at: ${kicadPython}`); return kicadPython; } } else if (isMac) { // macOS: Try KiCAD's bundled Python (check multiple versions and locations) - const kicadPythonVersions = ['3.9', '3.10', '3.11', '3.12', '3.13']; + const kicadPythonVersions = ["3.9", "3.10", "3.11", "3.12", "3.13"]; // Standard KiCAD installation paths const kicadAppPaths = [ - '/Applications/KiCad/KiCad.app', - '/Applications/KiCAD/KiCad.app', // Alternative capitalization - `${process.env.HOME}/Applications/KiCad/KiCad.app`, // User Applications folder + "/Applications/KiCad/KiCad.app", + "/Applications/KiCAD/KiCad.app", // Alternative capitalization + `${process.env.HOME}/Applications/KiCad/KiCad.app`, // User Applications folder ]; // Check all KiCAD app locations with all Python versions @@ -98,24 +110,26 @@ function findPythonExecutable(scriptPath: string): string { // Fallback to Homebrew Python (if pcbnew is installed via pip) const homebrewPaths = [ - '/opt/homebrew/bin/python3', // Apple Silicon - '/usr/local/bin/python3', // Intel Mac - '/opt/homebrew/bin/python3.12', - '/opt/homebrew/bin/python3.11', + "/opt/homebrew/bin/python3", // Apple Silicon + "/usr/local/bin/python3", // Intel Mac + "/opt/homebrew/bin/python3.12", + "/opt/homebrew/bin/python3.11", ]; for (const path of homebrewPaths) { if (existsSync(path)) { - logger.info(`Found Homebrew Python at: ${path} (ensure pcbnew is importable)`); + logger.info( + `Found Homebrew Python at: ${path} (ensure pcbnew is importable)`, + ); return path; } } } else if (isLinux) { // Linux: Try KiCAD bundled Python locations first const linuxKicadPaths = [ - '/usr/lib/kicad/bin/python3', - '/usr/local/lib/kicad/bin/python3', - '/opt/kicad/bin/python3', + "/usr/lib/kicad/bin/python3", + "/usr/local/lib/kicad/bin/python3", + "/opt/kicad/bin/python3", ]; for (const path of linuxKicadPaths) { @@ -127,17 +141,17 @@ function findPythonExecutable(scriptPath: string): string { // Resolve system python3 to full path using 'which' try { - const result = execSync('which python3', { encoding: 'utf-8' }).trim(); + const result = execSync("which python3", { encoding: "utf-8" }).trim(); if (result && existsSync(result)) { logger.info(`Resolved system Python via which: ${result}`); return result; } } catch (e) { - logger.warn('Failed to resolve python3 via which command'); + logger.warn("Failed to resolve python3 via which command"); } // Fallback to common system paths - const systemPaths = ['/usr/bin/python3', '/bin/python3']; + const systemPaths = ["/usr/bin/python3", "/bin/python3"]; for (const path of systemPaths) { if (existsSync(path)) { logger.info(`Found system Python at: ${path}`); @@ -147,8 +161,8 @@ function findPythonExecutable(scriptPath: string): string { } // Default to system Python (last resort) - logger.info('Using system Python (no venv found)'); - return isWindows ? 'python.exe' : 'python3'; + logger.info("Using system Python (no venv found)"); + return isWindows ? "python.exe" : "python3"; } /** @@ -159,11 +173,19 @@ export class KiCADMcpServer { private pythonProcess: ChildProcess | null = null; private kicadScriptPath: string; private stdioTransport!: StdioServerTransport; - private requestQueue: Array<{ request: any, resolve: Function, reject: Function }> = []; + private requestQueue: Array<{ + request: any; + resolve: Function; + reject: Function; + }> = []; private processingRequest = false; - private responseBuffer: string = ''; - private currentRequestHandler: { resolve: Function, reject: Function, timeoutHandle: NodeJS.Timeout } | null = null; - + private responseBuffer: string = ""; + private currentRequestHandler: { + resolve: Function; + reject: Function; + timeoutHandle: NodeJS.Timeout; + } | null = null; + /** * Constructor for the KiCAD MCP Server * @param kicadScriptPath Path to the Python KiCAD interface script @@ -171,37 +193,39 @@ export class KiCADMcpServer { */ constructor( kicadScriptPath: string, - logLevel: 'error' | 'warn' | 'info' | 'debug' = 'info' + logLevel: "error" | "warn" | "info" | "debug" = "info", ) { // Set up the logger logger.setLogLevel(logLevel); - + // Check if KiCAD script exists this.kicadScriptPath = kicadScriptPath; if (!existsSync(this.kicadScriptPath)) { - throw new Error(`KiCAD interface script not found: ${this.kicadScriptPath}`); + throw new Error( + `KiCAD interface script not found: ${this.kicadScriptPath}`, + ); } - + // Initialize the MCP server this.server = new McpServer({ - name: 'kicad-mcp-server', - version: '1.0.0', - description: 'MCP server for KiCAD PCB design operations' + name: "kicad-mcp-server", + version: "1.0.0", + description: "MCP server for KiCAD PCB design operations", }); - + // Initialize STDIO transport this.stdioTransport = new StdioServerTransport(); - logger.info('Using STDIO transport for local communication'); - + logger.info("Using STDIO transport for local communication"); + // Register tools, resources, and prompts this.registerAll(); } - + /** * Register all tools, resources, and prompts */ private registerAll(): void { - logger.info('Registering KiCAD tools, resources, and prompts...'); + logger.info("Registering KiCAD tools, resources, and prompts..."); // Register router tools FIRST (for tool discovery and execution) registerRouterTools(this.server, this.callKicadScript.bind(this)); @@ -230,20 +254,26 @@ export class KiCADMcpServer { registerRoutingPrompts(this.server); registerDesignPrompts(this.server); - logger.info('All KiCAD tools, resources, and prompts registered'); - logger.info('Router pattern enabled: 4 router tools + direct tools for discovery'); + logger.info("All KiCAD tools, resources, and prompts registered"); + logger.info( + "Router pattern enabled: 4 router tools + direct tools for discovery", + ); } - + /** * Validate prerequisites before starting the server */ private async validatePrerequisites(pythonExe: string): Promise { - const isWindows = process.platform === 'win32'; - const isLinux = process.platform !== 'win32' && process.platform !== 'darwin'; + const isWindows = process.platform === "win32"; + const isLinux = + process.platform !== "win32" && process.platform !== "darwin"; const errors: string[] = []; // Check if Python executable exists (for absolute paths) or is executable (for commands) - const isAbsolutePath = pythonExe.startsWith('/') || pythonExe.startsWith('C:') || pythonExe.startsWith('\\'); + const isAbsolutePath = + pythonExe.startsWith("/") || + pythonExe.startsWith("C:") || + pythonExe.startsWith("\\"); if (isAbsolutePath) { // Absolute path: use existsSync @@ -251,42 +281,61 @@ export class KiCADMcpServer { errors.push(`Python executable not found: ${pythonExe}`); if (isWindows) { - errors.push('Windows: Install KiCAD 9.0+ from https://www.kicad.org/download/windows/'); - errors.push('Or run: .\\setup-windows.ps1 for automatic configuration'); + errors.push( + "Windows: Install KiCAD 9.0+ from https://www.kicad.org/download/windows/", + ); + errors.push( + "Or run: .\\setup-windows.ps1 for automatic configuration", + ); } else if (isLinux) { - errors.push('Linux: Install KiCAD 9.0+ or set KICAD_PYTHON environment variable'); - errors.push('Set KICAD_PYTHON to specify a custom Python path'); + errors.push( + "Linux: Install KiCAD 9.0+ or set KICAD_PYTHON environment variable", + ); + errors.push("Set KICAD_PYTHON to specify a custom Python path"); } } } else { // Command name: verify it's executable via --version test logger.info(`Validating command-based Python executable: ${pythonExe}`); try { - const { stdout } = await new Promise<{stdout: string, stderr: string}>((resolve, reject) => { - exec(`"${pythonExe}" --version`, { - timeout: 3000, - env: { ...process.env } - }, (error: any, stdout: string, stderr: string) => { - if (error) { - reject(error); - } else { - resolve({ stdout, stderr }); - } - }); + const { stdout } = await new Promise<{ + stdout: string; + stderr: string; + }>((resolve, reject) => { + exec( + `"${pythonExe}" --version`, + { + timeout: 3000, + env: { ...process.env }, + }, + (error: any, stdout: string, stderr: string) => { + if (error) { + reject(error); + } else { + resolve({ stdout, stderr }); + } + }, + ); }); logger.info(`Python version check passed: ${stdout.trim()}`); } catch (error: any) { errors.push(`Python executable not found in PATH: ${pythonExe}`); errors.push(`Error: ${error.message}`); - errors.push('Set KICAD_PYTHON environment variable to specify full path'); + errors.push( + "Set KICAD_PYTHON environment variable to specify full path", + ); if (isLinux) { - errors.push(''); - errors.push('Linux troubleshooting:'); - errors.push('1. Check if python3 is installed: which python3'); - errors.push('2. Install KiCAD: sudo apt install kicad (Ubuntu/Debian)'); - errors.push('3. Set KICAD_PYTHON=/usr/bin/python3 in your MCP config'); + errors.push(""); + errors.push("Linux troubleshooting:"); + errors.push("1. Check if python3 is installed: which python3"); + errors.push( + "2. Install KiCAD: sudo apt install kicad (Ubuntu/Debian)", + ); + errors.push( + "3. Set KICAD_PYTHON=/usr/bin/python3 in your MCP config", + ); } } } @@ -297,76 +346,91 @@ export class KiCADMcpServer { } // Check if dist/index.js exists (if running from compiled code) - const distPath = join(dirname(dirname(this.kicadScriptPath)), 'dist', 'index.js'); + const distPath = join( + dirname(dirname(this.kicadScriptPath)), + "dist", + "index.js", + ); if (!existsSync(distPath)) { - errors.push('Project not built. Run: npm run build'); + errors.push("Project not built. Run: npm run build"); } // Try to test pcbnew import (quick validation) if (existsSync(pythonExe) && existsSync(this.kicadScriptPath)) { - logger.info('Validating pcbnew module access...'); + logger.info("Validating pcbnew module access..."); const testCommand = `"${pythonExe}" -c "import pcbnew; print('OK')"`; try { - const { stdout, stderr } = await new Promise<{stdout: string, stderr: string}>((resolve, reject) => { - exec(testCommand, { - timeout: 5000, - env: { ...process.env } - }, (error: any, stdout: string, stderr: string) => { - if (error) { - reject(error); - } else { - resolve({ stdout, stderr }); - } - }); + const { stdout, stderr } = await new Promise<{ + stdout: string; + stderr: string; + }>((resolve, reject) => { + exec( + testCommand, + { + timeout: 5000, + env: { ...process.env }, + }, + (error: any, stdout: string, stderr: string) => { + if (error) { + reject(error); + } else { + resolve({ stdout, stderr }); + } + }, + ); }); - if (!stdout.includes('OK')) { - errors.push('pcbnew module import test failed'); + if (!stdout.includes("OK")) { + errors.push("pcbnew module import test failed"); errors.push(`Output: ${stdout}`); errors.push(`Errors: ${stderr}`); if (isWindows) { - errors.push(''); - errors.push('Windows troubleshooting:'); - errors.push('1. Set PYTHONPATH=C:\\Program Files\\KiCad\\9.0\\lib\\python3\\dist-packages'); - errors.push('2. Test: "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe" -c "import pcbnew"'); - errors.push('3. Run: .\\setup-windows.ps1 for automatic fix'); - errors.push('4. See: docs/WINDOWS_TROUBLESHOOTING.md'); + errors.push(""); + errors.push("Windows troubleshooting:"); + errors.push( + "1. Set PYTHONPATH=C:\\Program Files\\KiCad\\9.0\\lib\\python3\\dist-packages", + ); + errors.push( + '2. Test: "C:\\Program Files\\KiCad\\9.0\\bin\\python.exe" -c "import pcbnew"', + ); + errors.push("3. Run: .\\setup-windows.ps1 for automatic fix"); + errors.push("4. See: docs/WINDOWS_TROUBLESHOOTING.md"); } } else { - logger.info('✓ pcbnew module validated successfully'); + logger.info("✓ pcbnew module validated successfully"); } } catch (error: any) { errors.push(`pcbnew validation failed: ${error.message}`); if (isWindows) { - errors.push(''); - errors.push('This usually means:'); - errors.push('- KiCAD is not installed'); - errors.push('- PYTHONPATH is incorrect'); - errors.push('- Python cannot find pcbnew module'); - errors.push(''); - errors.push('Quick fix: Run .\\setup-windows.ps1'); + errors.push(""); + errors.push("This usually means:"); + errors.push("- KiCAD is not installed"); + errors.push("- PYTHONPATH is incorrect"); + errors.push("- Python cannot find pcbnew module"); + errors.push(""); + errors.push("Quick fix: Run .\\setup-windows.ps1"); } } } // Log all errors if (errors.length > 0) { - logger.error('='.repeat(70)); - logger.error('STARTUP VALIDATION FAILED'); - logger.error('='.repeat(70)); - errors.forEach(err => logger.error(err)); - logger.error('='.repeat(70)); + logger.error("=".repeat(70)); + logger.error("STARTUP VALIDATION FAILED"); + logger.error("=".repeat(70)); + errors.forEach((err) => logger.error(err)); + logger.error("=".repeat(70)); // Also write to stderr for Claude Desktop to capture - process.stderr.write('\n' + '='.repeat(70) + '\n'); - process.stderr.write('KiCAD MCP Server - Startup Validation Failed\n'); - process.stderr.write('='.repeat(70) + '\n'); - errors.forEach(err => process.stderr.write(err + '\n')); - process.stderr.write('='.repeat(70) + '\n\n'); + process.stderr.write("\n" + "=".repeat(70) + "\n"); + process.stderr.write("KiCAD MCP Server - Startup Validation Failed\n"); + process.stderr.write("=".repeat(70) + "\n"); + errors.forEach((err) => process.stderr.write(err + "\n")); + process.stderr.write("=".repeat(70) + "\n\n"); return false; } @@ -379,10 +443,12 @@ export class KiCADMcpServer { */ async start(): Promise { try { - logger.info('Starting KiCAD MCP server...'); + logger.info("Starting KiCAD MCP server..."); // Start the Python process for KiCAD scripting - logger.info(`Starting Python process with script: ${this.kicadScriptPath}`); + logger.info( + `Starting Python process with script: ${this.kicadScriptPath}`, + ); const pythonExe = findPythonExecutable(this.kicadScriptPath); logger.info(`Using Python executable: ${pythonExe}`); @@ -390,76 +456,82 @@ export class KiCADMcpServer { // Validate prerequisites const isValid = await this.validatePrerequisites(pythonExe); if (!isValid) { - throw new Error('Prerequisites validation failed. See logs above for details.'); + throw new Error( + "Prerequisites validation failed. See logs above for details.", + ); } this.pythonProcess = spawn(pythonExe, [this.kicadScriptPath], { - stdio: ['pipe', 'pipe', 'pipe'], + stdio: ["pipe", "pipe", "pipe"], env: { ...process.env, - PYTHONPATH: process.env.PYTHONPATH || 'C:/Program Files/KiCad/9.0/lib/python3/dist-packages' - } + PYTHONPATH: + process.env.PYTHONPATH || + "C:/Program Files/KiCad/9.0/lib/python3/dist-packages", + }, }); - + // Listen for process exit - this.pythonProcess.on('exit', (code, signal) => { - logger.warn(`Python process exited with code ${code} and signal ${signal}`); + this.pythonProcess.on("exit", (code, signal) => { + logger.warn( + `Python process exited with code ${code} and signal ${signal}`, + ); this.pythonProcess = null; }); - + // Listen for process errors - this.pythonProcess.on('error', (err) => { + this.pythonProcess.on("error", (err) => { logger.error(`Python process error: ${err.message}`); }); - + // Set up error logging for stderr if (this.pythonProcess.stderr) { - this.pythonProcess.stderr.on('data', (data: Buffer) => { + this.pythonProcess.stderr.on("data", (data: Buffer) => { logger.error(`Python stderr: ${data.toString()}`); }); } // Set up persistent stdout handler (instead of adding/removing per request) if (this.pythonProcess.stdout) { - this.pythonProcess.stdout.on('data', (data: Buffer) => { + this.pythonProcess.stdout.on("data", (data: Buffer) => { this.handlePythonResponse(data); }); } // Connect server to STDIO transport - logger.info('Connecting MCP server to STDIO transport...'); + logger.info("Connecting MCP server to STDIO transport..."); try { await this.server.connect(this.stdioTransport); - logger.info('Successfully connected to STDIO transport'); + logger.info("Successfully connected to STDIO transport"); } catch (error) { logger.error(`Failed to connect to STDIO transport: ${error}`); throw error; } - + // Write a ready message to stderr (for debugging) - process.stderr.write('KiCAD MCP SERVER READY\n'); - - logger.info('KiCAD MCP server started and ready'); + process.stderr.write("KiCAD MCP SERVER READY\n"); + + logger.info("KiCAD MCP server started and ready"); } catch (error) { logger.error(`Failed to start KiCAD MCP server: ${error}`); throw error; } } - + /** * Stop the MCP server and clean up resources */ async stop(): Promise { - logger.info('Stopping KiCAD MCP server...'); - + logger.info("Stopping KiCAD MCP server..."); + // Kill the Python process if it's running if (this.pythonProcess) { this.pythonProcess.kill(); this.pythonProcess = null; } - - logger.info('KiCAD MCP server stopped'); + + logger.info("KiCAD MCP server stopped"); } - + /** * Call the KiCAD scripting interface to execute commands * @@ -471,7 +543,7 @@ export class KiCADMcpServer { return new Promise((resolve, reject) => { // Check if Python process is running if (!this.pythonProcess) { - logger.error('Python process is not running'); + logger.error("Python process is not running"); reject(new Error("Python process for KiCAD scripting is not running")); return; } @@ -479,17 +551,24 @@ export class KiCADMcpServer { // Determine timeout based on command type // DRC and export operations need longer timeouts for large boards let commandTimeout = 30000; // Default 30 seconds - const longRunningCommands = ['run_drc', 'export_gerber', 'export_pdf', 'export_3d']; + const longRunningCommands = [ + "run_drc", + "export_gerber", + "export_pdf", + "export_3d", + ]; if (longRunningCommands.includes(command)) { commandTimeout = 600000; // 10 minutes for long operations - logger.info(`Using extended timeout (${commandTimeout/1000}s) for command: ${command}`); + logger.info( + `Using extended timeout (${commandTimeout / 1000}s) for command: ${command}`, + ); } // Add request to queue with timeout info this.requestQueue.push({ request: { command, params, timeout: commandTimeout }, resolve, - reject + reject, }); // Process the queue if not already processing @@ -498,7 +577,7 @@ export class KiCADMcpServer { } }); } - + /** * Handle incoming data from Python process stdout * This is a persistent handler that processes all responses @@ -519,8 +598,10 @@ export class KiCADMcpServer { if (!this.currentRequestHandler) { // No pending request, clear buffer if it has data (shouldn't happen) if (this.responseBuffer.trim()) { - logger.warn(`Received data with no pending request: ${this.responseBuffer.substring(0, 100)}...`); - this.responseBuffer = ''; + logger.warn( + `Received data with no pending request: ${this.responseBuffer.substring(0, 100)}...`, + ); + this.responseBuffer = ""; } return; } @@ -530,7 +611,9 @@ export class KiCADMcpServer { const result = JSON.parse(this.responseBuffer); // If we get here, we have a valid JSON response - logger.debug(`Completed KiCAD command with result: ${result.success ? 'success' : 'failure'}`); + logger.debug( + `Completed KiCAD command with result: ${result.success ? "success" : "failure"}`, + ); // Clear the timeout since we got a response if (this.currentRequestHandler.timeoutHandle) { @@ -541,7 +624,7 @@ export class KiCADMcpServer { const handler = this.currentRequestHandler; // Clear state - this.responseBuffer = ''; + this.responseBuffer = ""; this.currentRequestHandler = null; this.processingRequest = false; @@ -550,7 +633,6 @@ export class KiCADMcpServer { // Process next request if any setTimeout(() => this.processNextRequest(), 0); - } catch (e) { // Not a complete JSON yet, keep collecting data // This is normal for large responses that come in chunks @@ -579,21 +661,29 @@ export class KiCADMcpServer { const requestStr = JSON.stringify(request); // Clear response buffer for new request - this.responseBuffer = ''; + this.responseBuffer = ""; // Set a timeout (use command-specific timeout or default) const timeoutDuration = request.timeout || 30000; const timeoutHandle = setTimeout(() => { - logger.error(`Command timeout after ${timeoutDuration/1000}s: ${request.command}`); - logger.error(`Buffer contents: ${this.responseBuffer.substring(0, 200)}...`); + logger.error( + `Command timeout after ${timeoutDuration / 1000}s: ${request.command}`, + ); + logger.error( + `Buffer contents: ${this.responseBuffer.substring(0, 200)}...`, + ); // Clear state - this.responseBuffer = ''; + this.responseBuffer = ""; this.currentRequestHandler = null; this.processingRequest = false; // Reject the promise - reject(new Error(`Command timeout after ${timeoutDuration/1000}s: ${request.command}`)); + reject( + new Error( + `Command timeout after ${timeoutDuration / 1000}s: ${request.command}`, + ), + ); // Process next request setTimeout(() => this.processNextRequest(), 0); @@ -604,7 +694,7 @@ export class KiCADMcpServer { // Write the request to the Python process logger.debug(`Sending request: ${requestStr}`); - this.pythonProcess?.stdin?.write(requestStr + '\n'); + this.pythonProcess?.stdin?.write(requestStr + "\n"); } catch (error) { logger.error(`Error processing request: ${error}`);