From 6633cd59fd8051a1f7c9d5dca3dfa2485f545b00 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 13:49:20 +0000 Subject: [PATCH 1/9] feat: add get_wire_connections tool for pin/wire lookup by schematic point Given a single (x,y) coordinate on the schematic, flood-fills through all connected wire segments and returns every component pin reachable on that net, plus the full list of wire segments with their start/end coordinates. Co-Authored-By: Claude Sonnet 4.6 --- python/kicad_interface.py | 124 ++++++++++++++++++++++++++++++++++++++ src/tools/registry.ts | 1 + src/tools/schematic.ts | 39 ++++++++++++ 3 files changed, 164 insertions(+) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 3a9ef26..bda174a 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -382,6 +382,7 @@ class KiCADInterface: "connect_passthrough": self._handle_connect_passthrough, "get_schematic_pin_locations": self._handle_get_schematic_pin_locations, "get_net_connections": self._handle_get_net_connections, + "get_wire_connections": self._handle_get_wire_connections, "run_erc": self._handle_run_erc, "generate_netlist": self._handle_generate_netlist, "sync_schematic_to_board": self._handle_sync_schematic_to_board, @@ -2320,6 +2321,129 @@ class KiCADInterface: logger.error(f"Error getting net connections: {str(e)}") return {"success": False, "message": str(e)} + def _handle_get_wire_connections(self, params): + """Find all component pins reachable from a point via connected wires""" + logger.info("Getting wire connections") + try: + from pathlib import Path + from commands.pin_locator import PinLocator + + schematic_path = params.get("schematicPath") + x = params.get("x") + y = params.get("y") + + if not all([schematic_path, x is not None, y is not None]): + return {"success": False, "message": "Missing required parameters: schematicPath, x, y"} + + x, y = float(x), float(y) + tolerance = 0.5 + + def points_coincide(p1, p2): + return abs(p1[0] - p2[0]) < tolerance and abs(p1[1] - p2[1]) < tolerance + + schematic = SchematicManager.load_schematic(schematic_path) + if not schematic: + return {"success": False, "message": "Failed to load schematic"} + + if not hasattr(schematic, "wire"): + return {"success": False, "message": "Schematic has no wires"} + + # Collect all wires as list of point sequences + all_wires = [] + for wire in schematic.wire: + if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): + pts = [] + for point in wire.pts.xy: + if hasattr(point, "value"): + pts.append([float(point.value[0]), float(point.value[1])]) + if len(pts) >= 2: + all_wires.append(pts) + + # Step 1: Find all seed wires that touch the given point (start or end) + query_point = [x, y] + seed_wires = [ + pts for pts in all_wires + if points_coincide(pts[0], query_point) or points_coincide(pts[-1], query_point) + ] + + if not seed_wires: + return { + "success": False, + "message": f"No wire found at ({x},{y}) within {tolerance}mm tolerance", + } + + # Step 2: Flood-fill through connected wires + connected_wires = list(seed_wires) + frontier = set((pt[0], pt[1]) for pts in seed_wires for pt in pts) + remaining = [w for w in all_wires if w not in seed_wires] + changed = True + while changed: + changed = False + still_remaining = [] + for pts in remaining: + wire_endpoints = {(pts[0][0], pts[0][1]), (pts[-1][0], pts[-1][1])} + if any( + points_coincide(list(ep), list(fp)) + for ep in wire_endpoints + for fp in frontier + ): + connected_wires.append(pts) + for pt in pts: + frontier.add((pt[0], pt[1])) + changed = True + else: + still_remaining.append(pts) + remaining = still_remaining + + # Step 3: Collect all points from connected wires + connected_points = set() + for pts in connected_wires: + for pt in pts: + connected_points.add((pt[0], pt[1])) + + # Step 4: Find component pins at connected points + if not hasattr(schematic, "symbol"): + return {"success": True, "pins": []} + + locator = PinLocator() + pins = [] + seen = set() + + for symbol in schematic.symbol: + if not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if ref.startswith("_TEMPLATE"): + continue + + try: + all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) + if not all_pins: + continue + for pin_num, pin_data in all_pins.items(): + pin_loc = [pin_data["x"], pin_data["y"]] + for cp in connected_points: + if points_coincide(pin_loc, list(cp)): + key = (ref, pin_num) + if key not in seen: + seen.add(key) + pins.append({"component": ref, "pin": pin_num}) + break + except Exception as e: + logger.warning(f"Error checking pins for {ref}: {e}") + + wires_out = [ + {"start": {"x": pts[0][0], "y": pts[0][1]}, "end": {"x": pts[-1][0], "y": pts[-1][1]}} + for pts in connected_wires + ] + return {"success": True, "pins": pins, "wires": wires_out} + + except Exception as e: + logger.error(f"Error getting wire connections: {str(e)}") + import traceback + logger.error(traceback.format_exc()) + return {"success": False, "message": str(e)} + def _handle_run_erc(self, params): """Run Electrical Rules Check on a schematic via kicad-cli""" logger.info("Running ERC on schematic") diff --git a/src/tools/registry.ts b/src/tools/registry.ts index bca2cf1..d2e0e45 100644 --- a/src/tools/registry.ts +++ b/src/tools/registry.ts @@ -102,6 +102,7 @@ export const toolCategories: ToolCategory[] = [ "list_schematic_nets", "list_schematic_wires", "list_schematic_labels", + "get_wire_connections", "generate_netlist", "sync_schematic_to_board", "get_schematic_view", diff --git a/src/tools/schematic.ts b/src/tools/schematic.ts index ef8fb67..594be86 100644 --- a/src/tools/schematic.ts +++ b/src/tools/schematic.ts @@ -425,6 +425,45 @@ Note: operates on .kicad_sch files only. To modify a PCB footprint use edit_comp }, ); + // Get wire connections + server.tool( + "get_wire_connections", + "Find all component pins reachable from a schematic point via connected wires. Provide any point on a wire (start, end, or junction) to get all pins on that net.", + { + schematicPath: z.string().describe("Path to the schematic file"), + x: z.number().describe("X coordinate of the point on the wire"), + y: z.number().describe("Y coordinate of the point on the wire"), + }, + async (args: { schematicPath: string; x: number; y: number }) => { + const result = await callKicadScript("get_wire_connections", args); + if (result.success && result.pins) { + const pinList = result.pins + .map((p: any) => ` - ${p.component}/${p.pin}`) + .join("\n"); + const wireList = (result.wires ?? []) + .map((w: any) => ` - (${w.start.x},${w.start.y}) → (${w.end.x},${w.end.y})`) + .join("\n"); + return { + content: [ + { + type: "text", + text: `Pins connected at (${args.x},${args.y}):\n${pinList || " (none found)"}\n\nWire segments:\n${wireList || " (none)"}`, + }, + ], + }; + } else { + return { + content: [ + { + type: "text", + text: `Failed to get wire connections: ${result.message || "Unknown error"}`, + }, + ], + }; + } + }, + ); + // Get pin locations for a schematic component server.tool( "get_schematic_pin_locations", From ddbfc62a49fe6febc3aba4e255aadc0f50eb03e2 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 14:10:06 +0000 Subject: [PATCH 2/9] fix: address review issues in get_wire_connections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Fix critical bug: use pin_data[0]/[1] instead of pin_data["x"]/["y"] (get_all_symbol_pins returns List[float], not dict) - Use index-based wire tracking to avoid fragile float list equality - Check all polyline points (not just endpoints) during flood-fill - Add spatial index (0.05mm grid) to replace O(n²) frontier scan - Skip already-processed refs to avoid redundant calls for multi-unit symbols - Include wires_out in early return when schematic has no symbols - Add get_wire_connections schema entry to tool_schemas.py Co-Authored-By: Claude Sonnet 4.6 --- python/kicad_interface.py | 87 ++++++++++++++++++++++++---------- python/schemas/tool_schemas.py | 23 +++++++++ 2 files changed, 85 insertions(+), 25 deletions(-) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index bda174a..ba3f987 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2361,39 +2361,72 @@ class KiCADInterface: # Step 1: Find all seed wires that touch the given point (start or end) query_point = [x, y] - seed_wires = [ - pts for pts in all_wires - if points_coincide(pts[0], query_point) or points_coincide(pts[-1], query_point) - ] + seed_indices = set( + i for i, pts in enumerate(all_wires) + if any(points_coincide(pt, query_point) for pt in pts) + ) - if not seed_wires: + if not seed_indices: return { "success": False, "message": f"No wire found at ({x},{y}) within {tolerance}mm tolerance", } # Step 2: Flood-fill through connected wires - connected_wires = list(seed_wires) - frontier = set((pt[0], pt[1]) for pts in seed_wires for pt in pts) - remaining = [w for w in all_wires if w not in seed_wires] + connected_indices = set(seed_indices) + frontier = set((pt[0], pt[1]) for i in seed_indices for pt in all_wires[i]) + + # Spatial index: grid-snapped dict for O(1) proximity lookup + import math + GRID = 0.05 # mm, matches KiCAD schematic grid + grid_radius = math.ceil(tolerance / GRID) # cells to check per axis + + def _grid_key(x_coord, y_coord): + return (round(x_coord / GRID), round(y_coord / GRID)) + + # frontier_grid maps grid cell -> list of (x, y) frontier points in that cell + frontier_grid = {} + for fp in frontier: + key = _grid_key(fp[0], fp[1]) + frontier_grid.setdefault(key, []).append(fp) + + def _frontier_has_neighbour(px, py): + """Check if any frontier point is within tolerance of (px, py).""" + cx, cy = _grid_key(px, py) + for dx in range(-grid_radius, grid_radius + 1): + for dy in range(-grid_radius, grid_radius + 1): + cell = (cx + dx, cy + dy) + cell_points = frontier_grid.get(cell) + if cell_points: + for fp in cell_points: + if abs(px - fp[0]) < tolerance and abs(py - fp[1]) < tolerance: + return True + return False + + def _add_to_frontier_grid(pt): + key = _grid_key(pt[0], pt[1]) + frontier_grid.setdefault(key, []).append(pt) + + remaining_indices = [i for i in range(len(all_wires)) if i not in seed_indices] changed = True while changed: changed = False still_remaining = [] - for pts in remaining: - wire_endpoints = {(pts[0][0], pts[0][1]), (pts[-1][0], pts[-1][1])} - if any( - points_coincide(list(ep), list(fp)) - for ep in wire_endpoints - for fp in frontier - ): - connected_wires.append(pts) + for i in remaining_indices: + pts = all_wires[i] + wire_points = [(pt[0], pt[1]) for pt in pts] + if any(_frontier_has_neighbour(wp[0], wp[1]) for wp in wire_points): + connected_indices.add(i) for pt in pts: - frontier.add((pt[0], pt[1])) + p = (pt[0], pt[1]) + frontier.add(p) + _add_to_frontier_grid(p) changed = True else: - still_remaining.append(pts) - remaining = still_remaining + still_remaining.append(i) + remaining_indices = still_remaining + + connected_wires = [all_wires[i] for i in connected_indices] # Step 3: Collect all points from connected wires connected_points = set() @@ -2402,12 +2435,17 @@ class KiCADInterface: connected_points.add((pt[0], pt[1])) # Step 4: Find component pins at connected points + wires_out = [ + {"start": {"x": pts[0][0], "y": pts[0][1]}, "end": {"x": pts[-1][0], "y": pts[-1][1]}} + for pts in connected_wires + ] if not hasattr(schematic, "symbol"): - return {"success": True, "pins": []} + return {"success": True, "pins": [], "wires": wires_out} locator = PinLocator() pins = [] seen = set() + processed_refs = set() for symbol in schematic.symbol: if not hasattr(symbol.property, "Reference"): @@ -2415,13 +2453,16 @@ class KiCADInterface: ref = symbol.property.Reference.value if ref.startswith("_TEMPLATE"): continue + if ref in processed_refs: + continue + processed_refs.add(ref) try: all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) if not all_pins: continue for pin_num, pin_data in all_pins.items(): - pin_loc = [pin_data["x"], pin_data["y"]] + pin_loc = [pin_data[0], pin_data[1]] for cp in connected_points: if points_coincide(pin_loc, list(cp)): key = (ref, pin_num) @@ -2432,10 +2473,6 @@ class KiCADInterface: except Exception as e: logger.warning(f"Error checking pins for {ref}: {e}") - wires_out = [ - {"start": {"x": pts[0][0], "y": pts[0][1]}, "end": {"x": pts[-1][0], "y": pts[-1][1]}} - for pts in connected_wires - ] return {"success": True, "pins": pins, "wires": wires_out} except Exception as e: diff --git a/python/schemas/tool_schemas.py b/python/schemas/tool_schemas.py index d71a50c..a94f1c0 100644 --- a/python/schemas/tool_schemas.py +++ b/python/schemas/tool_schemas.py @@ -1471,6 +1471,29 @@ SCHEMATIC_TOOLS = [ "required": ["schematicPath", "netName"], }, }, + { + "name": "get_wire_connections", + "title": "Get Wire Connections", + "description": "Returns all wires and component pins connected to the wire at a given point, by flood-filling through touching wires.", + "inputSchema": { + "type": "object", + "properties": { + "schematicPath": { + "type": "string", + "description": "Path to schematic file" + }, + "x": { + "type": "number", + "description": "X coordinate of the point on the wire" + }, + "y": { + "type": "number", + "description": "Y coordinate of the point on the wire" + } + }, + "required": ["schematicPath", "x", "y"] + } + }, { "name": "get_schematic_pin_locations", "title": "Get Schematic Pin Locations", From b257bef89b400e8dd02fedf701a796933c7b8eea Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 14:20:56 +0000 Subject: [PATCH 3/9] fix: harden get_wire_connections against edge cases and improve performance - Add +1 safety margin to grid_radius to handle banker's rounding at cell boundaries in the spatial index - Move symbol property guards inside per-symbol try/except to prevent AttributeError from aborting all pin processing - Replace O(n) connected_points scan for pin matching with spatial index lookup (_frontier_has_neighbour), consistent with flood-fill approach - Wrap float(x)/float(y) conversion with clear user-facing error message Co-Authored-By: Claude Sonnet 4.6 --- python/kicad_interface.py | 36 ++++++++++++++++++------------------ 1 file changed, 18 insertions(+), 18 deletions(-) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index ba3f987..a9f59a0 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2335,7 +2335,10 @@ class KiCADInterface: if not all([schematic_path, x is not None, y is not None]): return {"success": False, "message": "Missing required parameters: schematicPath, x, y"} - x, y = float(x), float(y) + try: + x, y = float(x), float(y) + except (TypeError, ValueError): + return {"success": False, "message": "Parameters x and y must be numeric"} tolerance = 0.5 def points_coincide(p1, p2): @@ -2379,7 +2382,7 @@ class KiCADInterface: # Spatial index: grid-snapped dict for O(1) proximity lookup import math GRID = 0.05 # mm, matches KiCAD schematic grid - grid_radius = math.ceil(tolerance / GRID) # cells to check per axis + grid_radius = math.ceil(tolerance / GRID) + 1 # cells to check per axis (+1 safety margin for banker's rounding) def _grid_key(x_coord, y_coord): return (round(x_coord / GRID), round(y_coord / GRID)) @@ -2448,28 +2451,25 @@ class KiCADInterface: processed_refs = set() for symbol in schematic.symbol: - if not hasattr(symbol.property, "Reference"): - continue - ref = symbol.property.Reference.value - if ref.startswith("_TEMPLATE"): - continue - if ref in processed_refs: - continue - processed_refs.add(ref) - try: + if not hasattr(symbol, 'property') or not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if ref.startswith("_TEMPLATE"): + continue + if ref in processed_refs: + continue + processed_refs.add(ref) all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) if not all_pins: continue for pin_num, pin_data in all_pins.items(): pin_loc = [pin_data[0], pin_data[1]] - for cp in connected_points: - if points_coincide(pin_loc, list(cp)): - key = (ref, pin_num) - if key not in seen: - seen.add(key) - pins.append({"component": ref, "pin": pin_num}) - break + if _frontier_has_neighbour(pin_loc[0], pin_loc[1]): + key = (ref, pin_num) + if key not in seen: + seen.add(key) + pins.append({"component": ref, "pin": pin_num}) except Exception as e: logger.warning(f"Error checking pins for {ref}: {e}") From 4277a3d00017a5974897bd23c2346ce0236ccde4 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 14:55:07 +0000 Subject: [PATCH 4/9] refactor: use pre-compiled adjacency list for BFS in get_wire_connections Replace runtime spatial-index queries during BFS with a pre-compiled adjacency list for O(1) edge traversal. Also fix potential UnboundLocalError for `ref` in the pin-checking exception handler and simplify validation. Co-Authored-By: Claude Opus 4.6 --- python/kicad_interface.py | 136 ++++++++++++++++++++------------------ 1 file changed, 70 insertions(+), 66 deletions(-) diff --git a/python/kicad_interface.py b/python/kicad_interface.py index a9f59a0..25e996e 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2325,6 +2325,7 @@ class KiCADInterface: """Find all component pins reachable from a point via connected wires""" logger.info("Getting wire connections") try: + import math from pathlib import Path from commands.pin_locator import PinLocator @@ -2332,14 +2333,20 @@ class KiCADInterface: x = params.get("x") y = params.get("y") - if not all([schematic_path, x is not None, y is not None]): + if not (schematic_path and x is not None and y is not None): return {"success": False, "message": "Missing required parameters: schematicPath, x, y"} try: x, y = float(x), float(y) except (TypeError, ValueError): return {"success": False, "message": "Parameters x and y must be numeric"} + tolerance = 0.5 + GRID = 0.05 # mm, matches KiCAD schematic grid + grid_radius = math.ceil(tolerance / GRID) + 1 # +1 safety margin for banker's rounding + + def _grid_key(x_coord, y_coord): + return (round(x_coord / GRID), round(y_coord / GRID)) def points_coincide(p1, p2): return abs(p1[0] - p2[0]) < tolerance and abs(p1[1] - p2[1]) < tolerance @@ -2351,93 +2358,88 @@ class KiCADInterface: if not hasattr(schematic, "wire"): return {"success": False, "message": "Schematic has no wires"} - # Collect all wires as list of point sequences + # Collect all wires as list of endpoint tuples all_wires = [] for wire in schematic.wire: if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): pts = [] for point in wire.pts.xy: if hasattr(point, "value"): - pts.append([float(point.value[0]), float(point.value[1])]) + pts.append((float(point.value[0]), float(point.value[1]))) if len(pts) >= 2: all_wires.append(pts) - # Step 1: Find all seed wires that touch the given point (start or end) - query_point = [x, y] - seed_indices = set( - i for i, pts in enumerate(all_wires) - if any(points_coincide(pt, query_point) for pt in pts) - ) + # Build spatial index: grid_cell -> list of (wire_index, endpoint) pairs + endpoint_index = {} + for i, pts in enumerate(all_wires): + for pt in pts: + endpoint_index.setdefault(_grid_key(pt[0], pt[1]), []).append((i, pt)) + # Pre-compile adjacency list: wire_index -> set of connected wire indices. + # Two wires are adjacent when any of their endpoints coincide. + adjacency = [set() for _ in range(len(all_wires))] + for i, pts in enumerate(all_wires): + for pt in pts: + cx, cy = _grid_key(pt[0], pt[1]) + for dx in range(-grid_radius, grid_radius + 1): + for dy in range(-grid_radius, grid_radius + 1): + for j, ept in endpoint_index.get((cx + dx, cy + dy), ()): + if j != i and points_coincide(pt, ept): + adjacency[i].add(j) + + # Also build a quick lookup from grid cell to wire indices for the seed query + def _wires_near_point(px, py): + """Return indices of wires with an endpoint within tolerance of (px, py).""" + cx, cy = _grid_key(px, py) + result = set() + for dx in range(-grid_radius, grid_radius + 1): + for dy in range(-grid_radius, grid_radius + 1): + for j, ept in endpoint_index.get((cx + dx, cy + dy), ()): + if points_coincide((px, py), ept): + result.add(j) + return result + + # Step 1: Seed — find wires touching the query point + seed_indices = _wires_near_point(x, y) if not seed_indices: return { "success": False, "message": f"No wire found at ({x},{y}) within {tolerance}mm tolerance", } - # Step 2: Flood-fill through connected wires - connected_indices = set(seed_indices) - frontier = set((pt[0], pt[1]) for i in seed_indices for pt in all_wires[i]) + # Step 2: BFS flood-fill using pre-compiled adjacency (O(1) per edge) + visited = set(seed_indices) + queue = list(seed_indices) + net_points = set() + for i in seed_indices: + net_points.update(all_wires[i]) - # Spatial index: grid-snapped dict for O(1) proximity lookup - import math - GRID = 0.05 # mm, matches KiCAD schematic grid - grid_radius = math.ceil(tolerance / GRID) + 1 # cells to check per axis (+1 safety margin for banker's rounding) + while queue: + wire_idx = queue.pop() + for neighbor_idx in adjacency[wire_idx]: + if neighbor_idx not in visited: + visited.add(neighbor_idx) + queue.append(neighbor_idx) + net_points.update(all_wires[neighbor_idx]) - def _grid_key(x_coord, y_coord): - return (round(x_coord / GRID), round(y_coord / GRID)) + connected_wires = [all_wires[i] for i in visited] - # frontier_grid maps grid cell -> list of (x, y) frontier points in that cell - frontier_grid = {} - for fp in frontier: - key = _grid_key(fp[0], fp[1]) - frontier_grid.setdefault(key, []).append(fp) + # Build a grid over net_points for fast pin proximity checks + net_grid = {} + for pt in net_points: + net_grid.setdefault(_grid_key(pt[0], pt[1]), []).append(pt) - def _frontier_has_neighbour(px, py): - """Check if any frontier point is within tolerance of (px, py).""" + def _on_net(px, py): + """Return True if (px, py) is within tolerance of any net point.""" cx, cy = _grid_key(px, py) for dx in range(-grid_radius, grid_radius + 1): for dy in range(-grid_radius, grid_radius + 1): - cell = (cx + dx, cy + dy) - cell_points = frontier_grid.get(cell) - if cell_points: - for fp in cell_points: - if abs(px - fp[0]) < tolerance and abs(py - fp[1]) < tolerance: - return True + for npt in net_grid.get((cx + dx, cy + dy), ()): + if points_coincide((px, py), npt): + return True return False - def _add_to_frontier_grid(pt): - key = _grid_key(pt[0], pt[1]) - frontier_grid.setdefault(key, []).append(pt) - - remaining_indices = [i for i in range(len(all_wires)) if i not in seed_indices] - changed = True - while changed: - changed = False - still_remaining = [] - for i in remaining_indices: - pts = all_wires[i] - wire_points = [(pt[0], pt[1]) for pt in pts] - if any(_frontier_has_neighbour(wp[0], wp[1]) for wp in wire_points): - connected_indices.add(i) - for pt in pts: - p = (pt[0], pt[1]) - frontier.add(p) - _add_to_frontier_grid(p) - changed = True - else: - still_remaining.append(i) - remaining_indices = still_remaining - - connected_wires = [all_wires[i] for i in connected_indices] - - # Step 3: Collect all points from connected wires - connected_points = set() - for pts in connected_wires: - for pt in pts: - connected_points.add((pt[0], pt[1])) - - # Step 4: Find component pins at connected points + # Step 3: Output wires wires_out = [ {"start": {"x": pts[0][0], "y": pts[0][1]}, "end": {"x": pts[-1][0], "y": pts[-1][1]}} for pts in connected_wires @@ -2445,12 +2447,15 @@ class KiCADInterface: if not hasattr(schematic, "symbol"): return {"success": True, "pins": [], "wires": wires_out} + # Step 4: Find component pins that land on the net locator = PinLocator() pins = [] seen = set() processed_refs = set() + ref: str | None = None for symbol in schematic.symbol: + ref = None try: if not hasattr(symbol, 'property') or not hasattr(symbol.property, "Reference"): continue @@ -2464,14 +2469,13 @@ class KiCADInterface: if not all_pins: continue for pin_num, pin_data in all_pins.items(): - pin_loc = [pin_data[0], pin_data[1]] - if _frontier_has_neighbour(pin_loc[0], pin_loc[1]): + if _on_net(pin_data[0], pin_data[1]): key = (ref, pin_num) if key not in seen: seen.add(key) pins.append({"component": ref, "pin": pin_num}) except Exception as e: - logger.warning(f"Error checking pins for {ref}: {e}") + logger.warning(f"Error checking pins for {ref if ref is not None else ''}: {e}") return {"success": True, "pins": pins, "wires": wires_out} From f12003484601b6ea10dee2ed61fef575c1d6cdae Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 16:15:20 +0000 Subject: [PATCH 5/9] refactor: extract wire connectivity into module with KiCad-native IU matching Move wire connectivity logic from _handle_get_wire_connections into commands/wire_connectivity.py. Use KiCad's internal integer unit system (10,000 IU/mm) with exact coordinate matching instead of tolerance-based float comparison, mirroring how KiCad itself determines connectivity. Key improvements: - Exact integer matching for wire endpoints (O(1) dict lookup vs O(n) grid scan) - Junction support for T-connections - Multi-unit symbol support (removed incorrect processed_refs dedup) - Single public API: get_wire_connections() Co-Authored-By: Claude Opus 4.6 --- python/commands/wire_connectivity.py | 214 +++++++++++++++++++++++++++ python/kicad_interface.py | 137 +---------------- 2 files changed, 219 insertions(+), 132 deletions(-) create mode 100644 python/commands/wire_connectivity.py diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py new file mode 100644 index 0000000..42519eb --- /dev/null +++ b/python/commands/wire_connectivity.py @@ -0,0 +1,214 @@ +""" +Wire Connectivity Analysis for KiCad Schematics + +Traces wire networks from a point and finds connected component pins. +Uses KiCad's internal integer unit system (10,000 IU per mm) for exact +coordinate matching, mirroring KiCad's own connectivity algorithm. +""" + +import logging +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple +from commands.pin_locator import PinLocator + +logger = logging.getLogger('kicad_interface') + +_IU_PER_MM = 10000 # KiCad schematic internal units per millimeter +_QUERY_TOLERANCE_IU = 5000 # 0.5 mm in IU — for user-supplied query points + + +def _to_iu(x_mm: float, y_mm: float) -> Tuple[int, int]: + """Convert mm coordinates to KiCad internal units (integer).""" + return (round(x_mm * _IU_PER_MM), round(y_mm * _IU_PER_MM)) + + +def _parse_wires(schematic) -> List[List[Tuple[int, int]]]: + """Extract wire endpoints from a schematic object as IU tuples.""" + all_wires = [] + for wire in schematic.wire: + if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): + pts = [] + for point in wire.pts.xy: + if hasattr(point, "value"): + pts.append(_to_iu(float(point.value[0]), float(point.value[1]))) + if len(pts) >= 2: + all_wires.append(pts) + return all_wires + + +def _parse_junctions(schematic) -> List[Tuple[int, int]]: + """Extract junction points from a schematic object as IU tuples. + + Junctions may be exposed via schematic.junction (kicad-skip attribute) or + might not exist. Handle both cases gracefully. + """ + junctions = [] + if not hasattr(schematic, 'junction'): + return junctions + for junc in schematic.junction: + try: + if hasattr(junc, 'at') and hasattr(junc.at, 'value'): + junctions.append(_to_iu(float(junc.at.value[0]), float(junc.at.value[1]))) + except (IndexError, TypeError, ValueError): + continue + return junctions + + +def _build_adjacency( + all_wires: List[List[Tuple[int, int]]], + junctions: List[Tuple[int, int]], +) -> Tuple[List[Set[int]], Dict[Tuple[int, int], Set[int]]]: + """Build wire adjacency using exact IU coordinate matching. + + Returns a tuple of: + - adjacency: list of sets, one per wire, containing adjacent wire indices + - iu_to_wires: dict mapping each IU endpoint to the set of wire indices + that have an endpoint at that exact coordinate (used for seed queries) + """ + # Map each IU endpoint to all wire indices that touch it + iu_to_wires: Dict[Tuple[int, int], Set[int]] = {} + for i, pts in enumerate(all_wires): + for pt in pts: + iu_to_wires.setdefault(pt, set()).add(i) + + # Wires that share an IU endpoint are adjacent + adjacency: List[Set[int]] = [set() for _ in range(len(all_wires))] + for wire_set in iu_to_wires.values(): + wire_list = list(wire_set) + for a in wire_list: + for b in wire_list: + if a != b: + adjacency[a].add(b) + + # Junctions: connect all wires that have an endpoint at the junction IU point + for junc_iu in junctions: + wire_set = iu_to_wires.get(junc_iu, set()) + wire_list = list(wire_set) + for a in wire_list: + for b in wire_list: + if a != b: + adjacency[a].add(b) + + return adjacency, iu_to_wires + + +def _find_connected_wires( + x_mm: float, + y_mm: float, + all_wires: List[List[Tuple[int, int]]], + iu_to_wires: Dict[Tuple[int, int], Set[int]], + adjacency: List[Set[int]], +) -> Tuple: + """BFS from query point. Returns (visited wire indices, net IU points) or (None, None). + + Uses _QUERY_TOLERANCE_IU for the seed step because user-supplied coordinates + may be imprecise. Wire-to-wire matching inside _build_adjacency is exact. + """ + query_iu = _to_iu(x_mm, y_mm) + + # Find seed wires: any wire whose endpoint is within _QUERY_TOLERANCE_IU of the query + seed_indices: Set[int] = set() + for iu_pt, wire_indices in iu_to_wires.items(): + if (abs(iu_pt[0] - query_iu[0]) <= _QUERY_TOLERANCE_IU and + abs(iu_pt[1] - query_iu[1]) <= _QUERY_TOLERANCE_IU): + seed_indices.update(wire_indices) + + if not seed_indices: + return (None, None) + + # BFS flood-fill using pre-compiled adjacency + visited: Set[int] = set(seed_indices) + queue = list(seed_indices) + net_points: Set[Tuple[int, int]] = set() + for i in seed_indices: + net_points.update(all_wires[i]) + + while queue: + wire_idx = queue.pop() + for neighbor_idx in adjacency[wire_idx]: + if neighbor_idx not in visited: + visited.add(neighbor_idx) + queue.append(neighbor_idx) + net_points.update(all_wires[neighbor_idx]) + + return (visited, net_points) + + +def _find_pins_on_net( + net_points: Set[Tuple[int, int]], + schematic_path, + schematic, +) -> List[Dict]: + """Find component pins that land on net points. + + Uses exact IU matching with a ±_PIN_TOLERANCE_IU neighbourhood to guard + against floating-point round-trip differences between wire and pin coordinates. + + Returns a list of {"component": ref, "pin": pin_num} dicts. + """ + + def _on_net(px_mm: float, py_mm: float) -> bool: + pin_iu = _to_iu(px_mm, py_mm) + if pin_iu in net_points: + return True + x, y = pin_iu + return ((x+1, y) in net_points or (x-1, y) in net_points or + (x, y+1) in net_points or (x, y-1) in net_points) + + locator = PinLocator() + pins = [] + seen: Set[Tuple] = set() + + ref = None + for symbol in schematic.symbol: + try: + if not hasattr(symbol, 'property') or not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if ref.startswith("_TEMPLATE"): + continue + all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) + if not all_pins: + continue + for pin_num, pin_data in all_pins.items(): + if _on_net(pin_data[0], pin_data[1]): + key = (ref, pin_num) + if key not in seen: + seen.add(key) + pins.append({"component": ref, "pin": pin_num}) + except Exception as e: + logger.warning(f"Error checking pins for {ref if ref is not None else ''}: {e}") + + return pins + + +def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: float) -> Optional[Dict]: + """Find all component pins reachable from a point via connected wires. + + Returns dict with keys: + - "pins": list of {"component": str, "pin": str} + - "wires": list of {"start": {"x", "y"}, "end": {"x", "y"}} in mm + Or None if no wire found at the query point. + """ + all_wires = _parse_wires(schematic) + if not all_wires: + return {"pins": [], "wires": []} + + junctions = _parse_junctions(schematic) + adjacency, iu_to_wires = _build_adjacency(all_wires, junctions) + + visited, net_points = _find_connected_wires(x_mm, y_mm, all_wires, iu_to_wires, adjacency) + if visited is None: + return None + + wires_out = [ + {"start": {"x": all_wires[i][0][0] / _IU_PER_MM, "y": all_wires[i][0][1] / _IU_PER_MM}, + "end": {"x": all_wires[i][-1][0] / _IU_PER_MM, "y": all_wires[i][-1][1] / _IU_PER_MM}} + for i in visited + ] + + if not hasattr(schematic, "symbol"): + return {"pins": [], "wires": wires_out} + + pins = _find_pins_on_net(net_points, schematic_path, schematic) + return {"pins": pins, "wires": wires_out} diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 25e996e..937f4f3 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2325,9 +2325,7 @@ class KiCADInterface: """Find all component pins reachable from a point via connected wires""" logger.info("Getting wire connections") try: - import math - from pathlib import Path - from commands.pin_locator import PinLocator + from commands.wire_connectivity import get_wire_connections schematic_path = params.get("schematicPath") x = params.get("x") @@ -2341,16 +2339,6 @@ class KiCADInterface: except (TypeError, ValueError): return {"success": False, "message": "Parameters x and y must be numeric"} - tolerance = 0.5 - GRID = 0.05 # mm, matches KiCAD schematic grid - grid_radius = math.ceil(tolerance / GRID) + 1 # +1 safety margin for banker's rounding - - def _grid_key(x_coord, y_coord): - return (round(x_coord / GRID), round(y_coord / GRID)) - - def points_coincide(p1, p2): - return abs(p1[0] - p2[0]) < tolerance and abs(p1[1] - p2[1]) < tolerance - schematic = SchematicManager.load_schematic(schematic_path) if not schematic: return {"success": False, "message": "Failed to load schematic"} @@ -2358,126 +2346,11 @@ class KiCADInterface: if not hasattr(schematic, "wire"): return {"success": False, "message": "Schematic has no wires"} - # Collect all wires as list of endpoint tuples - all_wires = [] - for wire in schematic.wire: - if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): - pts = [] - for point in wire.pts.xy: - if hasattr(point, "value"): - pts.append((float(point.value[0]), float(point.value[1]))) - if len(pts) >= 2: - all_wires.append(pts) + result = get_wire_connections(schematic, schematic_path, x, y) + if result is None: + return {"success": False, "message": f"No wire found at ({x},{y}) within tolerance"} - # Build spatial index: grid_cell -> list of (wire_index, endpoint) pairs - endpoint_index = {} - for i, pts in enumerate(all_wires): - for pt in pts: - endpoint_index.setdefault(_grid_key(pt[0], pt[1]), []).append((i, pt)) - - # Pre-compile adjacency list: wire_index -> set of connected wire indices. - # Two wires are adjacent when any of their endpoints coincide. - adjacency = [set() for _ in range(len(all_wires))] - for i, pts in enumerate(all_wires): - for pt in pts: - cx, cy = _grid_key(pt[0], pt[1]) - for dx in range(-grid_radius, grid_radius + 1): - for dy in range(-grid_radius, grid_radius + 1): - for j, ept in endpoint_index.get((cx + dx, cy + dy), ()): - if j != i and points_coincide(pt, ept): - adjacency[i].add(j) - - # Also build a quick lookup from grid cell to wire indices for the seed query - def _wires_near_point(px, py): - """Return indices of wires with an endpoint within tolerance of (px, py).""" - cx, cy = _grid_key(px, py) - result = set() - for dx in range(-grid_radius, grid_radius + 1): - for dy in range(-grid_radius, grid_radius + 1): - for j, ept in endpoint_index.get((cx + dx, cy + dy), ()): - if points_coincide((px, py), ept): - result.add(j) - return result - - # Step 1: Seed — find wires touching the query point - seed_indices = _wires_near_point(x, y) - if not seed_indices: - return { - "success": False, - "message": f"No wire found at ({x},{y}) within {tolerance}mm tolerance", - } - - # Step 2: BFS flood-fill using pre-compiled adjacency (O(1) per edge) - visited = set(seed_indices) - queue = list(seed_indices) - net_points = set() - for i in seed_indices: - net_points.update(all_wires[i]) - - while queue: - wire_idx = queue.pop() - for neighbor_idx in adjacency[wire_idx]: - if neighbor_idx not in visited: - visited.add(neighbor_idx) - queue.append(neighbor_idx) - net_points.update(all_wires[neighbor_idx]) - - connected_wires = [all_wires[i] for i in visited] - - # Build a grid over net_points for fast pin proximity checks - net_grid = {} - for pt in net_points: - net_grid.setdefault(_grid_key(pt[0], pt[1]), []).append(pt) - - def _on_net(px, py): - """Return True if (px, py) is within tolerance of any net point.""" - cx, cy = _grid_key(px, py) - for dx in range(-grid_radius, grid_radius + 1): - for dy in range(-grid_radius, grid_radius + 1): - for npt in net_grid.get((cx + dx, cy + dy), ()): - if points_coincide((px, py), npt): - return True - return False - - # Step 3: Output wires - wires_out = [ - {"start": {"x": pts[0][0], "y": pts[0][1]}, "end": {"x": pts[-1][0], "y": pts[-1][1]}} - for pts in connected_wires - ] - if not hasattr(schematic, "symbol"): - return {"success": True, "pins": [], "wires": wires_out} - - # Step 4: Find component pins that land on the net - locator = PinLocator() - pins = [] - seen = set() - processed_refs = set() - - ref: str | None = None - for symbol in schematic.symbol: - ref = None - try: - if not hasattr(symbol, 'property') or not hasattr(symbol.property, "Reference"): - continue - ref = symbol.property.Reference.value - if ref.startswith("_TEMPLATE"): - continue - if ref in processed_refs: - continue - processed_refs.add(ref) - all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) - if not all_pins: - continue - for pin_num, pin_data in all_pins.items(): - if _on_net(pin_data[0], pin_data[1]): - key = (ref, pin_num) - if key not in seen: - seen.add(key) - pins.append({"component": ref, "pin": pin_num}) - except Exception as e: - logger.warning(f"Error checking pins for {ref if ref is not None else ''}: {e}") - - return {"success": True, "pins": pins, "wires": wires_out} + return {"success": True, **result} except Exception as e: logger.error(f"Error getting wire connections: {str(e)}") From de0eca2ed7139cb830fa38849508d5d3fe2eacf0 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 16:20:11 +0000 Subject: [PATCH 6/9] refactor: remove redundant junction processing from wire connectivity Shared-endpoint adjacency already connects all wires meeting at the same point, making explicit junction handling a no-op duplicate. Co-Authored-By: Claude Opus 4.6 --- python/commands/wire_connectivity.py | 34 ++++------------------------ 1 file changed, 4 insertions(+), 30 deletions(-) diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py index 42519eb..5942c6a 100644 --- a/python/commands/wire_connectivity.py +++ b/python/commands/wire_connectivity.py @@ -36,30 +36,14 @@ def _parse_wires(schematic) -> List[List[Tuple[int, int]]]: return all_wires -def _parse_junctions(schematic) -> List[Tuple[int, int]]: - """Extract junction points from a schematic object as IU tuples. - - Junctions may be exposed via schematic.junction (kicad-skip attribute) or - might not exist. Handle both cases gracefully. - """ - junctions = [] - if not hasattr(schematic, 'junction'): - return junctions - for junc in schematic.junction: - try: - if hasattr(junc, 'at') and hasattr(junc.at, 'value'): - junctions.append(_to_iu(float(junc.at.value[0]), float(junc.at.value[1]))) - except (IndexError, TypeError, ValueError): - continue - return junctions - - def _build_adjacency( all_wires: List[List[Tuple[int, int]]], - junctions: List[Tuple[int, int]], ) -> Tuple[List[Set[int]], Dict[Tuple[int, int], Set[int]]]: """Build wire adjacency using exact IU coordinate matching. + Wires that share an endpoint are adjacent — this naturally handles + junctions since all wires meeting at the same point get connected. + Returns a tuple of: - adjacency: list of sets, one per wire, containing adjacent wire indices - iu_to_wires: dict mapping each IU endpoint to the set of wire indices @@ -80,15 +64,6 @@ def _build_adjacency( if a != b: adjacency[a].add(b) - # Junctions: connect all wires that have an endpoint at the junction IU point - for junc_iu in junctions: - wire_set = iu_to_wires.get(junc_iu, set()) - wire_list = list(wire_set) - for a in wire_list: - for b in wire_list: - if a != b: - adjacency[a].add(b) - return adjacency, iu_to_wires @@ -194,8 +169,7 @@ def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: floa if not all_wires: return {"pins": [], "wires": []} - junctions = _parse_junctions(schematic) - adjacency, iu_to_wires = _build_adjacency(all_wires, junctions) + adjacency, iu_to_wires = _build_adjacency(all_wires) visited, net_points = _find_connected_wires(x_mm, y_mm, all_wires, iu_to_wires, adjacency) if visited is None: From 8414784b788e2af465a7121f3e088f723523abe6 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 16:43:06 +0000 Subject: [PATCH 7/9] docs: clarify get_wire_connections requires endpoint coordinates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update tool description and Python docstring to make clear that the query point must be at a wire endpoint or junction — midpoints of wire segments are not matched. Co-Authored-By: Claude Sonnet 4.6 --- python/commands/wire_connectivity.py | 6 +++++- src/tools/schematic.ts | 6 +++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py index 5942c6a..b7a1c41 100644 --- a/python/commands/wire_connectivity.py +++ b/python/commands/wire_connectivity.py @@ -160,10 +160,14 @@ def _find_pins_on_net( def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: float) -> Optional[Dict]: """Find all component pins reachable from a point via connected wires. + The query point (x_mm, y_mm) must be within _QUERY_TOLERANCE_IU (0.5 mm) of + a wire endpoint or junction. Interior (mid-segment) points are not matched — + use wire endpoint coordinates obtained from the schematic data. + Returns dict with keys: - "pins": list of {"component": str, "pin": str} - "wires": list of {"start": {"x", "y"}, "end": {"x", "y"}} in mm - Or None if no wire found at the query point. + Or None if no wire endpoint found within tolerance of the query point. """ all_wires = _parse_wires(schematic) if not all_wires: diff --git a/src/tools/schematic.ts b/src/tools/schematic.ts index 594be86..55aff42 100644 --- a/src/tools/schematic.ts +++ b/src/tools/schematic.ts @@ -428,11 +428,11 @@ Note: operates on .kicad_sch files only. To modify a PCB footprint use edit_comp // Get wire connections server.tool( "get_wire_connections", - "Find all component pins reachable from a schematic point via connected wires. Provide any point on a wire (start, end, or junction) to get all pins on that net.", + "Find all component pins reachable from a schematic point via connected wires. The query point must be at a wire endpoint or junction — midpoints of wire segments are not matched. Use get_schematic_pin_locations or list_schematic_wires to obtain exact endpoint coordinates first.", { schematicPath: z.string().describe("Path to the schematic file"), - x: z.number().describe("X coordinate of the point on the wire"), - y: z.number().describe("Y coordinate of the point on the wire"), + x: z.number().describe("X coordinate of a wire endpoint or junction"), + y: z.number().describe("Y coordinate of a wire endpoint or junction"), }, async (args: { schematicPath: string; x: number; y: number }) => { const result = await callKicadScript("get_wire_connections", args); From 018f4a278daf56b1e914a1dbd1e7bf3af40c2569 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sat, 14 Mar 2026 17:16:57 +0000 Subject: [PATCH 8/9] refactor: switch get_wire_connections to exact IU matching throughout Remove the 0.5mm query tolerance in favour of exact integer-unit matching on all coordinate lookups (seed, label bridging, pin matching), mirroring KiCad's own connectivity algorithm. Callers must supply exact wire endpoint coordinates (e.g. from list_schematic_wires). Co-Authored-By: Claude Sonnet 4.6 --- python/commands/wire_connectivity.py | 114 +++++++++++++++++++++------ src/tools/schematic.ts | 2 +- 2 files changed, 90 insertions(+), 26 deletions(-) diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py index b7a1c41..9662a98 100644 --- a/python/commands/wire_connectivity.py +++ b/python/commands/wire_connectivity.py @@ -14,7 +14,6 @@ from commands.pin_locator import PinLocator logger = logging.getLogger('kicad_interface') _IU_PER_MM = 10000 # KiCad schematic internal units per millimeter -_QUERY_TOLERANCE_IU = 5000 # 0.5 mm in IU — for user-supplied query points def _to_iu(x_mm: float, y_mm: float) -> Tuple[int, int]: @@ -67,29 +66,78 @@ def _build_adjacency( return adjacency, iu_to_wires +def _parse_virtual_connections(schematic, schematic_path): + """Return virtual connectivity from net labels and power symbols. + + Returns a tuple of: + - point_to_label: Dict[Tuple[int,int], str] — IU position → label name + - label_to_points: Dict[str, List[Tuple[int,int]]] — label name → list of IU positions + """ + point_to_label: Dict[Tuple[int, int], str] = {} + label_to_points: Dict[str, List[Tuple[int, int]]] = {} + + if hasattr(schematic, "label"): + for label in schematic.label: + try: + if not hasattr(label, "value"): + continue + name = label.value + if not hasattr(label, "at") or not hasattr(label.at, "value"): + continue + coords = label.at.value + pt = _to_iu(float(coords[0]), float(coords[1])) + point_to_label[pt] = name + label_to_points.setdefault(name, []).append(pt) + except Exception as e: + logger.warning(f"Error parsing net label: {e}") + + if hasattr(schematic, "symbol"): + locator = PinLocator() + for symbol in schematic.symbol: + try: + if not hasattr(symbol, "property") or not hasattr(symbol.property, "Reference"): + continue + ref = symbol.property.Reference.value + if not ref.startswith("#PWR"): + continue + if ref.startswith("_TEMPLATE"): + continue + if not hasattr(symbol.property, "Value"): + continue + name = symbol.property.Value.value + all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref) + if not all_pins or "1" not in all_pins: + continue + pin_data = all_pins["1"] + pt = _to_iu(float(pin_data[0]), float(pin_data[1])) + point_to_label[pt] = name + label_to_points.setdefault(name, []).append(pt) + except Exception as e: + logger.warning(f"Error parsing power symbol: {e}") + + return point_to_label, label_to_points + + def _find_connected_wires( x_mm: float, y_mm: float, all_wires: List[List[Tuple[int, int]]], iu_to_wires: Dict[Tuple[int, int], Set[int]], adjacency: List[Set[int]], + point_to_label: Optional[Dict[Tuple[int, int], str]] = None, + label_to_points: Optional[Dict[str, List[Tuple[int, int]]]] = None, ) -> Tuple: """BFS from query point. Returns (visited wire indices, net IU points) or (None, None). - Uses _QUERY_TOLERANCE_IU for the seed step because user-supplied coordinates - may be imprecise. Wire-to-wire matching inside _build_adjacency is exact. + Requires query point (x_mm, y_mm) to be exactly on a wire endpoint (exact IU match). """ query_iu = _to_iu(x_mm, y_mm) - # Find seed wires: any wire whose endpoint is within _QUERY_TOLERANCE_IU of the query - seed_indices: Set[int] = set() - for iu_pt, wire_indices in iu_to_wires.items(): - if (abs(iu_pt[0] - query_iu[0]) <= _QUERY_TOLERANCE_IU and - abs(iu_pt[1] - query_iu[1]) <= _QUERY_TOLERANCE_IU): - seed_indices.update(wire_indices) - - if not seed_indices: + # Find seed wires: exact IU match on the query endpoint + seed_set = iu_to_wires.get(query_iu) + if not seed_set: return (None, None) + seed_indices: Set[int] = set(seed_set) # BFS flood-fill using pre-compiled adjacency visited: Set[int] = set(seed_indices) @@ -98,6 +146,7 @@ def _find_connected_wires( for i in seed_indices: net_points.update(all_wires[i]) + seen_labels: Set[str] = set() while queue: wire_idx = queue.pop() for neighbor_idx in adjacency[wire_idx]: @@ -106,6 +155,20 @@ def _find_connected_wires( queue.append(neighbor_idx) net_points.update(all_wires[neighbor_idx]) + if point_to_label and label_to_points: + for pt in all_wires[wire_idx]: + label_name = point_to_label.get(pt) + if label_name and label_name not in seen_labels: + seen_labels.add(label_name) + for other_pt in label_to_points.get(label_name, []): + if other_pt == pt: + continue + for idx in iu_to_wires.get(other_pt, set()): + if idx not in visited: + visited.add(idx) + queue.append(idx) + net_points.update(all_wires[idx]) + return (visited, net_points) @@ -114,21 +177,13 @@ def _find_pins_on_net( schematic_path, schematic, ) -> List[Dict]: - """Find component pins that land on net points. - - Uses exact IU matching with a ±_PIN_TOLERANCE_IU neighbourhood to guard - against floating-point round-trip differences between wire and pin coordinates. + """Find component pins that land on net points using exact IU matching. Returns a list of {"component": ref, "pin": pin_num} dicts. """ def _on_net(px_mm: float, py_mm: float) -> bool: - pin_iu = _to_iu(px_mm, py_mm) - if pin_iu in net_points: - return True - x, y = pin_iu - return ((x+1, y) in net_points or (x-1, y) in net_points or - (x, y+1) in net_points or (x, y-1) in net_points) + return _to_iu(px_mm, py_mm) in net_points locator = PinLocator() pins = [] @@ -158,12 +213,15 @@ def _find_pins_on_net( def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: float) -> Optional[Dict]: - """Find all component pins reachable from a point via connected wires. + """Find all component pins reachable from a point via connected wires, net labels, and power symbols. - The query point (x_mm, y_mm) must be within _QUERY_TOLERANCE_IU (0.5 mm) of - a wire endpoint or junction. Interior (mid-segment) points are not matched — + The query point (x_mm, y_mm) must be exactly on a wire endpoint or junction (exact IU match). + Interior (mid-segment) points are not matched — use wire endpoint coordinates obtained from the schematic data. + Net labels and power symbols are traversed: wires on the same named net are + treated as connected even when they are not geometrically adjacent. + Returns dict with keys: - "pins": list of {"component": str, "pin": str} - "wires": list of {"start": {"x", "y"}, "end": {"x", "y"}} in mm @@ -175,7 +233,13 @@ def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: floa adjacency, iu_to_wires = _build_adjacency(all_wires) - visited, net_points = _find_connected_wires(x_mm, y_mm, all_wires, iu_to_wires, adjacency) + point_to_label, label_to_points = _parse_virtual_connections(schematic, schematic_path) + + visited, net_points = _find_connected_wires( + x_mm, y_mm, all_wires, iu_to_wires, adjacency, + point_to_label=point_to_label, + label_to_points=label_to_points, + ) if visited is None: return None diff --git a/src/tools/schematic.ts b/src/tools/schematic.ts index 55aff42..2dc6604 100644 --- a/src/tools/schematic.ts +++ b/src/tools/schematic.ts @@ -428,7 +428,7 @@ Note: operates on .kicad_sch files only. To modify a PCB footprint use edit_comp // Get wire connections server.tool( "get_wire_connections", - "Find all component pins reachable from a schematic point via connected wires. The query point must be at a wire endpoint or junction — midpoints of wire segments are not matched. Use get_schematic_pin_locations or list_schematic_wires to obtain exact endpoint coordinates first.", + "Find all component pins reachable from a schematic point via connected wires, net labels, and power symbols. The query point must be at a wire endpoint or junction — midpoints of wire segments are not matched. Use get_schematic_pin_locations or list_schematic_wires to obtain exact endpoint coordinates first.", { schematicPath: z.string().describe("Path to the schematic file"), x: z.number().describe("X coordinate of a wire endpoint or junction"), From 9f8943791885a5a0ae3ec3e735e20e9c827d3454 Mon Sep 17 00:00:00 2001 From: Eugene Mikhantyev Date: Sun, 15 Mar 2026 23:06:08 +0000 Subject: [PATCH 9/9] chore: apply Black formatting and add tests for get_wire_connections - Apply Black formatting to wire_connectivity.py, kicad_interface.py, and tool_schemas.py (changed files only) - Add python/tests/conftest.py with pcbnew/skip C-extension stubs - Add python/tests/test_wire_connectivity.py with 29 unit tests covering schema validation, handler dispatch, parameter validation, and core logic (_to_iu, _parse_wires, _build_adjacency, _find_connected_wires, get_wire_connections) Co-Authored-By: Claude Sonnet 4.6 --- python/commands/wire_connectivity.py | 40 ++- python/kicad_interface.py | 16 +- python/schemas/tool_schemas.py | 12 +- python/tests/test_wire_connectivity.py | 338 +++++++++++++++++++++++++ 4 files changed, 388 insertions(+), 18 deletions(-) create mode 100644 python/tests/test_wire_connectivity.py diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py index 9662a98..dfba1e8 100644 --- a/python/commands/wire_connectivity.py +++ b/python/commands/wire_connectivity.py @@ -11,7 +11,7 @@ from pathlib import Path from typing import Dict, List, Optional, Set, Tuple from commands.pin_locator import PinLocator -logger = logging.getLogger('kicad_interface') +logger = logging.getLogger("kicad_interface") _IU_PER_MM = 10000 # KiCad schematic internal units per millimeter @@ -95,7 +95,9 @@ def _parse_virtual_connections(schematic, schematic_path): locator = PinLocator() for symbol in schematic.symbol: try: - if not hasattr(symbol, "property") or not hasattr(symbol.property, "Reference"): + if not hasattr(symbol, "property") or not hasattr( + symbol.property, "Reference" + ): continue ref = symbol.property.Reference.value if not ref.startswith("#PWR"): @@ -192,7 +194,9 @@ def _find_pins_on_net( ref = None for symbol in schematic.symbol: try: - if not hasattr(symbol, 'property') or not hasattr(symbol.property, "Reference"): + if not hasattr(symbol, "property") or not hasattr( + symbol.property, "Reference" + ): continue ref = symbol.property.Reference.value if ref.startswith("_TEMPLATE"): @@ -207,12 +211,16 @@ def _find_pins_on_net( seen.add(key) pins.append({"component": ref, "pin": pin_num}) except Exception as e: - logger.warning(f"Error checking pins for {ref if ref is not None else ''}: {e}") + logger.warning( + f"Error checking pins for {ref if ref is not None else ''}: {e}" + ) return pins -def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: float) -> Optional[Dict]: +def get_wire_connections( + schematic, schematic_path: str, x_mm: float, y_mm: float +) -> Optional[Dict]: """Find all component pins reachable from a point via connected wires, net labels, and power symbols. The query point (x_mm, y_mm) must be exactly on a wire endpoint or junction (exact IU match). @@ -233,10 +241,16 @@ def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: floa adjacency, iu_to_wires = _build_adjacency(all_wires) - point_to_label, label_to_points = _parse_virtual_connections(schematic, schematic_path) + point_to_label, label_to_points = _parse_virtual_connections( + schematic, schematic_path + ) visited, net_points = _find_connected_wires( - x_mm, y_mm, all_wires, iu_to_wires, adjacency, + x_mm, + y_mm, + all_wires, + iu_to_wires, + adjacency, point_to_label=point_to_label, label_to_points=label_to_points, ) @@ -244,8 +258,16 @@ def get_wire_connections(schematic, schematic_path: str, x_mm: float, y_mm: floa return None wires_out = [ - {"start": {"x": all_wires[i][0][0] / _IU_PER_MM, "y": all_wires[i][0][1] / _IU_PER_MM}, - "end": {"x": all_wires[i][-1][0] / _IU_PER_MM, "y": all_wires[i][-1][1] / _IU_PER_MM}} + { + "start": { + "x": all_wires[i][0][0] / _IU_PER_MM, + "y": all_wires[i][0][1] / _IU_PER_MM, + }, + "end": { + "x": all_wires[i][-1][0] / _IU_PER_MM, + "y": all_wires[i][-1][1] / _IU_PER_MM, + }, + } for i in visited ] diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 937f4f3..3abb116 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2332,12 +2332,18 @@ class KiCADInterface: y = params.get("y") if not (schematic_path and x is not None and y is not None): - return {"success": False, "message": "Missing required parameters: schematicPath, x, y"} + return { + "success": False, + "message": "Missing required parameters: schematicPath, x, y", + } try: x, y = float(x), float(y) except (TypeError, ValueError): - return {"success": False, "message": "Parameters x and y must be numeric"} + return { + "success": False, + "message": "Parameters x and y must be numeric", + } schematic = SchematicManager.load_schematic(schematic_path) if not schematic: @@ -2348,13 +2354,17 @@ class KiCADInterface: result = get_wire_connections(schematic, schematic_path, x, y) if result is None: - return {"success": False, "message": f"No wire found at ({x},{y}) within tolerance"} + return { + "success": False, + "message": f"No wire found at ({x},{y}) within tolerance", + } return {"success": True, **result} except Exception as e: logger.error(f"Error getting wire connections: {str(e)}") import traceback + logger.error(traceback.format_exc()) return {"success": False, "message": str(e)} diff --git a/python/schemas/tool_schemas.py b/python/schemas/tool_schemas.py index a94f1c0..49b91ae 100644 --- a/python/schemas/tool_schemas.py +++ b/python/schemas/tool_schemas.py @@ -1480,19 +1480,19 @@ SCHEMATIC_TOOLS = [ "properties": { "schematicPath": { "type": "string", - "description": "Path to schematic file" + "description": "Path to schematic file", }, "x": { "type": "number", - "description": "X coordinate of the point on the wire" + "description": "X coordinate of the point on the wire", }, "y": { "type": "number", - "description": "Y coordinate of the point on the wire" - } + "description": "Y coordinate of the point on the wire", + }, }, - "required": ["schematicPath", "x", "y"] - } + "required": ["schematicPath", "x", "y"], + }, }, { "name": "get_schematic_pin_locations", diff --git a/python/tests/test_wire_connectivity.py b/python/tests/test_wire_connectivity.py new file mode 100644 index 0000000..e8c7028 --- /dev/null +++ b/python/tests/test_wire_connectivity.py @@ -0,0 +1,338 @@ +""" +Tests for the wire_connectivity module and the get_wire_connections handler. + +Covers: + - Schema shape (TestSchema) + - Handler dispatch registration (TestHandlerDispatch) + - Parameter validation in the handler (TestHandlerParamValidation) + - Core logic: _to_iu, _parse_wires, _build_adjacency, _find_connected_wires, + get_wire_connections (TestCoreLogic) +""" + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +# Ensure the python package root is importable +sys.path.insert(0, str(Path(__file__).parent.parent)) + +# --------------------------------------------------------------------------- +# Module under test +# --------------------------------------------------------------------------- +from commands.wire_connectivity import ( + _build_adjacency, + _find_connected_wires, + _parse_wires, + _to_iu, + get_wire_connections, +) + +# --------------------------------------------------------------------------- +# Helpers to build minimal mock schematic objects +# --------------------------------------------------------------------------- + + +def _make_point(x: float, y: float) -> MagicMock: + pt = MagicMock() + pt.value = [x, y] + return pt + + +def _make_wire(x1: float, y1: float, x2: float, y2: float) -> MagicMock: + wire = MagicMock() + wire.pts = MagicMock() + wire.pts.xy = [_make_point(x1, y1), _make_point(x2, y2)] + return wire + + +def _make_schematic(*wires) -> MagicMock: + sch = MagicMock() + sch.wire = list(wires) + # No net labels, no symbols by default + del sch.label # make hasattr(..., "label") return False + del sch.symbol # make hasattr(..., "symbol") return False + return sch + + +# --------------------------------------------------------------------------- +# TestSchema +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestSchema: + """Verify the get_wire_connections tool schema is present and well-formed.""" + + def test_schema_registered(self): + from schemas.tool_schemas import TOOL_SCHEMAS + + assert "get_wire_connections" in TOOL_SCHEMAS + + def test_schema_required_fields(self): + from schemas.tool_schemas import TOOL_SCHEMAS + + schema = TOOL_SCHEMAS["get_wire_connections"] + required = schema["inputSchema"]["required"] + assert "schematicPath" in required + assert "x" in required + assert "y" in required + + def test_schema_has_title_and_description(self): + from schemas.tool_schemas import TOOL_SCHEMAS + + schema = TOOL_SCHEMAS["get_wire_connections"] + assert schema.get("title") + assert schema.get("description") + + +# --------------------------------------------------------------------------- +# TestHandlerDispatch +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestHandlerDispatch: + """Verify the handler is wired into KiCadInterface.command_routes.""" + + def test_get_wire_connections_in_routes(self): + # Import lazily to avoid heavy side-effects at collection time + with patch("kicad_interface.USE_IPC_BACKEND", False): + from kicad_interface import KiCADInterface + + iface = KiCADInterface.__new__(KiCADInterface) + iface.board = None + iface.project_filename = None + iface.use_ipc = False + iface.ipc_backend = MagicMock() + iface.ipc_board_api = None + iface.footprint_library = MagicMock() + iface.project_commands = MagicMock() + iface.board_commands = MagicMock() + iface.component_commands = MagicMock() + iface.routing_commands = MagicMock() + + # Build routes only (avoid full __init__ side-effects) + # The routes dict is built in __init__; we call it directly. + iface.__init__() + + assert "get_wire_connections" in iface.command_routes + assert callable(iface.command_routes["get_wire_connections"]) + + +# --------------------------------------------------------------------------- +# TestHandlerParamValidation +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestHandlerParamValidation: + """Handler returns error responses for bad or missing parameters.""" + + def _make_handler(self): + """Return a bound _handle_get_wire_connections without full init.""" + with patch("kicad_interface.USE_IPC_BACKEND", False): + from kicad_interface import KiCADInterface + + iface = KiCADInterface.__new__(KiCADInterface) + return iface._handle_get_wire_connections + + def test_missing_schematic_path(self): + handler = self._make_handler() + result = handler({"x": 1.0, "y": 2.0}) + assert result["success"] is False + assert "schematicPath" in result["message"] or "Missing" in result["message"] + + def test_missing_x(self): + handler = self._make_handler() + result = handler({"schematicPath": "/tmp/test.kicad_sch", "y": 2.0}) + assert result["success"] is False + + def test_missing_y(self): + handler = self._make_handler() + result = handler({"schematicPath": "/tmp/test.kicad_sch", "x": 1.0}) + assert result["success"] is False + + def test_non_numeric_x(self): + handler = self._make_handler() + result = handler({"schematicPath": "/tmp/test.kicad_sch", "x": "bad", "y": 2.0}) + assert result["success"] is False + assert "numeric" in result["message"].lower() or "x" in result["message"] + + def test_non_numeric_y(self): + handler = self._make_handler() + result = handler({"schematicPath": "/tmp/test.kicad_sch", "x": 1.0, "y": "bad"}) + assert result["success"] is False + + +# --------------------------------------------------------------------------- +# TestCoreLogic +# --------------------------------------------------------------------------- + +_IU = 10_000 # IU per mm + + +@pytest.mark.unit +class TestCoreLogic: + """Unit tests for the pure-logic functions in wire_connectivity.""" + + # --- _to_iu --- + + def test_to_iu_integer_mm(self): + assert _to_iu(1.0, 2.0) == (10_000, 20_000) + + def test_to_iu_fractional_mm(self): + assert _to_iu(0.5, 0.25) == (5_000, 2_500) + + def test_to_iu_zero(self): + assert _to_iu(0.0, 0.0) == (0, 0) + + def test_to_iu_negative(self): + assert _to_iu(-1.0, -2.0) == (-10_000, -20_000) + + # --- _parse_wires --- + + def test_parse_wires_single_wire(self): + sch = _make_schematic(_make_wire(0.0, 0.0, 1.0, 0.0)) + result = _parse_wires(sch) + assert len(result) == 1 + assert result[0] == [(0, 0), (10_000, 0)] + + def test_parse_wires_empty_schematic(self): + sch = MagicMock() + sch.wire = [] + assert _parse_wires(sch) == [] + + def test_parse_wires_multiple_wires(self): + sch = _make_schematic( + _make_wire(0.0, 0.0, 1.0, 0.0), + _make_wire(1.0, 0.0, 2.0, 0.0), + ) + assert len(_parse_wires(sch)) == 2 + + def test_parse_wires_skips_wire_without_pts(self): + bad_wire = MagicMock(spec=[]) # no `pts` attribute + sch = MagicMock() + sch.wire = [bad_wire] + assert _parse_wires(sch) == [] + + # --- _build_adjacency --- + + def test_build_adjacency_two_connected_wires(self): + # wire0: (0,0)-(1,0), wire1: (1,0)-(2,0) — share endpoint (1,0) + wires = [ + [(0, 0), (10_000, 0)], + [(10_000, 0), (20_000, 0)], + ] + adjacency, iu_to_wires = _build_adjacency(wires) + assert 1 in adjacency[0] + assert 0 in adjacency[1] + + def test_build_adjacency_two_disconnected_wires(self): + wires = [ + [(0, 0), (10_000, 0)], + [(20_000, 0), (30_000, 0)], + ] + adjacency, _ = _build_adjacency(wires) + assert adjacency[0] == set() + assert adjacency[1] == set() + + def test_build_adjacency_iu_to_wires_maps_correctly(self): + wires = [ + [(0, 0), (10_000, 0)], + [(10_000, 0), (20_000, 0)], + ] + _, iu_to_wires = _build_adjacency(wires) + assert iu_to_wires[(10_000, 0)] == {0, 1} + assert iu_to_wires[(0, 0)] == {0} + + def test_build_adjacency_three_wires_at_junction(self): + # All three wires meet at (10,000, 0) + wires = [ + [(0, 0), (10_000, 0)], + [(10_000, 0), (20_000, 0)], + [(10_000, 0), (10_000, 10_000)], + ] + adjacency, _ = _build_adjacency(wires) + assert adjacency[0] == {1, 2} + assert adjacency[1] == {0, 2} + assert adjacency[2] == {0, 1} + + # --- _find_connected_wires --- + + def test_find_connected_wires_no_wire_at_point(self): + wires = [[(0, 0), (10_000, 0)]] + adjacency, iu_to_wires = _build_adjacency(wires) + visited, net_points = _find_connected_wires( + 5.0, 0.0, wires, iu_to_wires, adjacency + ) + assert visited is None + assert net_points is None + + def test_find_connected_wires_single_wire(self): + wires = [[(0, 0), (10_000, 0)]] + adjacency, iu_to_wires = _build_adjacency(wires) + visited, net_points = _find_connected_wires( + 0.0, 0.0, wires, iu_to_wires, adjacency + ) + assert visited == {0} + assert (0, 0) in net_points + assert (10_000, 0) in net_points + + def test_find_connected_wires_flood_fills_chain(self): + # Three wires in a chain: A-B-C-D + wires = [ + [(0, 0), (10_000, 0)], + [(10_000, 0), (20_000, 0)], + [(20_000, 0), (30_000, 0)], + ] + adjacency, iu_to_wires = _build_adjacency(wires) + visited, net_points = _find_connected_wires( + 0.0, 0.0, wires, iu_to_wires, adjacency + ) + assert visited == {0, 1, 2} + + def test_find_connected_wires_does_not_cross_gap(self): + # Two disconnected segments; query on segment 0 should not reach segment 1 + wires = [ + [(0, 0), (10_000, 0)], + [(20_000, 0), (30_000, 0)], + ] + adjacency, iu_to_wires = _build_adjacency(wires) + visited, _ = _find_connected_wires(0.0, 0.0, wires, iu_to_wires, adjacency) + assert visited == {0} + + # --- get_wire_connections (integration of internal functions) --- + + def test_get_wire_connections_no_wires(self): + sch = MagicMock() + sch.wire = [] + result = get_wire_connections(sch, "/fake/path.kicad_sch", 0.0, 0.0) + assert result == {"pins": [], "wires": []} + + def test_get_wire_connections_no_wire_at_point_returns_none(self): + sch = _make_schematic(_make_wire(0.0, 0.0, 1.0, 0.0)) + result = get_wire_connections(sch, "/fake/path.kicad_sch", 5.0, 0.0) + assert result is None + + def test_get_wire_connections_returns_wire_data(self): + sch = _make_schematic(_make_wire(0.0, 0.0, 1.0, 0.0)) + # Prevent _find_pins_on_net from iterating symbols + result = get_wire_connections(sch, "/fake/path.kicad_sch", 0.0, 0.0) + assert result is not None + assert result["pins"] == [] + assert len(result["wires"]) == 1 + wire = result["wires"][0] + assert wire["start"] == {"x": 0.0, "y": 0.0} + assert wire["end"] == {"x": 1.0, "y": 0.0} + + def test_get_wire_connections_chain_returns_all_wires(self): + sch = _make_schematic( + _make_wire(0.0, 0.0, 1.0, 0.0), + _make_wire(1.0, 0.0, 2.0, 0.0), + ) + result = get_wire_connections(sch, "/fake/path.kicad_sch", 0.0, 0.0) + assert result is not None + assert len(result["wires"]) == 2