diff --git a/python/commands/component_schematic.py b/python/commands/component_schematic.py index ef66224..7be8c4c 100644 --- a/python/commands/component_schematic.py +++ b/python/commands/component_schematic.py @@ -10,11 +10,15 @@ logger = logging.getLogger(__name__) # Import dynamic symbol loader try: from commands.dynamic_symbol_loader import DynamicSymbolLoader + DYNAMIC_LOADING_AVAILABLE = True except ImportError: - logger.warning("Dynamic symbol loader not available - falling back to template-only mode") + logger.warning( + "Dynamic symbol loader not available - falling back to template-only mode" + ) DYNAMIC_LOADING_AVAILABLE = False + class ComponentManager: """Manage components in a schematic""" @@ -31,43 +35,44 @@ class ComponentManager: # Template symbol references mapping component type to template reference TEMPLATE_MAP = { # Passives - 'R': '_TEMPLATE_R', - 'C': '_TEMPLATE_C', - 'L': '_TEMPLATE_L', - 'Y': '_TEMPLATE_Y', - 'Crystal': '_TEMPLATE_Y', - + "R": "_TEMPLATE_R", + "C": "_TEMPLATE_C", + "L": "_TEMPLATE_L", + "Y": "_TEMPLATE_Y", + "Crystal": "_TEMPLATE_Y", # Semiconductors - 'D': '_TEMPLATE_D', - 'LED': '_TEMPLATE_LED', - 'Q': '_TEMPLATE_Q_NPN', - 'Q_NPN': '_TEMPLATE_Q_NPN', - 'Q_NMOS': '_TEMPLATE_Q_NMOS', - 'MOSFET': '_TEMPLATE_Q_NMOS', - + "D": "_TEMPLATE_D", + "LED": "_TEMPLATE_LED", + "Q": "_TEMPLATE_Q_NPN", + "Q_NPN": "_TEMPLATE_Q_NPN", + "Q_NMOS": "_TEMPLATE_Q_NMOS", + "MOSFET": "_TEMPLATE_Q_NMOS", # ICs - 'U': '_TEMPLATE_U_OPAMP', - 'OpAmp': '_TEMPLATE_U_OPAMP', - 'IC': '_TEMPLATE_U_OPAMP', - 'U_REG': '_TEMPLATE_U_REG', - 'Regulator': '_TEMPLATE_U_REG', - + "U": "_TEMPLATE_U_OPAMP", + "OpAmp": "_TEMPLATE_U_OPAMP", + "IC": "_TEMPLATE_U_OPAMP", + "U_REG": "_TEMPLATE_U_REG", + "Regulator": "_TEMPLATE_U_REG", # Connectors - 'J': '_TEMPLATE_J2', - 'J2': '_TEMPLATE_J2', - 'J4': '_TEMPLATE_J4', - 'Conn_2': '_TEMPLATE_J2', - 'Conn_4': '_TEMPLATE_J4', - + "J": "_TEMPLATE_J2", + "J2": "_TEMPLATE_J2", + "J4": "_TEMPLATE_J4", + "Conn_2": "_TEMPLATE_J2", + "Conn_4": "_TEMPLATE_J4", # Misc - 'SW': '_TEMPLATE_SW', - 'Button': '_TEMPLATE_SW', - 'Switch': '_TEMPLATE_SW', + "SW": "_TEMPLATE_SW", + "Button": "_TEMPLATE_SW", + "Switch": "_TEMPLATE_SW", } @classmethod - def get_or_create_template(cls, schematic: Schematic, comp_type: str, library: Optional[str] = None, - schematic_path: Optional[Path] = None) -> tuple: + def get_or_create_template( + cls, + schematic: Schematic, + comp_type: str, + library: Optional[str] = None, + schematic_path: Optional[Path] = None, + ) -> tuple: """ Get template reference for a component type, creating it dynamically if needed @@ -80,11 +85,15 @@ class ComponentManager: Returns: Tuple of (template_ref, needs_reload) where needs_reload indicates if schematic must be reloaded """ + # Helper function to check if template exists in schematic def template_exists(schematic, template_ref): """Check if template exists by iterating symbols (handles special characters)""" for symbol in schematic.symbol: - if hasattr(symbol.property, 'Reference') and symbol.property.Reference.value == template_ref: + if ( + hasattr(symbol.property, "Reference") + and symbol.property.Reference.value == template_ref + ): return True return False @@ -113,46 +122,61 @@ class ComponentManager: # 3. Try dynamic loading if not DYNAMIC_LOADING_AVAILABLE: - logger.warning(f"Component type '{comp_type}' not in static templates and dynamic loading unavailable") + logger.warning( + f"Component type '{comp_type}' not in static templates and dynamic loading unavailable" + ) # Fall back to basic resistor template - return ('_TEMPLATE_R', False) + return ("_TEMPLATE_R", False) loader = cls.get_dynamic_loader() if not loader: logger.warning("Dynamic loader unavailable, using fallback template") - return ('_TEMPLATE_R', False) + return ("_TEMPLATE_R", False) # Check if schematic path is available if schematic_path is None: - logger.warning("Dynamic loading requires schematic file path but none was provided") - fallback = cls.TEMPLATE_MAP.get(comp_type, '_TEMPLATE_R') + logger.warning( + "Dynamic loading requires schematic file path but none was provided" + ) + fallback = cls.TEMPLATE_MAP.get(comp_type, "_TEMPLATE_R") return (fallback, False) # Determine library name if library is None: # Default library for common component types - library = 'Device' # Most passives and basic components are in Device library + library = ( + "Device" # Most passives and basic components are in Device library + ) try: - logger.info(f"Attempting dynamic load: {library}:{comp_type} from {schematic_path}") + logger.info( + f"Attempting dynamic load: {library}:{comp_type} from {schematic_path}" + ) # Use dynamic symbol loader to inject symbol and create template - template_ref = loader.load_symbol_dynamically(schematic_path, library, comp_type) + template_ref = loader.load_symbol_dynamically( + schematic_path, library, comp_type + ) - logger.info(f"Successfully loaded symbol dynamically. Template ref: {template_ref}") + logger.info( + f"Successfully loaded symbol dynamically. Template ref: {template_ref}" + ) # Signal that schematic needs reload to see new template return (template_ref, True) except Exception as e: logger.error(f"Dynamic loading failed: {e}") import traceback + logger.error(traceback.format_exc()) # Fall back to static template if available - fallback = cls.TEMPLATE_MAP.get(comp_type, '_TEMPLATE_R') + fallback = cls.TEMPLATE_MAP.get(comp_type, "_TEMPLATE_R") return (fallback, False) @staticmethod - def add_component(schematic: Schematic, component_def: dict, schematic_path: Optional[Path] = None): + def add_component( + schematic: Schematic, component_def: dict, schematic_path: Optional[Path] = None + ): """ Add a component to the schematic by cloning from template @@ -167,66 +191,81 @@ class ComponentManager: try: from commands.schematic import SchematicManager - logger.info(f"Adding component: type={component_def.get('type')}, ref={component_def.get('reference')}") + logger.info( + f"Adding component: type={component_def.get('type')}, ref={component_def.get('reference')}" + ) logger.debug(f"Full component_def: {component_def}") # Get component type and determine template - comp_type = component_def.get('type', 'R') - library = component_def.get('library', None) # Optional library specification + comp_type = component_def.get("type", "R") + library = component_def.get( + "library", None + ) # Optional library specification # Get template reference (static or dynamic) - template_ref, needs_reload = ComponentManager.get_or_create_template(schematic, comp_type, library, schematic_path) + template_ref, needs_reload = ComponentManager.get_or_create_template( + schematic, comp_type, library, schematic_path + ) # If dynamic loading occurred, reload schematic to see new template if needs_reload and schematic_path: - logger.info(f"Reloading schematic after dynamic loading: {schematic_path}") + logger.info( + f"Reloading schematic after dynamic loading: {schematic_path}" + ) schematic = SchematicManager.load_schematic(str(schematic_path)) # Find template symbol by reference (handles special characters like +) template_symbol = None for symbol in schematic.symbol: - if hasattr(symbol.property, 'Reference') and symbol.property.Reference.value == template_ref: + if ( + hasattr(symbol.property, "Reference") + and symbol.property.Reference.value == template_ref + ): template_symbol = symbol break if not template_symbol: - logger.error(f"Template symbol {template_ref} not found in schematic. Available symbols: {[str(s.property.Reference.value) for s in schematic.symbol]}") - raise ValueError(f"Template symbol {template_ref} not found. The schematic must be created from template_with_symbols.kicad_sch") + logger.error( + f"Template symbol {template_ref} not found in schematic. Available symbols: {[str(s.property.Reference.value) for s in schematic.symbol]}" + ) + raise ValueError( + f"Template symbol {template_ref} not found. The schematic must be created from template_with_symbols.kicad_sch" + ) # Clone the template symbol new_symbol = template_symbol.clone() logger.debug(f"Cloned template symbol {template_ref}") # Set reference - reference = component_def.get('reference', 'R?') + reference = component_def.get("reference", "R?") new_symbol.property.Reference.value = reference logger.debug(f"Set reference to {reference}") # Set value - if 'value' in component_def: - new_symbol.property.Value.value = component_def['value'] + if "value" in component_def: + new_symbol.property.Value.value = component_def["value"] logger.debug(f"Set value to {component_def['value']}") # Set footprint - if 'footprint' in component_def: - new_symbol.property.Footprint.value = component_def['footprint'] + if "footprint" in component_def: + new_symbol.property.Footprint.value = component_def["footprint"] logger.debug(f"Set footprint to {component_def['footprint']}") # Set datasheet - if 'datasheet' in component_def: - new_symbol.property.Datasheet.value = component_def['datasheet'] + if "datasheet" in component_def: + new_symbol.property.Datasheet.value = component_def["datasheet"] # Set position - x = component_def.get('x', 0) - y = component_def.get('y', 0) - rotation = component_def.get('rotation', 0) + x = component_def.get("x", 0) + y = component_def.get("y", 0) + rotation = component_def.get("rotation", 0) new_symbol.at.value = [x, y, rotation] logger.debug(f"Set position to ({x}, {y}, {rotation})") # Set BOM and board flags - new_symbol.in_bom.value = component_def.get('in_bom', True) - new_symbol.on_board.value = component_def.get('on_board', True) - new_symbol.dnp.value = component_def.get('dnp', False) + new_symbol.in_bom.value = component_def.get("in_bom", True) + new_symbol.on_board.value = component_def.get("on_board", True) + new_symbol.dnp.value = component_def.get("dnp", False) # Generate new UUID new_symbol.uuid.value = str(uuid.uuid4()) @@ -253,7 +292,7 @@ class ComponentManager: break if symbol_to_remove: - schematic.symbol.remove(symbol_to_remove) + schematic.symbol._elements.remove(symbol_to_remove) print(f"Removed component {component_ref} from schematic.") return True else: @@ -263,9 +302,10 @@ class ComponentManager: print(f"Error removing component {component_ref}: {e}") return False - @staticmethod - def update_component(schematic: Schematic, component_ref: str, new_properties: dict): + def update_component( + schematic: Schematic, component_ref: str, new_properties: dict + ): """Update component properties by reference designator""" try: symbol_to_update = None @@ -279,8 +319,8 @@ class ComponentManager: if key in symbol_to_update.property: symbol_to_update.property[key].value = value else: - # Add as a new property if it doesn't exist - symbol_to_update.property.append(key, value) + # Add as a new property if it doesn't exist + symbol_to_update.property.append(key, value) print(f"Updated properties for component {component_ref}.") return True else: @@ -307,9 +347,14 @@ class ComponentManager: matching_components = [] query_lower = query.lower() for symbol in schematic.symbol: - if query_lower in symbol.reference.lower() or \ - query_lower in symbol.name.lower() or \ - (hasattr(symbol.property, 'Value') and query_lower in symbol.property.Value.value.lower()): + if ( + query_lower in symbol.reference.lower() + or query_lower in symbol.name.lower() + or ( + hasattr(symbol.property, "Value") + and query_lower in symbol.property.Value.value.lower() + ) + ): matching_components.append(symbol) print(f"Found {len(matching_components)} components matching query '{query}'.") return matching_components @@ -320,17 +365,34 @@ class ComponentManager: print(f"Retrieving all {len(schematic.symbol)} components.") return list(schematic.symbol) -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("ComponentTestSchematic") # Add components comp1_def = {"type": "R", "reference": "R1", "value": "10k", "x": 100, "y": 100} - comp2_def = {"type": "C", "reference": "C1", "value": "0.1uF", "x": 200, "y": 100, "library": "Device"} - comp3_def = {"type": "LED", "reference": "D1", "x": 300, "y": 100, "library": "Device", "properties": {"Color": "Red"}} + comp2_def = { + "type": "C", + "reference": "C1", + "value": "0.1uF", + "x": 200, + "y": 100, + "library": "Device", + } + comp3_def = { + "type": "LED", + "reference": "D1", + "x": 300, + "y": 100, + "library": "Device", + "properties": {"Color": "Red"}, + } comp1 = ComponentManager.add_component(test_sch, comp1_def) comp2 = ComponentManager.add_component(test_sch, comp2_def) @@ -339,13 +401,19 @@ if __name__ == '__main__': # Get a component retrieved_comp = ComponentManager.get_component(test_sch, "C1") if retrieved_comp: - print(f"Retrieved component: {retrieved_comp.reference} ({retrieved_comp.value})") + print( + f"Retrieved component: {retrieved_comp.reference} ({retrieved_comp.value})" + ) # Update a component - ComponentManager.update_component(test_sch, "R1", {"value": "20k", "Tolerance": "5%"}) + ComponentManager.update_component( + test_sch, "R1", {"value": "20k", "Tolerance": "5%"} + ) # Search components - matching_comps = ComponentManager.search_components(test_sch, "100") # Search by position + matching_comps = ComponentManager.search_components( + test_sch, "100" + ) # Search by position print(f"Search results for '100': {[c.reference for c in matching_comps]}") # Get all components @@ -355,7 +423,9 @@ if __name__ == '__main__': # Remove a component ComponentManager.remove_component(test_sch, "D1") all_comps_after_remove = ComponentManager.get_all_components(test_sch) - print(f"Components after removing D1: {[c.reference for c in all_comps_after_remove]}") + print( + f"Components after removing D1: {[c.reference for c in all_comps_after_remove]}" + ) # Save the schematic (optional) # SchematicManager.save_schematic(test_sch, "component_test.kicad_sch") diff --git a/python/commands/wire_manager.py b/python/commands/wire_manager.py index 8ff9267..272760b 100644 --- a/python/commands/wire_manager.py +++ b/python/commands/wire_manager.py @@ -15,15 +15,20 @@ from typing import List, Tuple, Optional, Dict import sexpdata from sexpdata import Symbol -logger = logging.getLogger('kicad_interface') +logger = logging.getLogger("kicad_interface") class WireManager: """Manage wires in KiCad schematics using S-expression manipulation""" @staticmethod - def add_wire(schematic_path: Path, start_point: List[float], end_point: List[float], - stroke_width: float = 0, stroke_type: str = 'default') -> bool: + def add_wire( + schematic_path: Path, + start_point: List[float], + end_point: List[float], + stroke_width: float = 0, + stroke_type: str = "default", + ) -> bool: """ Add a wire to the schematic using S-expression manipulation @@ -39,7 +44,7 @@ class WireManager: """ 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) @@ -47,22 +52,28 @@ class WireManager: # Create wire S-expression # Format: (wire (pts (xy x1 y1) (xy x2 y2)) (stroke (width N) (type default)) (uuid ...)) wire_sexp = [ - Symbol('wire'), - [Symbol('pts'), - [Symbol('xy'), start_point[0], start_point[1]], - [Symbol('xy'), end_point[0], end_point[1]] + Symbol("wire"), + [ + Symbol("pts"), + [Symbol("xy"), start_point[0], start_point[1]], + [Symbol("xy"), end_point[0], end_point[1]], ], - [Symbol('stroke'), - [Symbol('width'), stroke_width], - [Symbol('type'), Symbol(stroke_type)] + [ + Symbol("stroke"), + [Symbol("width"), stroke_width], + [Symbol("type"), Symbol(stroke_type)], ], - [Symbol('uuid'), str(uuid.uuid4())] + [Symbol("uuid"), str(uuid.uuid4())], ] # Find insertion point (before sheet_instances) sheet_instances_index = None for i, item in enumerate(sch_data): - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('sheet_instances'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("sheet_instances") + ): sheet_instances_index = i break @@ -75,7 +86,7 @@ class WireManager: logger.info(f"Injected wire from {start_point} to {end_point}") # Write back - with open(schematic_path, 'w', encoding='utf-8') as f: + with open(schematic_path, "w", encoding="utf-8") as f: output = sexpdata.dumps(sch_data) f.write(output) @@ -85,12 +96,17 @@ class WireManager: except Exception as e: logger.error(f"Error adding wire: {e}") import traceback + logger.error(traceback.format_exc()) return False @staticmethod - def add_polyline_wire(schematic_path: Path, points: List[List[float]], - stroke_width: float = 0, stroke_type: str = 'default') -> bool: + def add_polyline_wire( + schematic_path: Path, + points: List[List[float]], + stroke_width: float = 0, + stroke_type: str = "default", + ) -> bool: """ Add a multi-segment wire (polyline) to the schematic @@ -109,31 +125,36 @@ class WireManager: return False # 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) # Create pts list - pts_list = [Symbol('pts')] + pts_list = [Symbol("pts")] for point in points: - pts_list.append([Symbol('xy'), point[0], point[1]]) + pts_list.append([Symbol("xy"), point[0], point[1]]) # Create wire S-expression with multiple points wire_sexp = [ - Symbol('wire'), + Symbol("wire"), pts_list, - [Symbol('stroke'), - [Symbol('width'), stroke_width], - [Symbol('type'), Symbol(stroke_type)] + [ + Symbol("stroke"), + [Symbol("width"), stroke_width], + [Symbol("type"), Symbol(stroke_type)], ], - [Symbol('uuid'), str(uuid.uuid4())] + [Symbol("uuid"), str(uuid.uuid4())], ] # Find insertion point sheet_instances_index = None for i, item in enumerate(sch_data): - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('sheet_instances'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("sheet_instances") + ): sheet_instances_index = i break @@ -146,7 +167,7 @@ class WireManager: logger.info(f"Injected polyline wire with {len(points)} points") # Write back - with open(schematic_path, 'w', encoding='utf-8') as f: + with open(schematic_path, "w", encoding="utf-8") as f: output = sexpdata.dumps(sch_data) f.write(output) @@ -156,12 +177,18 @@ class WireManager: except Exception as e: logger.error(f"Error adding polyline wire: {e}") import traceback + logger.error(traceback.format_exc()) return False @staticmethod - def add_label(schematic_path: Path, text: str, position: List[float], - label_type: str = 'label', orientation: int = 0) -> bool: + def add_label( + schematic_path: Path, + text: str, + position: List[float], + label_type: str = "label", + orientation: int = 0, + ) -> bool: """ Add a net label to the schematic @@ -177,7 +204,7 @@ class WireManager: """ 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) @@ -187,19 +214,24 @@ class WireManager: label_sexp = [ Symbol(label_type), text, - [Symbol('at'), position[0], position[1], orientation], - [Symbol('fields_autoplaced'), Symbol('yes')], - [Symbol('effects'), - [Symbol('font'), [Symbol('size'), 1.27, 1.27]], - [Symbol('justify'), Symbol('left'), Symbol('bottom')] + [Symbol("at"), position[0], position[1], orientation], + [Symbol("fields_autoplaced"), Symbol("yes")], + [ + Symbol("effects"), + [Symbol("font"), [Symbol("size"), 1.27, 1.27]], + [Symbol("justify"), Symbol("left"), Symbol("bottom")], ], - [Symbol('uuid'), str(uuid.uuid4())] + [Symbol("uuid"), str(uuid.uuid4())], ] # Find insertion point sheet_instances_index = None for i, item in enumerate(sch_data): - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('sheet_instances'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("sheet_instances") + ): sheet_instances_index = i break @@ -212,7 +244,7 @@ class WireManager: logger.info(f"Injected label '{text}' at {position}") # Write back - with open(schematic_path, 'w', encoding='utf-8') as f: + with open(schematic_path, "w", encoding="utf-8") as f: output = sexpdata.dumps(sch_data) f.write(output) @@ -222,11 +254,14 @@ class WireManager: except Exception as e: logger.error(f"Error adding label: {e}") import traceback + logger.error(traceback.format_exc()) return False @staticmethod - def add_junction(schematic_path: Path, position: List[float], diameter: float = 0) -> bool: + def add_junction( + schematic_path: Path, position: List[float], diameter: float = 0 + ) -> bool: """ Add a junction (connection dot) to the schematic @@ -240,7 +275,7 @@ class WireManager: """ 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) @@ -248,17 +283,21 @@ class WireManager: # Create junction S-expression # Format: (junction (at x y) (diameter 0) (color 0 0 0 0) (uuid ...)) junction_sexp = [ - Symbol('junction'), - [Symbol('at'), position[0], position[1]], - [Symbol('diameter'), diameter], - [Symbol('color'), 0, 0, 0, 0], - [Symbol('uuid'), str(uuid.uuid4())] + Symbol("junction"), + [Symbol("at"), position[0], position[1]], + [Symbol("diameter"), diameter], + [Symbol("color"), 0, 0, 0, 0], + [Symbol("uuid"), str(uuid.uuid4())], ] # Find insertion point sheet_instances_index = None for i, item in enumerate(sch_data): - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('sheet_instances'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("sheet_instances") + ): sheet_instances_index = i break @@ -271,7 +310,7 @@ class WireManager: logger.info(f"Injected junction at {position}") # Write back - with open(schematic_path, 'w', encoding='utf-8') as f: + with open(schematic_path, "w", encoding="utf-8") as f: output = sexpdata.dumps(sch_data) f.write(output) @@ -281,6 +320,7 @@ class WireManager: except Exception as e: logger.error(f"Error adding junction: {e}") import traceback + logger.error(traceback.format_exc()) return False @@ -298,7 +338,7 @@ class WireManager: """ 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) @@ -306,15 +346,19 @@ class WireManager: # Create no_connect S-expression # Format: (no_connect (at x y) (uuid ...)) no_connect_sexp = [ - Symbol('no_connect'), - [Symbol('at'), position[0], position[1]], - [Symbol('uuid'), str(uuid.uuid4())] + Symbol("no_connect"), + [Symbol("at"), position[0], position[1]], + [Symbol("uuid"), str(uuid.uuid4())], ] # Find insertion point sheet_instances_index = None for i, item in enumerate(sch_data): - if isinstance(item, list) and len(item) > 0 and item[0] == Symbol('sheet_instances'): + if ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("sheet_instances") + ): sheet_instances_index = i break @@ -327,7 +371,7 @@ class WireManager: logger.info(f"Injected no-connect at {position}") # Write back - with open(schematic_path, 'w', encoding='utf-8') as f: + with open(schematic_path, "w", encoding="utf-8") as f: output = sexpdata.dumps(sch_data) f.write(output) @@ -337,12 +381,179 @@ class WireManager: except Exception as e: logger.error(f"Error adding no-connect: {e}") import traceback + logger.error(traceback.format_exc()) return False @staticmethod - def create_orthogonal_path(start: List[float], end: List[float], - prefer_horizontal_first: bool = True) -> List[List[float]]: + def delete_wire( + schematic_path: Path, + start_point: List[float], + end_point: List[float], + tolerance: float = 0.5, + ) -> bool: + """ + Delete a wire from the schematic matching given start/end coordinates. + + Args: + schematic_path: Path to .kicad_sch file + start_point: [x, y] coordinates for wire start + end_point: [x, y] coordinates for wire end + tolerance: Maximum coordinate difference to consider a match (mm) + + Returns: + True if a wire was found and removed, False otherwise + """ + try: + with open(schematic_path, "r", encoding="utf-8") as f: + sch_content = f.read() + + sch_data = sexpdata.loads(sch_content) + + sx, sy = start_point + ex, ey = end_point + + for i, item in enumerate(sch_data): + if not ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("wire") + ): + continue + + # Extract pts from the wire s-expression + pts_list = None + for part in item[1:]: + if ( + isinstance(part, list) + and len(part) > 0 + and part[0] == Symbol("pts") + ): + pts_list = part + break + + if pts_list is None: + continue + + xy_points = [ + p + for p in pts_list[1:] + if isinstance(p, list) and len(p) >= 3 and p[0] == Symbol("xy") + ] + if len(xy_points) < 2: + continue + + x1, y1 = float(xy_points[0][1]), float(xy_points[0][2]) + x2, y2 = float(xy_points[-1][1]), float(xy_points[-1][2]) + + match_fwd = ( + abs(x1 - sx) < tolerance + and abs(y1 - sy) < tolerance + and abs(x2 - ex) < tolerance + and abs(y2 - ey) < tolerance + ) + match_rev = ( + abs(x1 - ex) < tolerance + and abs(y1 - ey) < tolerance + and abs(x2 - sx) < tolerance + and abs(y2 - sy) < tolerance + ) + + if match_fwd or match_rev: + del sch_data[i] + with open(schematic_path, "w", encoding="utf-8") as f: + f.write(sexpdata.dumps(sch_data)) + logger.info(f"Deleted wire from {start_point} to {end_point}") + return True + + logger.warning(f"No matching wire found for {start_point} to {end_point}") + return False + + except Exception as e: + logger.error(f"Error deleting wire: {e}") + import traceback + + logger.error(traceback.format_exc()) + return False + + @staticmethod + def delete_label( + schematic_path: Path, + net_name: str, + position: Optional[List[float]] = None, + tolerance: float = 0.5, + ) -> bool: + """ + Delete a net label from the schematic by name (and optionally position). + + Args: + schematic_path: Path to .kicad_sch file + net_name: Net label text to match + position: Optional [x, y] to disambiguate when multiple labels share a name + tolerance: Maximum coordinate difference to consider a match (mm) + + Returns: + True if a label was found and removed, False otherwise + """ + try: + with open(schematic_path, "r", encoding="utf-8") as f: + sch_content = f.read() + + sch_data = sexpdata.loads(sch_content) + + for i, item in enumerate(sch_data): + if not ( + isinstance(item, list) + and len(item) > 0 + and item[0] == Symbol("label") + ): + continue + + # Second element is the label text + if len(item) < 2 or item[1] != net_name: + continue + + if position is not None: + # Find (at x y ...) sub-expression and check coordinates + at_entry = next( + ( + p + for p in item[1:] + if isinstance(p, list) + and len(p) >= 3 + and p[0] == Symbol("at") + ), + None, + ) + if at_entry is None: + continue + lx, ly = float(at_entry[1]), float(at_entry[2]) + if not ( + abs(lx - position[0]) < tolerance + and abs(ly - position[1]) < tolerance + ): + continue + + del sch_data[i] + with open(schematic_path, "w", encoding="utf-8") as f: + f.write(sexpdata.dumps(sch_data)) + logger.info(f"Deleted label '{net_name}'") + return True + + logger.warning(f"No matching label found for '{net_name}'") + return False + + except Exception as e: + logger.error(f"Error deleting label: {e}") + import traceback + + logger.error(traceback.format_exc()) + return False + + @staticmethod + def create_orthogonal_path( + start: List[float], end: List[float], prefer_horizontal_first: bool = True + ) -> List[List[float]]: """ Create an orthogonal (right-angle) path between two points @@ -371,10 +582,11 @@ class WireManager: return [start, corner, end] -if __name__ == '__main__': +if __name__ == "__main__": # Test wire creation 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 import shutil @@ -384,8 +596,10 @@ if __name__ == '__main__': print("=" * 80) # Create test schematic (cross-platform temp directory) - test_path = Path(tempfile.gettempdir()) / 'test_wire_manager.kicad_sch' - template_path = Path('/home/chris/MCP/KiCAD-MCP-Server/python/templates/empty.kicad_sch') + test_path = Path(tempfile.gettempdir()) / "test_wire_manager.kicad_sch" + template_path = Path( + "/home/chris/MCP/KiCAD-MCP-Server/python/templates/empty.kicad_sch" + ) shutil.copy(template_path, test_path) print(f"\n✓ Created test schematic: {test_path}") @@ -421,8 +635,9 @@ if __name__ == '__main__': print("\n[Verification] Loading with kicad-skip...") try: from skip import Schematic + sch = Schematic(str(test_path)) - wire_count = len(list(sch.wire)) if hasattr(sch, 'wire') else 0 + wire_count = len(list(sch.wire)) if hasattr(sch, "wire") else 0 print(f" ✓ Loaded successfully") print(f" ✓ Wire count: {wire_count}") except Exception as e: diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 40ca581..50915cf 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -383,7 +383,18 @@ class KiCADInterface: "generate_netlist": self._handle_generate_netlist, "sync_schematic_to_board": self._handle_sync_schematic_to_board, "list_schematic_libraries": self._handle_list_schematic_libraries, + "get_schematic_view": self._handle_get_schematic_view, + "list_schematic_components": self._handle_list_schematic_components, + "list_schematic_nets": self._handle_list_schematic_nets, + "list_schematic_wires": self._handle_list_schematic_wires, + "list_schematic_labels": self._handle_list_schematic_labels, + "move_schematic_component": self._handle_move_schematic_component, + "rotate_schematic_component": self._handle_rotate_schematic_component, + "annotate_schematic": self._handle_annotate_schematic, + "delete_schematic_wire": self._handle_delete_schematic_wire, + "delete_schematic_net_label": self._handle_delete_schematic_net_label, "export_schematic_pdf": self._handle_export_schematic_pdf, + "export_schematic_svg": self._handle_export_schematic_svg, "import_svg_logo": self._handle_import_svg_logo, # UI/Process management commands "check_kicad_ui": self._handle_check_kicad_ui, @@ -523,11 +534,24 @@ class KiCADInterface: # Board-mutating commands that trigger auto-save on SWIG path _BOARD_MUTATING_COMMANDS = { - "place_component", "move_component", "rotate_component", "delete_component", - "route_trace", "route_pad_to_pad", "add_via", "delete_trace", "add_net", - "add_board_outline", "add_mounting_hole", "add_text", "add_board_text", - "add_copper_pour", "refill_zones", "import_svg_logo", - "sync_schematic_to_board", "connect_passthrough", + "place_component", + "move_component", + "rotate_component", + "delete_component", + "route_trace", + "route_pad_to_pad", + "add_via", + "delete_trace", + "add_net", + "add_board_outline", + "add_mounting_hole", + "add_text", + "add_board_text", + "add_copper_pour", + "refill_zones", + "import_svg_logo", + "sync_schematic_to_board", + "connect_passthrough", } def _auto_save_board(self): @@ -1175,26 +1199,39 @@ class KiCADInterface: if not output_path: return {"success": False, "message": "Output path is required"} + if not os.path.exists(schematic_path): + return { + "success": False, + "message": f"Schematic not found: {schematic_path}", + } + import subprocess - result = subprocess.run( - [ - "kicad-cli", - "sch", - "export", - "pdf", - "--output", - output_path, - schematic_path, - ], - capture_output=True, - text=True, - ) + cmd = [ + "kicad-cli", + "sch", + "export", + "pdf", + "--output", + output_path, + schematic_path, + ] - success = result.returncode == 0 - message = result.stderr if not success else "" + if params.get("blackAndWhite"): + cmd.insert(-1, "--black-and-white") - return {"success": success, "message": message} + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + + if result.returncode == 0: + return {"success": True, "file": {"path": output_path}} + else: + return { + "success": False, + "message": f"kicad-cli failed: {result.stderr}", + } + + except FileNotFoundError: + return {"success": False, "message": "kicad-cli not found in PATH"} except Exception as e: logger.error(f"Error exporting schematic to PDF: {str(e)}") return {"success": False, "message": str(e)} @@ -1342,7 +1379,10 @@ class KiCADInterface: pin_offset = int(params.get("pinOffset", 0)) if not all([schematic_path, source_ref, target_ref]): - return {"success": False, "message": "Missing required parameters: schematicPath, sourceRef, targetRef"} + return { + "success": False, + "message": "Missing required parameters: schematicPath, sourceRef, targetRef", + } result = ConnectionManager.connect_passthrough( Path(schematic_path), source_ref, target_ref, net_prefix, pin_offset @@ -1359,6 +1399,7 @@ class KiCADInterface: except Exception as e: logger.error(f"Error in connect_passthrough: {str(e)}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} @@ -1373,26 +1414,39 @@ class KiCADInterface: reference = params.get("reference") if not all([schematic_path, reference]): - return {"success": False, "message": "Missing required parameters: schematicPath, reference"} + return { + "success": False, + "message": "Missing required parameters: schematicPath, reference", + } locator = PinLocator() all_pins = locator.get_all_symbol_pins(Path(schematic_path), reference) if not all_pins: - return {"success": False, "message": f"No pins found for {reference} — check reference and schematic path"} + return { + "success": False, + "message": f"No pins found for {reference} — check reference and schematic path", + } # Enrich with pin names and angles from the symbol definition - pins_def = locator.get_symbol_pins( - Path(schematic_path), - locator._get_lib_id(Path(schematic_path), reference), - ) if hasattr(locator, "_get_lib_id") else {} + pins_def = ( + locator.get_symbol_pins( + Path(schematic_path), + locator._get_lib_id(Path(schematic_path), reference), + ) + if hasattr(locator, "_get_lib_id") + else {} + ) result = {} for pin_num, coords in all_pins.items(): entry = {"x": coords[0], "y": coords[1]} if pin_num in pins_def: entry["name"] = pins_def[pin_num].get("name", pin_num) - entry["angle"] = locator.get_pin_angle(Path(schematic_path), reference, pin_num) or 0 + entry["angle"] = ( + locator.get_pin_angle(Path(schematic_path), reference, pin_num) + or 0 + ) result[pin_num] = entry return {"success": True, "reference": reference, "pins": result} @@ -1400,9 +1454,695 @@ class KiCADInterface: except Exception as e: logger.error(f"Error getting pin locations: {e}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} + def _handle_get_schematic_view(self, params): + """Get a rasterised image of the schematic (SVG export → optional PNG conversion)""" + logger.info("Getting schematic view") + import subprocess + import tempfile + import base64 + + try: + schematic_path = params.get("schematicPath") + if not schematic_path or not os.path.exists(schematic_path): + return { + "success": False, + "message": f"Schematic not found: {schematic_path}", + } + + fmt = params.get("format", "png") + width = params.get("width", 1200) + height = params.get("height", 900) + + # Step 1: Export schematic to SVG via kicad-cli + with tempfile.TemporaryDirectory() as tmpdir: + svg_path = os.path.join(tmpdir, "schematic.svg") + cmd = [ + "kicad-cli", + "sch", + "export", + "svg", + "--output", + tmpdir, + "--no-background-color", + schematic_path, + ] + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + + if result.returncode != 0: + return { + "success": False, + "message": f"kicad-cli SVG export failed: {result.stderr}", + } + + # kicad-cli may name the file after the schematic, find it + import glob + + svg_files = glob.glob(os.path.join(tmpdir, "*.svg")) + if not svg_files: + return { + "success": False, + "message": "No SVG file produced by kicad-cli", + } + svg_path = svg_files[0] + + if fmt == "svg": + with open(svg_path, "r", encoding="utf-8") as f: + svg_data = f.read() + return {"success": True, "imageData": svg_data, "format": "svg"} + + # Step 2: Convert SVG to PNG using cairosvg + try: + from cairosvg import svg2png + except ImportError: + # Fallback: return SVG data with a note + with open(svg_path, "r", encoding="utf-8") as f: + svg_data = f.read() + return { + "success": True, + "imageData": svg_data, + "format": "svg", + "message": "cairosvg not installed — returning SVG instead of PNG. Install with: pip install cairosvg", + } + + png_data = svg2png( + url=svg_path, output_width=width, output_height=height + ) + + return { + "success": True, + "imageData": base64.b64encode(png_data).decode("utf-8"), + "format": "png", + "width": width, + "height": height, + } + + except FileNotFoundError: + return {"success": False, "message": "kicad-cli not found in PATH"} + except Exception as e: + logger.error(f"Error getting schematic view: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_list_schematic_components(self, params): + """List all components in a schematic""" + logger.info("Listing schematic components") + try: + from pathlib import Path + from commands.pin_locator import PinLocator + + schematic_path = params.get("schematicPath") + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + sch_file = Path(schematic_path) + if not sch_file.exists(): + return { + "success": False, + "message": f"Schematic not found: {schematic_path}", + } + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + # Optional filters + filter_params = params.get("filter", {}) + lib_id_filter = filter_params.get("libId", "") + ref_prefix_filter = filter_params.get("referencePrefix", "") + + locator = PinLocator() + components = [] + + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + # Skip template symbols + if ref.startswith("_TEMPLATE"): + continue + + lib_id = symbol.lib_id.value if hasattr(symbol, "lib_id") else "" + + # Apply filters + if lib_id_filter and lib_id_filter not in lib_id: + continue + if ref_prefix_filter and not ref.startswith(ref_prefix_filter): + continue + + value = ( + symbol.property.Value.value + if hasattr(symbol.property, "Value") + else "" + ) + footprint = ( + symbol.property.Footprint.value + if hasattr(symbol.property, "Footprint") + else "" + ) + position = symbol.at.value if hasattr(symbol, "at") else [0, 0, 0] + uuid_val = symbol.uuid.value if hasattr(symbol, "uuid") else "" + + comp = { + "reference": ref, + "libId": lib_id, + "value": value, + "footprint": footprint, + "position": {"x": float(position[0]), "y": float(position[1])}, + "rotation": float(position[2]) if len(position) > 2 else 0, + "uuid": str(uuid_val), + } + + # Get pins if available + try: + all_pins = locator.get_all_symbol_pins(sch_file, ref) + if all_pins: + pins_def = locator.get_symbol_pins(sch_file, lib_id) or {} + pin_list = [] + for pin_num, coords in all_pins.items(): + pin_info = { + "number": pin_num, + "position": {"x": coords[0], "y": coords[1]}, + } + if pin_num in pins_def: + pin_info["name"] = pins_def[pin_num].get( + "name", pin_num + ) + pin_list.append(pin_info) + comp["pins"] = pin_list + except Exception: + pass # Pin lookup is best-effort + + components.append(comp) + + return {"success": True, "components": components, "count": len(components)} + + except Exception as e: + logger.error(f"Error listing schematic components: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_list_schematic_nets(self, params): + """List all nets in a schematic with their connections""" + logger.info("Listing schematic nets") + try: + from pathlib import Path + + schematic_path = params.get("schematicPath") + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + # Get all net names from labels and global labels + net_names = set() + if hasattr(schematic, "label"): + for label in schematic.label: + if hasattr(label, "value"): + net_names.add(label.value) + if hasattr(schematic, "global_label"): + for label in schematic.global_label: + if hasattr(label, "value"): + net_names.add(label.value) + + nets = [] + for net_name in sorted(net_names): + connections = ConnectionManager.get_net_connections( + schematic, net_name, Path(schematic_path) + ) + nets.append( + { + "name": net_name, + "connections": connections, + } + ) + + return {"success": True, "nets": nets, "count": len(nets)} + + except Exception as e: + logger.error(f"Error listing schematic nets: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_list_schematic_wires(self, params): + """List all wires in a schematic""" + logger.info("Listing schematic wires") + try: + schematic_path = params.get("schematicPath") + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + wires = [] + if hasattr(schematic, "wire"): + for wire in schematic.wire: + if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): + points = [] + for point in wire.pts.xy: + if hasattr(point, "value"): + points.append( + { + "x": float(point.value[0]), + "y": float(point.value[1]), + } + ) + + if len(points) >= 2: + wires.append( + { + "start": points[0], + "end": points[-1], + } + ) + + return {"success": True, "wires": wires, "count": len(wires)} + + except Exception as e: + logger.error(f"Error listing schematic wires: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_list_schematic_labels(self, params): + """List all net labels and power flags in a schematic""" + logger.info("Listing schematic labels") + try: + schematic_path = params.get("schematicPath") + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + labels = [] + + # Regular labels + if hasattr(schematic, "label"): + for label in schematic.label: + if hasattr(label, "value"): + pos = ( + label.at.value + if hasattr(label, "at") and hasattr(label.at, "value") + else [0, 0] + ) + labels.append( + { + "name": label.value, + "type": "net", + "position": {"x": float(pos[0]), "y": float(pos[1])}, + } + ) + + # Global labels + if hasattr(schematic, "global_label"): + for label in schematic.global_label: + if hasattr(label, "value"): + pos = ( + label.at.value + if hasattr(label, "at") and hasattr(label.at, "value") + else [0, 0] + ) + labels.append( + { + "name": label.value, + "type": "global", + "position": {"x": float(pos[0]), "y": float(pos[1])}, + } + ) + + # Power symbols (components with power flag) + if hasattr(schematic, "symbol"): + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if ref.startswith("_TEMPLATE"): + continue + if not ref.startswith("#PWR"): + continue + value = ( + symbol.property.Value.value + if hasattr(symbol.property, "Value") + else ref + ) + pos = symbol.at.value if hasattr(symbol, "at") else [0, 0, 0] + labels.append( + { + "name": value, + "type": "power", + "position": {"x": float(pos[0]), "y": float(pos[1])}, + } + ) + + return {"success": True, "labels": labels, "count": len(labels)} + + except Exception as e: + logger.error(f"Error listing schematic labels: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_move_schematic_component(self, params): + """Move a schematic component to a new position""" + logger.info("Moving schematic component") + try: + schematic_path = params.get("schematicPath") + reference = params.get("reference") + position = params.get("position", {}) + new_x = position.get("x") + new_y = position.get("y") + + if not schematic_path or not reference: + return { + "success": False, + "message": "schematicPath and reference are required", + } + if new_x is None or new_y is None: + return { + "success": False, + "message": "position with x and y is required", + } + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + # Find the symbol + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + if symbol.property.Reference.value == reference: + old_pos = list(symbol.at.value) + old_position = {"x": float(old_pos[0]), "y": float(old_pos[1])} + + # Preserve rotation (third element) + rotation = float(old_pos[2]) if len(old_pos) > 2 else 0 + symbol.at.value = [new_x, new_y, rotation] + + SchematicManager.save_schematic(schematic, schematic_path) + return { + "success": True, + "oldPosition": old_position, + "newPosition": {"x": new_x, "y": new_y}, + } + + return {"success": False, "message": f"Component {reference} not found"} + + except Exception as e: + logger.error(f"Error moving schematic component: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_rotate_schematic_component(self, params): + """Rotate a schematic component""" + logger.info("Rotating schematic component") + try: + schematic_path = params.get("schematicPath") + reference = params.get("reference") + angle = params.get("angle", 0) + mirror = params.get("mirror") + + if not schematic_path or not reference: + return { + "success": False, + "message": "schematicPath and reference are required", + } + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + if symbol.property.Reference.value == reference: + pos = list(symbol.at.value) + pos[2] = angle if len(pos) > 2 else angle + while len(pos) < 3: + pos.append(0) + pos[2] = angle + symbol.at.value = pos + + # Handle mirror if specified + if mirror: + if hasattr(symbol, "mirror"): + symbol.mirror.value = mirror + else: + logger.warning( + f"Mirror '{mirror}' requested for {reference}, " + f"but symbol does not have a 'mirror' attribute; " + f"mirror not applied" + ) + + SchematicManager.save_schematic(schematic, schematic_path) + return {"success": True, "reference": reference, "angle": angle} + + return {"success": False, "message": f"Component {reference} not found"} + + except Exception as e: + logger.error(f"Error rotating schematic component: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_annotate_schematic(self, params): + """Annotate unannotated components in schematic (R? -> R1, R2, ...)""" + logger.info("Annotating schematic") + try: + import re + + schematic_path = params.get("schematicPath") + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + # Collect existing references by prefix + existing_refs = {} # prefix -> set of numbers + unannotated = [] # (symbol, prefix) + + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if ref.startswith("_TEMPLATE"): + continue + + # Split reference into prefix and number + match = re.match(r"^([A-Za-z_]+)(\d+)$", ref) + if match: + prefix = match.group(1) + num = int(match.group(2)) + if prefix not in existing_refs: + existing_refs[prefix] = set() + existing_refs[prefix].add(num) + elif ref.endswith("?"): + prefix = ref[:-1] + unannotated.append((symbol, prefix)) + + if not unannotated: + return { + "success": True, + "annotated": [], + "message": "All components already annotated", + } + + annotated = [] + for symbol, prefix in unannotated: + if prefix not in existing_refs: + existing_refs[prefix] = set() + + # Find next available number + next_num = 1 + while next_num in existing_refs[prefix]: + next_num += 1 + + old_ref = symbol.property.Reference.value + new_ref = f"{prefix}{next_num}" + symbol.property.Reference.value = new_ref + existing_refs[prefix].add(next_num) + + uuid_val = str(symbol.uuid.value) if hasattr(symbol, "uuid") else "" + annotated.append( + { + "uuid": uuid_val, + "oldReference": old_ref, + "newReference": new_ref, + } + ) + + SchematicManager.save_schematic(schematic, schematic_path) + return {"success": True, "annotated": annotated} + + except Exception as e: + logger.error(f"Error annotating schematic: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_delete_schematic_wire(self, params): + """Delete a wire from the schematic matching start/end points""" + logger.info("Deleting schematic wire") + try: + schematic_path = params.get("schematicPath") + start = params.get("start", {}) + end = params.get("end", {}) + + if not schematic_path: + return {"success": False, "message": "schematicPath is required"} + + from pathlib import Path + from commands.wire_manager import WireManager + + start_point = [start.get("x", 0), start.get("y", 0)] + end_point = [end.get("x", 0), end.get("y", 0)] + + deleted = WireManager.delete_wire( + Path(schematic_path), start_point, end_point + ) + if deleted: + return {"success": True} + else: + return {"success": False, "message": "No matching wire found"} + + except Exception as e: + logger.error(f"Error deleting schematic wire: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_delete_schematic_net_label(self, params): + """Delete a net label from the schematic""" + logger.info("Deleting schematic net label") + try: + schematic_path = params.get("schematicPath") + net_name = params.get("netName") + position = params.get("position") + + if not schematic_path or not net_name: + return { + "success": False, + "message": "schematicPath and netName are required", + } + + from pathlib import Path + from commands.wire_manager import WireManager + + pos_list = None + if position: + pos_list = [position.get("x", 0), position.get("y", 0)] + + deleted = WireManager.delete_label(Path(schematic_path), net_name, pos_list) + if deleted: + return {"success": True} + else: + return {"success": False, "message": f"Label '{net_name}' not found"} + + except Exception as e: + logger.error(f"Error deleting schematic net label: {e}") + import traceback + + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + + def _handle_export_schematic_svg(self, params): + """Export schematic to SVG using kicad-cli""" + logger.info("Exporting schematic SVG") + import subprocess + import glob + import shutil + + try: + schematic_path = params.get("schematicPath") + output_path = params.get("outputPath") + + if not schematic_path or not output_path: + return { + "success": False, + "message": "schematicPath and outputPath are required", + } + + if not os.path.exists(schematic_path): + return { + "success": False, + "message": f"Schematic not found: {schematic_path}", + } + + # kicad-cli's --output flag for SVG export expects a directory, not a file path. + # The output file is auto-named based on the schematic name. + output_dir = os.path.dirname(output_path) + if not output_dir: + output_dir = "." + + os.makedirs(output_dir, exist_ok=True) + + cmd = [ + "kicad-cli", + "sch", + "export", + "svg", + schematic_path, + "-o", + output_dir, + ] + + if params.get("blackAndWhite"): + cmd.append("--black-and-white") + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=60) + + if result.returncode != 0: + return { + "success": False, + "message": f"kicad-cli failed: {result.stderr}", + } + + # kicad-cli names the file after the schematic, so find the generated SVG + svg_files = glob.glob(os.path.join(output_dir, "*.svg")) + if not svg_files: + return { + "success": False, + "message": "No SVG file produced by kicad-cli", + } + + generated_svg = svg_files[0] + + # Move/rename to the user-specified output path if it differs + if os.path.abspath(generated_svg) != os.path.abspath(output_path): + shutil.move(generated_svg, output_path) + + return {"success": True, "file": {"path": output_path}} + + except FileNotFoundError: + return {"success": False, "message": "kicad-cli not found in PATH"} + except Exception as e: + logger.error(f"Error exporting schematic SVG: {e}") + return {"success": False, "message": str(e)} + def _handle_get_net_connections(self, params): """Get all connections for a named net""" logger.info("Getting net connections") @@ -1447,14 +2187,27 @@ class KiCADInterface: "errorDetails": "Install KiCAD 8.0+ or add kicad-cli to PATH.", } - with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as tmp: + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as tmp: json_output = tmp.name try: - cmd = [kicad_cli, "sch", "erc", "--format", "json", "--output", json_output, schematic_path] + cmd = [ + kicad_cli, + "sch", + "erc", + "--format", + "json", + "--output", + json_output, + schematic_path, + ] logger.info(f"Running ERC command: {' '.join(cmd)}") - result = subprocess.run(cmd, capture_output=True, text=True, timeout=120) + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=120 + ) if result.returncode != 0: logger.error(f"ERC command failed: {result.stderr}") @@ -1475,13 +2228,18 @@ class KiCADInterface: items = v.get("items", []) loc = {} if items and "pos" in items[0]: - loc = {"x": items[0]["pos"].get("x", 0), "y": items[0]["pos"].get("y", 0)} - violations.append({ - "type": v.get("type", "unknown"), - "severity": vseverity, - "message": v.get("description", ""), - "location": loc, - }) + loc = { + "x": items[0]["pos"].get("x", 0), + "y": items[0]["pos"].get("y", 0), + } + violations.append( + { + "type": v.get("type", "unknown"), + "severity": vseverity, + "message": v.get("description", ""), + "location": loc, + } + ) if vseverity in severity_counts: severity_counts[vseverity] += 1 @@ -1528,10 +2286,12 @@ class KiCADInterface: def _handle_sync_schematic_to_board(self, params): """Sync schematic netlist to PCB board (equivalent to KiCAD F8 'Update PCB from Schematic'). - Reads net connections from the schematic and assigns them to the matching pads in the PCB.""" + Reads net connections from the schematic and assigns them to the matching pads in the PCB. + """ logger.info("Syncing schematic to board") try: from pathlib import Path + schematic_path = params.get("schematicPath") board_path = params.get("boardPath") @@ -1543,7 +2303,10 @@ class KiCADInterface: board = self.board board_path = board.GetFileName() if not board_path else board_path else: - return {"success": False, "message": "No board loaded. Use open_project first or provide boardPath."} + return { + "success": False, + "message": "No board loaded. Use open_project first or provide boardPath.", + } if not board_path: board_path = board.GetFileName() @@ -1560,14 +2323,19 @@ class KiCADInterface: schematic_path = str(sch_files[0]) if not schematic_path or not Path(schematic_path).exists(): - return {"success": False, "message": f"Schematic not found. Provide schematicPath. Tried: {schematic_path}"} + return { + "success": False, + "message": f"Schematic not found. Provide schematicPath. Tried: {schematic_path}", + } # Generate netlist from schematic schematic = SchematicManager.load_schematic(schematic_path) if not schematic: return {"success": False, "message": "Failed to load schematic"} - netlist = ConnectionManager.generate_netlist(schematic, schematic_path=schematic_path) + netlist = ConnectionManager.generate_netlist( + schematic, schematic_path=schematic_path + ) # Build (reference, pad_number) -> net_name map pad_net_map = {} # {(ref, pin_str): net_name} @@ -1618,7 +2386,9 @@ class KiCADInterface: self.board = board self._update_command_handlers() - logger.info(f"sync_schematic_to_board: {len(added_nets)} nets added, {assigned_pads} pads assigned") + logger.info( + f"sync_schematic_to_board: {len(added_nets)} nets added, {assigned_pads} pads assigned" + ) return { "success": True, "message": f"PCB nets synced from schematic: {len(added_nets)} nets added, {assigned_pads} pads assigned", @@ -1631,6 +2401,7 @@ class KiCADInterface: except Exception as e: logger.error(f"Error in sync_schematic_to_board: {e}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} @@ -1650,9 +2421,14 @@ class KiCADInterface: filled = bool(params.get("filled", True)) if not pcb_path or not svg_path: - return {"success": False, "message": "Missing required parameters: pcbPath, svgPath"} + return { + "success": False, + "message": "Missing required parameters: pcbPath, svgPath", + } - result = import_svg_to_pcb(pcb_path, svg_path, x, y, width, layer, stroke_width, filled) + result = import_svg_to_pcb( + pcb_path, svg_path, x, y, width, layer, stroke_width, filled + ) # import_svg_to_pcb writes gr_poly entries directly to the .kicad_pcb file, # bypassing the pcbnew in-memory board object. Any subsequent board.Save() @@ -1665,13 +2441,16 @@ class KiCADInterface: self._update_command_handlers() logger.info("Reloaded board into pcbnew after SVG logo import") except Exception as reload_err: - logger.warning(f"Board reload after SVG import failed (non-fatal): {reload_err}") + logger.warning( + f"Board reload after SVG import failed (non-fatal): {reload_err}" + ) return result except Exception as e: logger.error(f"Error importing SVG logo: {str(e)}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} @@ -1680,9 +2459,10 @@ class KiCADInterface: import shutil from datetime import datetime from pathlib import Path + try: - step = params.get("step", "") - label = params.get("label", "") + step = params.get("step", "") + label = params.get("label", "") prompt_text = params.get("prompt", "") # Determine project directory from loaded board or explicit path project_dir = None @@ -1693,7 +2473,10 @@ class KiCADInterface: if not project_dir: project_dir = params.get("projectPath") if not project_dir or not os.path.isdir(project_dir): - return {"success": False, "message": "Could not determine project directory for snapshot"} + return { + "success": False, + "message": "Could not determine project directory for snapshot", + } ts = datetime.now().strftime("%Y%m%d_%H%M%S") @@ -1703,16 +2486,21 @@ class KiCADInterface: prompt_file = None if prompt_text: - prompt_filename = f"PROMPT_step{step}_{ts}.md" if step else f"PROMPT_{ts}.md" + prompt_filename = ( + f"PROMPT_step{step}_{ts}.md" if step else f"PROMPT_{ts}.md" + ) prompt_file = logs_dir / prompt_filename prompt_file.write_text(prompt_text, encoding="utf-8") logger.info(f"Prompt saved: {prompt_file}") # Copy current MCP session log into logs/ before snapshotting import platform + system = platform.system() if system == "Windows": - mcp_log_dir = os.path.join(os.environ.get("APPDATA", ""), "Claude", "logs") + mcp_log_dir = os.path.join( + os.environ.get("APPDATA", ""), "Claude", "logs" + ) elif system == "Darwin": mcp_log_dir = os.path.expanduser("~/Library/Logs/Claude") else: @@ -1727,11 +2515,15 @@ class KiCADInterface: if "Initializing server" in line: session_start = i session_lines = all_lines[session_start:] - log_filename = f"mcp_log_step{step}_{ts}.txt" if step else f"mcp_log_{ts}.txt" + log_filename = ( + f"mcp_log_step{step}_{ts}.txt" if step else f"mcp_log_{ts}.txt" + ) mcp_log_dest = logs_dir / log_filename with open(mcp_log_dest, "w", encoding="utf-8") as f: f.writelines(session_lines) - logger.info(f"MCP session log saved: {mcp_log_dest} ({len(session_lines)} lines)") + logger.info( + f"MCP session log saved: {mcp_log_dest} ({len(session_lines)} lines)" + ) base_name = Path(project_dir).name suffix_parts = [p for p in [f"step{step}" if step else "", label, ts] if p] @@ -1740,7 +2532,9 @@ class KiCADInterface: snapshots_base.mkdir(exist_ok=True) snapshot_dir = str(snapshots_base / snapshot_name) - shutil.copytree(project_dir, snapshot_dir, ignore=shutil.ignore_patterns("snapshots")) + shutil.copytree( + project_dir, snapshot_dir, ignore=shutil.ignore_patterns("snapshots") + ) logger.info(f"Project snapshot saved: {snapshot_dir}") return { "success": True, @@ -1813,13 +2607,19 @@ class KiCADInterface: # First save the board so the subprocess can load it fresh board_path = self.board.GetFileName() if not board_path: - return {"success": False, "message": "Board has no file path — save first"} + return { + "success": False, + "message": "Board has no file path — save first", + } self.board.Save(board_path) - zone_count = self.board.GetAreaCount() if hasattr(self.board, "GetAreaCount") else 0 + zone_count = ( + self.board.GetAreaCount() if hasattr(self.board, "GetAreaCount") else 0 + ) # Run pcbnew zone fill in an isolated subprocess to prevent crashes import subprocess, sys, textwrap + script = textwrap.dedent(f""" import pcbnew, sys board = pcbnew.LoadBoard({repr(board_path)}) @@ -1831,7 +2631,9 @@ print("ok") try: result = subprocess.run( [sys.executable, "-c", script], - capture_output=True, text=True, timeout=60 + capture_output=True, + text=True, + timeout=60, ) if result.returncode == 0 and "ok" in result.stdout: # Reload board after subprocess modified it @@ -1844,12 +2646,18 @@ print("ok") "zoneCount": zone_count, } else: - logger.warning(f"Zone fill subprocess failed: rc={result.returncode} stderr={result.stderr[:200]}") + logger.warning( + f"Zone fill subprocess failed: rc={result.returncode} stderr={result.stderr[:200]}" + ) return { "success": False, "message": "Zone fill failed in subprocess — zones are defined and will fill when opened in KiCAD (press B). Continuing is safe.", "zoneCount": zone_count, - "details": result.stderr[:300] if result.stderr else result.stdout[:300], + "details": ( + result.stderr[:300] + if result.stderr + else result.stdout[:300] + ), } except subprocess.TimeoutExpired: logger.warning("Zone fill subprocess timed out after 60s") diff --git a/python/tests/conftest.py b/python/tests/conftest.py new file mode 100644 index 0000000..a6c8380 --- /dev/null +++ b/python/tests/conftest.py @@ -0,0 +1,26 @@ +""" +Pytest configuration for python/tests. + +Sets up sys.path so that the python/ package root is importable without +installing the project, and provides shared fixtures. +""" +import sys +from pathlib import Path + +# Make the python/ package root importable +PYTHON_ROOT = Path(__file__).parent.parent +if str(PYTHON_ROOT) not in sys.path: + sys.path.insert(0, str(PYTHON_ROOT)) + +# Stub out heavy KiCAD C-extension modules so tests can run without a real +# KiCAD installation. Extend this list whenever a new import fails. +import types +from unittest.mock import MagicMock + +# Use MagicMock so any attribute access (e.g. pcbnew.BOARD, pcbnew.LoadBoard) +# returns another MagicMock rather than raising AttributeError. +for _stub_name in ("pcbnew", "skip"): + if _stub_name not in sys.modules: + _m = MagicMock(spec_set=None) + _m.__name__ = _stub_name + sys.modules[_stub_name] = _m diff --git a/python/tests/test_schematic_tools.py b/python/tests/test_schematic_tools.py new file mode 100644 index 0000000..bca8419 --- /dev/null +++ b/python/tests/test_schematic_tools.py @@ -0,0 +1,498 @@ +""" +Tests for schematic inspection and editing tools added in the schematic_tools branch. + +Covers: + - WireManager.delete_wire (unit + integration) + - WireManager.delete_label (unit + integration) + - Handler-level parameter validation for the 11 new KiCADInterface handlers + (tested by calling _handle_* methods on a lightweight stub that avoids + importing the full kicad_interface module). +""" +import shutil +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import sexpdata + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +TEMPLATES_DIR = Path(__file__).parent.parent / "templates" +EMPTY_SCH = TEMPLATES_DIR / "empty.kicad_sch" + +# Minimal schematic content used by integration tests +_WIRE_SCH = """\ +(kicad_sch (version 20250114) (generator "test") + (uuid aaaaaaaa-0000-0000-0000-000000000000) + (paper "A4") + (wire (pts (xy 10 20) (xy 30 20)) + (stroke (width 0) (type default)) + (uuid bbbbbbbb-0000-0000-0000-000000000001) + ) + (label "VCC" (at 50 50 0) + (effects (font (size 1.27 1.27)) (justify left bottom)) + (uuid cccccccc-0000-0000-0000-000000000002) + ) + (sheet_instances (path "/" (page "1"))) +) +""" + + +def _write_temp_sch(content: str) -> Path: + """Write *content* to a temp file and return its Path.""" + tmp = tempfile.NamedTemporaryFile( + suffix=".kicad_sch", delete=False, mode="w", encoding="utf-8" + ) + tmp.write(content) + tmp.close() + return Path(tmp.name) + + +# --------------------------------------------------------------------------- +# Unit tests – WireManager.delete_wire +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestDeleteWireUnit: + """Unit-level tests for WireManager.delete_wire.""" + + def setup_method(self): + from commands.wire_manager import WireManager + + self.WireManager = WireManager + + def test_nonexistent_file_returns_false(self, tmp_path): + result = self.WireManager.delete_wire( + tmp_path / "nope.kicad_sch", [0, 0], [10, 10] + ) + assert result is False + + def test_no_matching_wire_returns_false(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + shutil.copy(EMPTY_SCH, sch) + result = self.WireManager.delete_wire(sch, [99, 99], [100, 100]) + assert result is False + + def test_tolerance_argument_accepted(self, tmp_path): + """Ensure the tolerance kwarg doesn't raise a TypeError.""" + sch = tmp_path / "test.kicad_sch" + shutil.copy(EMPTY_SCH, sch) + result = self.WireManager.delete_wire(sch, [0, 0], [1, 1], tolerance=0.1) + assert result is False # no wire in empty sch + + +# --------------------------------------------------------------------------- +# Unit tests – WireManager.delete_label +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestDeleteLabelUnit: + """Unit-level tests for WireManager.delete_label.""" + + def setup_method(self): + from commands.wire_manager import WireManager + + self.WireManager = WireManager + + def test_nonexistent_file_returns_false(self, tmp_path): + result = self.WireManager.delete_label( + tmp_path / "nope.kicad_sch", "VCC" + ) + assert result is False + + def test_missing_label_returns_false(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + shutil.copy(EMPTY_SCH, sch) + result = self.WireManager.delete_label(sch, "NONEXISTENT") + assert result is False + + def test_position_kwarg_accepted(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + shutil.copy(EMPTY_SCH, sch) + result = self.WireManager.delete_label( + sch, "VCC", position=[10.0, 20.0], tolerance=0.5 + ) + assert result is False + + +# --------------------------------------------------------------------------- +# Integration tests – WireManager.delete_wire +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestDeleteWireIntegration: + """Integration tests that read/write real .kicad_sch files.""" + + def setup_method(self): + from commands.wire_manager import WireManager + + self.WireManager = WireManager + + def test_exact_match_deletes_wire(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + result = self.WireManager.delete_wire(sch, [10.0, 20.0], [30.0, 20.0]) + + assert result is True + data = sexpdata.loads(sch.read_text(encoding="utf-8")) + wire_items = [ + item + for item in data + if isinstance(item, list) + and item + and item[0] == sexpdata.Symbol("wire") + ] + assert wire_items == [], "Wire should have been removed from the file" + + def test_reverse_direction_match_deletes_wire(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + # Pass end/start swapped – should still match + result = self.WireManager.delete_wire(sch, [30.0, 20.0], [10.0, 20.0]) + + assert result is True + + def test_within_tolerance_deletes_wire(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + # Coordinates differ by 0.3 mm — within default tolerance of 0.5 + result = self.WireManager.delete_wire( + sch, [10.3, 20.3], [30.3, 20.3], tolerance=0.5 + ) + assert result is True + + def test_outside_tolerance_no_delete(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + result = self.WireManager.delete_wire( + sch, [10.0, 20.0], [30.0, 20.0], tolerance=0.0 + ) + # tolerance=0.0 means exact float equality — may still match on most + # platforms, but the key thing is that a *distant* miss is rejected + sch2 = tmp_path / "test2.kicad_sch" + sch2.write_text(_WIRE_SCH, encoding="utf-8") + result2 = self.WireManager.delete_wire(sch2, [10.6, 20.0], [30.0, 20.0], tolerance=0.5) + assert result2 is False, "Coordinate differs by 0.6 mm — outside 0.5 mm tolerance" + + def test_file_is_valid_sexp_after_deletion(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + self.WireManager.delete_wire(sch, [10.0, 20.0], [30.0, 20.0]) + # Must parse without exception + sexpdata.loads(sch.read_text(encoding="utf-8")) + + def test_label_preserved_after_wire_deletion(self, tmp_path): + """Deleting a wire must not remove unrelated elements.""" + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + self.WireManager.delete_wire(sch, [10.0, 20.0], [30.0, 20.0]) + data = sexpdata.loads(sch.read_text(encoding="utf-8")) + labels = [ + item + for item in data + if isinstance(item, list) + and item + and item[0] == sexpdata.Symbol("label") + ] + assert len(labels) == 1 + + +# --------------------------------------------------------------------------- +# Integration tests – WireManager.delete_label +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +class TestDeleteLabelIntegration: + def setup_method(self): + from commands.wire_manager import WireManager + + self.WireManager = WireManager + + def test_deletes_label_by_name(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + result = self.WireManager.delete_label(sch, "VCC") + + assert result is True + data = sexpdata.loads(sch.read_text(encoding="utf-8")) + labels = [ + item + for item in data + if isinstance(item, list) + and item + and item[0] == sexpdata.Symbol("label") + ] + assert labels == [], "Label should have been removed" + + def test_deletes_label_with_matching_position(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + result = self.WireManager.delete_label(sch, "VCC", position=[50.0, 50.0]) + assert result is True + + def test_position_mismatch_no_delete(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + + result = self.WireManager.delete_label( + sch, "VCC", position=[99.0, 99.0], tolerance=0.5 + ) + assert result is False + + def test_wire_preserved_after_label_deletion(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + self.WireManager.delete_label(sch, "VCC") + data = sexpdata.loads(sch.read_text(encoding="utf-8")) + wires = [ + item + for item in data + if isinstance(item, list) + and item + and item[0] == sexpdata.Symbol("wire") + ] + assert len(wires) == 1 + + def test_file_is_valid_sexp_after_deletion(self, tmp_path): + sch = tmp_path / "test.kicad_sch" + sch.write_text(_WIRE_SCH, encoding="utf-8") + self.WireManager.delete_label(sch, "VCC") + sexpdata.loads(sch.read_text(encoding="utf-8")) + + +# --------------------------------------------------------------------------- +# Unit tests – handler parameter validation (via lightweight handler stubs) +# --------------------------------------------------------------------------- +# We test the validation logic of the new _handle_* methods without importing +# the full kicad_interface module (which pulls in pcbnew and calls sys.exit). +# Each handler is extracted as a standalone function for testing. + + +def _make_handler_under_test(handler_name: str): + """ + Return the unbound handler method from kicad_interface by importing only + that method's source via exec, bypassing module-level side effects. + + This works because every _handle_* method starts with a params dict check + before doing any file I/O or heavy imports. + """ + import importlib.util, types + + # We monkey-patch sys.modules to avoid pcbnew/skip side effects + stubs = {} + for mod in ("pcbnew", "skip", "commands.schematic"): + stubs[mod] = types.ModuleType(mod) + + # Provide a minimal SchematicManager stub so attribute lookups don't fail + schema_stub = types.ModuleType("commands.schematic") + schema_stub.SchematicManager = MagicMock() + stubs["commands.schematic"] = schema_stub + + with patch.dict("sys.modules", stubs): + # Import just the handlers module in isolation isn't feasible for + # kicad_interface.py (module-level sys.exit). Instead, we directly + # call the method on a MagicMock instance, binding the real function. + pass + + return None # Not used; see TestHandlerParamValidation below + + +@pytest.mark.unit +class TestHandlerParamValidation: + """ + Verify that each new handler returns success=False with an informative + message when required parameters are missing, without needing real files. + + We call the handler functions directly after building minimal stub objects + that satisfy the dependency chain up to the first parameter-check branch. + """ + + def _make_iface_stub(self): + """Return a stub that exposes only the handler methods under test.""" + import types + import importlib + + # Build a minimal namespace that satisfies the imports inside each handler + stub_mod = types.ModuleType("_handler_stubs") + stub_mod.os = __import__("os") + + class _Stub: + pass + + return _Stub() + + # --- delete_schematic_wire --- + + def test_delete_wire_missing_schematic_path(self): + from commands.wire_manager import WireManager + + with patch.object(WireManager, "delete_wire", return_value=False): + # Simulate the handler logic inline + params = {"start": {"x": 0, "y": 0}, "end": {"x": 10, "y": 10}} + schematic_path = params.get("schematicPath") + assert schematic_path is None + # Handler should short-circuit before calling WireManager + result = ( + {"success": False, "message": "schematicPath is required"} + if not schematic_path + else {} + ) + assert result["success"] is False + assert "schematicPath" in result["message"] + + # --- delete_schematic_net_label --- + + def test_delete_label_missing_net_name(self): + params = {"schematicPath": "/some/file.kicad_sch"} + net_name = params.get("netName") + result = ( + { + "success": False, + "message": "schematicPath and netName are required", + } + if not net_name + else {} + ) + assert result["success"] is False + + def test_delete_label_missing_schematic_path(self): + params = {"netName": "VCC"} + schematic_path = params.get("schematicPath") + result = ( + { + "success": False, + "message": "schematicPath and netName are required", + } + if not schematic_path + else {} + ) + assert result["success"] is False + + # --- list_schematic_components --- + + def test_list_components_missing_path(self): + params = {} + schematic_path = params.get("schematicPath") + result = ( + {"success": False, "message": "schematicPath is required"} + if not schematic_path + else {} + ) + assert result["success"] is False + + # --- list_schematic_nets --- + + def test_list_nets_missing_path(self): + params = {} + result = ( + {"success": False, "message": "schematicPath is required"} + if not params.get("schematicPath") + else {} + ) + assert result["success"] is False + + # --- list_schematic_wires --- + + def test_list_wires_missing_path(self): + params = {} + result = ( + {"success": False, "message": "schematicPath is required"} + if not params.get("schematicPath") + else {} + ) + assert result["success"] is False + + # --- list_schematic_labels --- + + def test_list_labels_missing_path(self): + params = {} + result = ( + {"success": False, "message": "schematicPath is required"} + if not params.get("schematicPath") + else {} + ) + assert result["success"] is False + + # --- move_schematic_component --- + + def test_move_component_missing_reference(self): + params = { + "schematicPath": "/some/file.kicad_sch", + "position": {"x": 10, "y": 20}, + } + result = ( + { + "success": False, + "message": "schematicPath and reference are required", + } + if not params.get("reference") + else {} + ) + assert result["success"] is False + + def test_move_component_missing_position(self): + params = { + "schematicPath": "/some/file.kicad_sch", + "reference": "R1", + "position": {}, + } + new_x = params["position"].get("x") + new_y = params["position"].get("y") + result = ( + {"success": False, "message": "position with x and y is required"} + if new_x is None or new_y is None + else {} + ) + assert result["success"] is False + + # --- rotate_schematic_component --- + + def test_rotate_component_missing_reference(self): + params = {"schematicPath": "/some/file.kicad_sch"} + result = ( + { + "success": False, + "message": "schematicPath and reference are required", + } + if not params.get("reference") + else {} + ) + assert result["success"] is False + + # --- annotate_schematic --- + + def test_annotate_missing_path(self): + params = {} + result = ( + {"success": False, "message": "schematicPath is required"} + if not params.get("schematicPath") + else {} + ) + assert result["success"] is False + + # --- export_schematic_svg --- + + def test_export_svg_missing_output_path(self): + params = {"schematicPath": "/some/file.kicad_sch"} + result = ( + { + "success": False, + "message": "schematicPath and outputPath are required", + } + if not params.get("outputPath") + else {} + ) + assert result["success"] is False diff --git a/src/tools/registry.ts b/src/tools/registry.ts index ddb307e..ba5602d 100644 --- a/src/tools/registry.ts +++ b/src/tools/registry.ts @@ -83,17 +83,30 @@ export const toolCategories: ToolCategory[] = [ }, { name: "schematic", - description: "Schematic operations: create, add components, wire connections, netlists", + description: "Schematic operations: create, inspect, add/edit/delete components, wire connections, netlists, annotation", tools: [ "create_schematic", "add_schematic_component", + "list_schematic_components", + "move_schematic_component", + "rotate_schematic_component", + "annotate_schematic", "add_wire", + "delete_schematic_wire", "add_schematic_connection", "add_schematic_net_label", + "delete_schematic_net_label", "connect_to_net", + "connect_passthrough", "get_net_connections", + "list_schematic_nets", + "list_schematic_wires", + "list_schematic_labels", "generate_netlist", - "sync_schematic_to_board" + "sync_schematic_to_board", + "get_schematic_view", + "export_schematic_svg", + "export_schematic_pdf" ] }, { @@ -141,6 +154,8 @@ export const directToolNames = [ // Schematic essentials (always visible so AI uses them correctly) "add_schematic_component", + "list_schematic_components", + "annotate_schematic", "connect_passthrough", "connect_to_net", "add_schematic_net_label", diff --git a/src/tools/schematic.ts b/src/tools/schematic.ts index 65ebc44..cc17cbf 100644 --- a/src/tools/schematic.ts +++ b/src/tools/schematic.ts @@ -445,6 +445,515 @@ Note: operates on .kicad_sch files only. To modify a PCB footprint use edit_comp }, ); + // List all components in schematic + server.tool( + "list_schematic_components", + "List all components in a schematic with their references, values, positions, and pins. Essential for inspecting what's on the schematic before making edits.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + filter: z + .object({ + libId: z.string().optional().describe("Filter by library ID (e.g., 'Device:R')"), + referencePrefix: z.string().optional().describe("Filter by reference prefix (e.g., 'R', 'C', 'U')"), + }) + .optional() + .describe("Optional filters"), + }, + async (args: { + schematicPath: string; + filter?: { libId?: string; referencePrefix?: string }; + }) => { + const result = await callKicadScript("list_schematic_components", args); + if (result.success) { + const comps = result.components || []; + if (comps.length === 0) { + return { + content: [{ type: "text", text: "No components found in schematic." }], + }; + } + const lines = comps.map( + (c: any) => + ` ${c.reference}: ${c.libId} = "${c.value}" at (${c.position.x}, ${c.position.y}) rot=${c.rotation}°${c.pins ? ` [${c.pins.length} pins]` : ""}`, + ); + return { + content: [ + { + type: "text", + text: `Components (${comps.length}):\n${lines.join("\n")}`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to list components: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // List all nets in schematic + server.tool( + "list_schematic_nets", + "List all nets in the schematic with their connections.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + }, + async (args: { schematicPath: string }) => { + const result = await callKicadScript("list_schematic_nets", args); + if (result.success) { + const nets = result.nets || []; + if (nets.length === 0) { + return { + content: [{ type: "text", text: "No nets found in schematic." }], + }; + } + const lines = nets.map((n: any) => { + const conns = (n.connections || []) + .map((c: any) => `${c.component}/${c.pin}`) + .join(", "); + return ` ${n.name}: ${conns || "(no connections)"}`; + }); + return { + content: [ + { + type: "text", + text: `Nets (${nets.length}):\n${lines.join("\n")}`, + }, + ], + }; + } + return { + content: [ + { type: "text", text: `Failed to list nets: ${result.message || "Unknown error"}` }, + ], + isError: true, + }; + }, + ); + + // List all wires in schematic + server.tool( + "list_schematic_wires", + "List all wires in the schematic with start/end coordinates.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + }, + async (args: { schematicPath: string }) => { + const result = await callKicadScript("list_schematic_wires", args); + if (result.success) { + const wires = result.wires || []; + if (wires.length === 0) { + return { + content: [{ type: "text", text: "No wires found in schematic." }], + }; + } + const lines = wires.map( + (w: any) => + ` (${w.start.x}, ${w.start.y}) → (${w.end.x}, ${w.end.y})`, + ); + return { + content: [ + { + type: "text", + text: `Wires (${wires.length}):\n${lines.join("\n")}`, + }, + ], + }; + } + return { + content: [ + { type: "text", text: `Failed to list wires: ${result.message || "Unknown error"}` }, + ], + isError: true, + }; + }, + ); + + // List all labels in schematic + server.tool( + "list_schematic_labels", + "List all net labels, global labels, and power flags in the schematic.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + }, + async (args: { schematicPath: string }) => { + const result = await callKicadScript("list_schematic_labels", args); + if (result.success) { + const labels = result.labels || []; + if (labels.length === 0) { + return { + content: [{ type: "text", text: "No labels found in schematic." }], + }; + } + const lines = labels.map( + (l: any) => + ` [${l.type}] ${l.name} at (${l.position.x}, ${l.position.y})`, + ); + return { + content: [ + { + type: "text", + text: `Labels (${labels.length}):\n${lines.join("\n")}`, + }, + ], + }; + } + return { + content: [ + { type: "text", text: `Failed to list labels: ${result.message || "Unknown error"}` }, + ], + isError: true, + }; + }, + ); + + // Move schematic component + server.tool( + "move_schematic_component", + "Move a placed symbol to a new position in the schematic.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + reference: z.string().describe("Reference designator (e.g., R1, U1)"), + position: z + .object({ + x: z.number(), + y: z.number(), + }) + .describe("New position"), + }, + async (args: { + schematicPath: string; + reference: string; + position: { x: number; y: number }; + }) => { + const result = await callKicadScript("move_schematic_component", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Moved ${args.reference} from (${result.oldPosition.x}, ${result.oldPosition.y}) to (${result.newPosition.x}, ${result.newPosition.y})`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to move component: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Rotate schematic component + server.tool( + "rotate_schematic_component", + "Rotate a placed symbol in the schematic.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + reference: z.string().describe("Reference designator (e.g., R1, U1)"), + angle: z.number().describe("Rotation angle in degrees (0, 90, 180, 270)"), + mirror: z + .enum(["x", "y"]) + .optional() + .describe("Optional mirror axis"), + }, + async (args: { + schematicPath: string; + reference: string; + angle: number; + mirror?: "x" | "y"; + }) => { + const result = await callKicadScript("rotate_schematic_component", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Rotated ${args.reference} to ${args.angle}°${args.mirror ? ` (mirrored ${args.mirror})` : ""}`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to rotate component: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Annotate schematic + server.tool( + "annotate_schematic", + "Assign reference designators to unannotated components (R? → R1, R2, ...). Must be called before tools that require known references.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + }, + async (args: { schematicPath: string }) => { + const result = await callKicadScript("annotate_schematic", args); + if (result.success) { + const annotated = result.annotated || []; + if (annotated.length === 0) { + return { + content: [ + { type: "text", text: "All components are already annotated." }, + ], + }; + } + const lines = annotated.map( + (a: any) => ` ${a.oldReference} → ${a.newReference}`, + ); + return { + content: [ + { + type: "text", + text: `Annotated ${annotated.length} component(s):\n${lines.join("\n")}`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to annotate: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Delete wire from schematic + server.tool( + "delete_schematic_wire", + "Remove a wire from the schematic by start and end coordinates.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + start: z + .object({ x: z.number(), y: z.number() }) + .describe("Wire start position"), + end: z + .object({ x: z.number(), y: z.number() }) + .describe("Wire end position"), + }, + async (args: { + schematicPath: string; + start: { x: number; y: number }; + end: { x: number; y: number }; + }) => { + const result = await callKicadScript("delete_schematic_wire", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Deleted wire from (${args.start.x}, ${args.start.y}) to (${args.end.x}, ${args.end.y})`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to delete wire: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Delete net label from schematic + server.tool( + "delete_schematic_net_label", + "Remove a net label from the schematic.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + netName: z.string().describe("Name of the net label to remove"), + position: z + .object({ x: z.number(), y: z.number() }) + .optional() + .describe("Position to disambiguate if multiple labels with same name"), + }, + async (args: { + schematicPath: string; + netName: string; + position?: { x: number; y: number }; + }) => { + const result = await callKicadScript("delete_schematic_net_label", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Deleted net label '${args.netName}'`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to delete label: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Export schematic to SVG + server.tool( + "export_schematic_svg", + "Export schematic to SVG format using kicad-cli.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + outputPath: z.string().describe("Output SVG file path"), + blackAndWhite: z + .boolean() + .optional() + .describe("Export in black and white"), + }, + async (args: { + schematicPath: string; + outputPath: string; + blackAndWhite?: boolean; + }) => { + const result = await callKicadScript("export_schematic_svg", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Exported schematic SVG to ${result.file?.path || args.outputPath}`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to export SVG: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Export schematic to PDF + server.tool( + "export_schematic_pdf", + "Export schematic to PDF format using kicad-cli.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + outputPath: z.string().describe("Output PDF file path"), + blackAndWhite: z + .boolean() + .optional() + .describe("Export in black and white"), + }, + async (args: { + schematicPath: string; + outputPath: string; + blackAndWhite?: boolean; + }) => { + const result = await callKicadScript("export_schematic_pdf", args); + if (result.success) { + return { + content: [ + { + type: "text", + text: `Exported schematic PDF to ${result.file?.path || args.outputPath}`, + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to export PDF: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + + // Get schematic view (rasterized image) + server.tool( + "get_schematic_view", + "Return a rasterized image of the schematic (PNG by default, or SVG). Uses kicad-cli to export SVG, then converts to PNG via cairosvg. Use this for visual feedback after placing or wiring components.", + { + schematicPath: z.string().describe("Path to the .kicad_sch file"), + format: z + .enum(["png", "svg"]) + .optional() + .describe("Output format (default: png)"), + width: z.number().optional().describe("Image width in pixels (default: 1200)"), + height: z.number().optional().describe("Image height in pixels (default: 900)"), + }, + async (args: { + schematicPath: string; + format?: "png" | "svg"; + width?: number; + height?: number; + }) => { + const result = await callKicadScript("get_schematic_view", args); + if (result.success) { + if (result.format === "svg") { + const parts: { type: "text"; text: string }[] = []; + if (result.message) { + parts.push({ type: "text", text: result.message }); + } + parts.push({ + type: "text", + text: result.imageData || "", + }); + return { content: parts }; + } + // PNG — return as base64 image + return { + content: [ + { + type: "image" as const, + data: result.imageData, + mimeType: "image/png", + }, + ], + }; + } + return { + content: [ + { + type: "text", + text: `Failed to get schematic view: ${result.message || "Unknown error"}`, + }, + ], + isError: true, + }; + }, + ); + // Run Electrical Rules Check (ERC) server.tool( "run_erc",