diff --git a/python/commands/connection_schematic.py b/python/commands/connection_schematic.py index b4d04d0..299d8be 100644 --- a/python/commands/connection_schematic.py +++ b/python/commands/connection_schematic.py @@ -420,7 +420,9 @@ class ConnectionManager: } """ try: - netlist = {"nets": [], "components": []} + from commands.wire_connectivity import get_connections_for_net + + netlist: Dict[str, Any] = {"nets": [], "components": []} # Gather all components if hasattr(schematic, "symbol"): @@ -438,20 +440,19 @@ class ConnectionManager: } netlist["components"].append(component_info) - # Gather all nets from labels - if hasattr(schematic, "label"): - net_names = set() - for label in schematic.label: - if hasattr(label, "value"): - net_names.add(label.value) + # Gather all nets from labels and global labels + net_names: set = set() + for attr_name in ("label", "global_label"): + if hasattr(schematic, attr_name): + for label in getattr(schematic, attr_name): + if hasattr(label, "value"): + net_names.add(label.value) - # For each net, get connections - for net_name in net_names: - connections = ConnectionManager.get_net_connections( - schematic, net_name, schematic_path - ) - if connections: - netlist["nets"].append({"name": net_name, "connections": connections}) + sch_path_str = str(schematic_path) if schematic_path else "" + for net_name in net_names: + connections = get_connections_for_net(schematic, sch_path_str, net_name) + if connections: + netlist["nets"].append({"name": net_name, "connections": connections}) logger.info( f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components" diff --git a/python/commands/pin_locator.py b/python/commands/pin_locator.py index bfddca1..2db7aa5 100644 --- a/python/commands/pin_locator.py +++ b/python/commands/pin_locator.py @@ -126,16 +126,42 @@ class PinLocator: logger.error("No lib_symbols section found in schematic") return {} - # Find the specific symbol definition - for item in lib_symbols[1:]: # Skip 'lib_symbols' itself - if isinstance(item, list) and len(item) > 1 and item[0] == Symbol("symbol"): - symbol_name = str(item[1]).strip('"') - if symbol_name == lib_id: - # Found the symbol, parse pins - pins = self.parse_symbol_definition(item) - self.pin_definition_cache[cache_key] = pins - logger.info(f"Extracted {len(pins)} pins from {lib_id}") - return pins + # Find the specific symbol definition. + # KiCad lib_symbols may use a different name than the instance lib_id: + # instance lib_id: "stat-tis-custom:BAT_18650" + # lib_symbols name: "BAT_18650_3" (prefix stripped, unit suffix added) + # Strategy: exact match first, then bare-name prefix match. + bare_name = lib_id.split(":")[-1] if ":" in lib_id else lib_id + + best_match = None + for item in lib_symbols[1:]: + if not (isinstance(item, list) and len(item) > 1 and item[0] == Symbol("symbol")): + continue + symbol_name = str(item[1]).strip('"') + if symbol_name == lib_id: + best_match = item + break + if best_match is None: + sn_bare = symbol_name.split(":")[-1] if ":" in symbol_name else symbol_name + if sn_bare == bare_name or ( + sn_bare.startswith(bare_name) + and len(sn_bare) > len(bare_name) + and sn_bare[len(bare_name)] == "_" + and sn_bare[len(bare_name) + 1 :].isdigit() + ): + best_match = item + + if best_match is not None: + matched_name = str(best_match[1]).strip('"') + pins = self.parse_symbol_definition(best_match) + self.pin_definition_cache[cache_key] = pins + if matched_name != lib_id: + logger.info( + f"Matched {lib_id} → lib_symbols '{matched_name}' ({len(pins)} pins)" + ) + else: + logger.info(f"Extracted {len(pins)} pins from {lib_id}") + return pins logger.warning(f"Symbol {lib_id} not found in lib_symbols") return {} @@ -228,8 +254,26 @@ class PinLocator: else: return None - # Pin definition angle + symbol rotation = absolute outward direction + mirror_x = False + mirror_y = False + if hasattr(target_symbol, "mirror"): + mirror_val = ( + str(target_symbol.mirror.value) + if hasattr(target_symbol.mirror, "value") + else "" + ) + if mirror_val == "x": + mirror_x = True + elif mirror_val == "y": + mirror_y = True + pin_def_angle = pins[pin_number].get("angle", 0) + # Y-negate flips the angle across the x-axis + pin_def_angle = (360 - pin_def_angle) % 360 + if mirror_x: + pin_def_angle = (360 - pin_def_angle) % 360 + if mirror_y: + pin_def_angle = (180 - pin_def_angle) % 360 absolute_angle = (pin_def_angle + symbol_rotation) % 360 return absolute_angle @@ -270,12 +314,25 @@ class PinLocator: logger.error(f"Symbol {symbol_reference} not found in schematic") return None - # Get symbol position and rotation + # Get symbol position, rotation, and mirror state symbol_at = target_symbol.at.value symbol_x = float(symbol_at[0]) symbol_y = float(symbol_at[1]) symbol_rotation = float(symbol_at[2]) if len(symbol_at) > 2 else 0.0 + mirror_x = False + mirror_y = False + if hasattr(target_symbol, "mirror"): + mirror_val = ( + str(target_symbol.mirror.value) + if hasattr(target_symbol.mirror, "value") + else "" + ) + if mirror_val == "x": + mirror_x = True + elif mirror_val == "y": + mirror_y = True + # Get symbol lib_id lib_id = target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None if not lib_id: @@ -283,7 +340,8 @@ class PinLocator: return None logger.debug( - f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, lib_id={lib_id}" + f"Symbol {symbol_reference}: pos=({symbol_x}, {symbol_y}), rot={symbol_rotation}, " + f"mirror_x={mirror_x}, mirror_y={mirror_y}, lib_id={lib_id}" ) # Get pin definitions for this symbol @@ -319,10 +377,21 @@ class PinLocator: logger.debug(f"Pin {pin_number} relative position: ({pin_rel_x}, {pin_rel_y})") + # lib_symbols uses y-up; schematic uses y-down + pin_rel_y = -pin_rel_y + + # Mirror in local coords after y-negate (KiCad transform order) + # mirror_x = flip across X axis → negate y + # mirror_y = flip across Y axis → negate x + if mirror_x: + pin_rel_y = -pin_rel_y + if mirror_y: + pin_rel_x = -pin_rel_x + # Apply symbol rotation to pin position if symbol_rotation != 0: pin_rel_x, pin_rel_y = self.rotate_point(pin_rel_x, pin_rel_y, symbol_rotation) - logger.debug(f"After rotation {symbol_rotation}°: ({pin_rel_x}, {pin_rel_y})") + logger.debug(f"After transform (y-neg/mirror/rot): ({pin_rel_x}, {pin_rel_y})") # Calculate absolute position abs_x = symbol_x + pin_rel_x diff --git a/python/commands/wire_connectivity.py b/python/commands/wire_connectivity.py index b488fc1..7f9e70c 100644 --- a/python/commands/wire_connectivity.py +++ b/python/commands/wire_connectivity.py @@ -4,13 +4,18 @@ 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. + +Supports hierarchical (multi-sheet) schematics by recursively discovering +sub-sheet files and bridging nets via hierarchical labels / sheet pins. """ import logging from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple +import sexpdata from commands.pin_locator import PinLocator +from sexpdata import Symbol logger = logging.getLogger("kicad_interface") @@ -22,12 +27,49 @@ def _to_iu(x_mm: float, y_mm: float) -> Tuple[int, int]: return (round(x_mm * _IU_PER_MM), round(y_mm * _IU_PER_MM)) +def _load_sexp(schematic_path: str) -> list: + """Load and cache the raw sexpdata tree for a schematic file.""" + with open(schematic_path, "r", encoding="utf-8") as f: + return sexpdata.loads(f.read()) + + +def _parse_wires_sexp(sexp: list) -> List[List[Tuple[int, int]]]: + """Extract wire endpoints from raw sexpdata as IU tuples. + + Parses ``(wire (pts (xy X Y) (xy X Y)))`` directly, bypassing + kicad-skip which may silently drop elements. + """ + all_wires: List[List[Tuple[int, int]]] = [] + for item in sexp: + if not isinstance(item, list) or not item: + continue + if item[0] != Symbol("wire"): + continue + for sub in item: + if not isinstance(sub, list) or not sub or sub[0] != Symbol("pts"): + continue + pts: List[Tuple[int, int]] = [] + for xy_elem in sub[1:]: + if isinstance(xy_elem, list) and len(xy_elem) >= 3 and xy_elem[0] == Symbol("xy"): + pts.append(_to_iu(float(xy_elem[1]), float(xy_elem[2]))) + if len(pts) >= 2: + all_wires.append(pts) + return all_wires + + def _parse_wires(schematic: Any) -> List[List[Tuple[int, int]]]: - """Extract wire endpoints from a schematic object as IU tuples.""" - all_wires = [] + """Extract wire endpoints from a kicad-skip schematic object as IU tuples. + + Used by the single-sheet handlers (``get_wire_connections``, + ``list_floating_labels``, ``get_net_at_point``) which receive a kicad-skip + schematic. Multi-sheet code paths use :func:`_parse_wires_sexp` instead. + """ + all_wires: List[List[Tuple[int, int]]] = [] + if not hasattr(schematic, "wire"): + return all_wires for wire in schematic.wire: if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): - pts = [] + pts: List[Tuple[int, int]] = [] for point in wire.pts.xy: if hasattr(point, "value"): pts.append(_to_iu(float(point.value[0]), float(point.value[1]))) @@ -36,6 +78,54 @@ def _parse_wires(schematic: Any) -> List[List[Tuple[int, int]]]: return all_wires +def _parse_labels_sexp( + sexp: list, +) -> Tuple[Dict[Tuple[int, int], str], Dict[str, List[Tuple[int, int]]]]: + """Parse label, global_label, and hierarchical_label from raw sexpdata. + + Returns (point_to_label, label_to_points) in IU coordinates. + Bypasses kicad-skip which may not iterate all labels correctly. + """ + point_to_label: Dict[Tuple[int, int], str] = {} + label_to_points: Dict[str, List[Tuple[int, int]]] = {} + + label_types = {Symbol("label"), Symbol("global_label"), Symbol("hierarchical_label")} + + for item in sexp: + if not isinstance(item, list) or len(item) < 2: + continue + if item[0] not in label_types: + continue + name = str(item[1]).strip('"') + for sub in item[2:]: + if isinstance(sub, list) and sub and sub[0] == Symbol("at") and len(sub) >= 3: + pt = _to_iu(float(sub[1]), float(sub[2])) + point_to_label[pt] = name + label_to_points.setdefault(name, []).append(pt) + logger.debug( + f"Parsed {item[0]} '{name}' at IU {pt} " + f"(mm {float(sub[1])}, {float(sub[2])})" + ) + break + + return point_to_label, label_to_points + + +def _point_on_segment(px: int, py: int, ax: int, ay: int, bx: int, by: int) -> bool: + """Check if point (px,py) lies strictly between endpoints (ax,ay)-(bx,by). + + Only handles axis-aligned (horizontal/vertical) segments, which covers + virtually all KiCad schematic wires. + """ + if ay == by == py: + lo, hi = (ax, bx) if ax < bx else (bx, ax) + return lo < px < hi + if ax == bx == px: + lo, hi = (ay, by) if ay < by else (by, ay) + return lo < py < hi + return False + + def _build_adjacency( all_wires: List[List[Tuple[int, int]]], ) -> Tuple[List[Set[int]], Dict[Tuple[int, int], Set[int]]]: @@ -44,6 +134,9 @@ def _build_adjacency( Wires that share an endpoint are adjacent — this naturally handles junctions since all wires meeting at the same point get connected. + Also detects T-junctions where a wire endpoint falls on the interior of + another wire segment (common when KiCad doesn't split the longer wire). + 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 @@ -55,7 +148,22 @@ def _build_adjacency( for pt in pts: iu_to_wires.setdefault(pt, set()).add(i) - # Wires that share an IU endpoint are adjacent + # Detect T-junctions: a wire endpoint landing on the interior of another + # wire segment. When found, register the endpoint against that segment's + # wire index so adjacency is established through the shared point. + all_endpoints = list(iu_to_wires.keys()) + for i, pts in enumerate(all_wires): + if len(pts) < 2: + continue + ax, ay = pts[0] + bx, by = pts[-1] + for ep in all_endpoints: + if ep == (ax, ay) or ep == (bx, by): + continue + if _point_on_segment(ep[0], ep[1], ax, ay, bx, by): + iu_to_wires[ep].add(i) + + # Wires that share an IU endpoint (including T-junction points) 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) @@ -68,9 +176,17 @@ def _build_adjacency( def _parse_virtual_connections( - schematic: Any, schematic_path: Any + schematic: Any, schematic_path: Any, sexp: Optional[list] = None ) -> Tuple[Dict[Tuple[int, int], str], Dict[str, List[Tuple[int, int]]]]: - """Return virtual connectivity from net labels and power symbols. + """Return virtual connectivity from net labels, global labels, and power symbols. + + Labels (label, global_label, hierarchical_label) are parsed directly from the + raw sexpdata tree for reliability — kicad-skip's collection iteration can + silently miss elements. If the sexp tree cannot be loaded (e.g. the path + does not exist in unit tests), falls back to kicad-skip's ``schematic.label`` + so callers that pass a mock schematic still get the labels they registered. + + Power symbols are still resolved via kicad-skip's symbol collection. Returns a tuple of: - point_to_label: Dict[Tuple[int,int], str] — IU position → label name @@ -79,20 +195,39 @@ def _parse_virtual_connections( 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 sexp is None: + try: + sexp = _load_sexp(schematic_path) + except Exception as e: + logger.debug( + f"Could not load sexp for {schematic_path} ({e}); " + "falling back to kicad-skip label collection" + ) + sexp = None + + if sexp is not None: + point_to_label, label_to_points = _parse_labels_sexp(sexp) + logger.debug( + f"Parsed {sum(len(v) for v in label_to_points.values())} label instances " + f"across {len(label_to_points)} unique net names from {schematic_path}" + ) + else: + for attr in ("label", "global_label"): + if not hasattr(schematic, attr): + continue + for label in getattr(schematic, attr): + 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() @@ -132,12 +267,24 @@ def _find_connected_wires( ) -> Tuple: """BFS from query point. Returns (visited wire indices, net IU points) or (None, None). - Requires query point (x_mm, y_mm) to be exactly on a wire endpoint (exact IU match). + First tries exact IU match on a wire endpoint, then falls back to + checking if the point lies on the interior of any wire segment + (handles labels placed mid-wire). """ query_iu = _to_iu(x_mm, y_mm) # Find seed wires: exact IU match on the query endpoint seed_set = iu_to_wires.get(query_iu) + if not seed_set: + # Fallback: check if query point lies on the interior of any wire segment + px, py = query_iu + for i, pts in enumerate(all_wires): + if len(pts) >= 2 and _point_on_segment( + px, py, pts[0][0], pts[0][1], pts[-1][0], pts[-1][1] + ): + seed_set = {i} + iu_to_wires.setdefault(query_iu, set()).add(i) + break if not seed_set: return (None, None) seed_indices: Set[int] = set(seed_set) @@ -175,44 +322,133 @@ def _find_connected_wires( return (visited, net_points) +def _parse_symbol_instances_sexp( + sexp: list, +) -> List[Dict]: + """Parse all placed symbol instances from raw sexpdata. + + Returns a list of dicts with keys: ref, lib_id, x, y, rotation, mirror_x, mirror_y. + Bypasses kicad-skip's symbol collection which may miss elements. + """ + instances: List[Dict] = [] + for item in sexp: + if not isinstance(item, list) or not item or item[0] != Symbol("symbol"): + continue + + inst: Dict = { + "ref": None, + "lib_id": None, + "x": 0.0, + "y": 0.0, + "rotation": 0.0, + "mirror_x": False, + "mirror_y": False, + } + + for sub in item[1:]: + if not isinstance(sub, list) or not sub: + continue + tag = sub[0] + if tag == Symbol("lib_id") and len(sub) >= 2: + inst["lib_id"] = str(sub[1]).strip('"') + elif tag == Symbol("at") and len(sub) >= 3: + inst["x"] = float(sub[1]) + inst["y"] = float(sub[2]) + if len(sub) >= 4: + inst["rotation"] = float(sub[3]) + elif tag == Symbol("mirror"): + if len(sub) >= 2: + mv = str(sub[1]).strip('"') + if mv == "x": + inst["mirror_x"] = True + elif mv == "y": + inst["mirror_y"] = True + elif tag == Symbol("property") and len(sub) >= 3: + prop_name = str(sub[1]).strip('"') + if prop_name == "Reference": + inst["ref"] = str(sub[2]).strip('"') + + if inst["ref"] and inst["lib_id"]: + instances.append(inst) + + return instances + + def _find_pins_on_net( net_points: Set[Tuple[int, int]], schematic_path: Any, schematic: Any, + sexp: Optional[list] = None, ) -> List[Dict]: - """Find component pins that land on net points using exact IU matching. + """Find component pins that land on net points. + + Parses symbol instances directly from sexpdata to avoid kicad-skip's + collection iteration issues. Uses exact IU matching first, then falls + back to a ±1 IU tolerance for floating-point rounding edge cases. Returns a list of {"component": ref, "pin": pin_num} dicts. """ def _on_net(px_mm: float, py_mm: float) -> bool: - return _to_iu(px_mm, py_mm) in net_points + pt = _to_iu(px_mm, py_mm) + if pt in net_points: + return True + ix, iy = pt + for dx in (-1, 0, 1): + for dy in (-1, 0, 1): + if (ix + dx, iy + dy) in net_points: + return True + return False + + if sexp is None: + sexp = _load_sexp(schematic_path) + + logger.debug(f"Searching {len(net_points)} net points for matching pins") locator = PinLocator() - pins = [] - seen: Set[Tuple] = set() + instances = _parse_symbol_instances_sexp(sexp) + logger.debug(f"Found {len(instances)} symbol instances via sexpdata") - ref = None - for symbol in schematic.symbol: + pins: List[Dict] = [] + seen: Set[Tuple[str, str]] = set() + + for inst in instances: + ref = inst["ref"] try: - if not hasattr(symbol, "property") or not hasattr(symbol.property, "Reference"): + if ref.startswith("_TEMPLATE") or ref.startswith("#"): continue - ref = symbol.property.Reference.value - if ref.startswith("_TEMPLATE"): + + lib_id = inst["lib_id"] + pin_defs = locator.get_symbol_pins(Path(schematic_path), lib_id) + if not pin_defs: + logger.debug(f" {ref}: no pin definitions for lib_id={lib_id}") 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]): + + sym_x = inst["x"] + sym_y = inst["y"] + sym_rot = inst["rotation"] + mirror_x = inst["mirror_x"] + mirror_y = inst["mirror_y"] + + for pin_num, pdata in pin_defs.items(): + px, py = pdata["x"], pdata["y"] + # y-negate: lib_symbols y-up → schematic y-down + py = -py + if mirror_x: + py = -py + if mirror_y: + px = -px + if sym_rot != 0: + px, py = locator.rotate_point(px, py, sym_rot) + abs_x = sym_x + px + abs_y = sym_y + py + if _on_net(abs_x, abs_y): 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}" - ) + logger.warning(f"Error checking pins for {ref}: {e}") return pins @@ -492,3 +728,189 @@ def get_net_at_point( return {"net_name": net, "position": position, "source": "wire_endpoint"} return {"net_name": None, "position": position, "source": None} + + +# --------------------------------------------------------------------------- +# Multi-sheet (hierarchical) connectivity +# +# The functions below extend single-sheet net tracing to hierarchical KiCad +# projects: ``get_connections_for_net`` discovers and recurses into every +# referenced sub-sheet, processing each one with ``_process_single_sheet`` +# (which uses the sexp-based parsers above for reliability across all label +# kinds, including ``hierarchical_label``). +# --------------------------------------------------------------------------- + + +def _discover_sub_sheets(schematic_path: str) -> List[str]: + """Recursively discover all sub-sheet .kicad_sch files referenced by the schematic. + + Returns a list of absolute paths to sub-sheet files (does NOT include the + top-level schematic_path itself). + """ + parent_dir = Path(schematic_path).parent + result: List[str] = [] + try: + with open(schematic_path, "r", encoding="utf-8") as f: + content = f.read() + sexp = sexpdata.loads(content) + except Exception as e: + logger.warning(f"Could not parse {schematic_path} for sub-sheets: {e}") + return result + + for item in sexp: + if not isinstance(item, list) or not item or item[0] != Symbol("sheet"): + continue + for sub in item: + if not isinstance(sub, list) or len(sub) < 3: + continue + if sub[0] != Symbol("property"): + continue + prop_name = str(sub[1]).strip('"') + if prop_name == "Sheetfile": + sheet_file = str(sub[2]).strip('"') + sheet_path = parent_dir / sheet_file + if sheet_path.exists(): + abs_path = str(sheet_path.resolve()) + result.append(abs_path) + result.extend(_discover_sub_sheets(abs_path)) + else: + logger.warning(f"Sub-sheet not found: {sheet_path}") + return result + + +def _parse_hierarchical_labels_sexp( + schematic_path: str, +) -> Dict[str, List[Tuple[int, int]]]: + """Parse hierarchical_label elements from a .kicad_sch file using sexpdata. + + kicad-skip does not expose hierarchical labels, so we parse them directly. + Returns {label_name: [iu_position, ...]}. + """ + result: Dict[str, List[Tuple[int, int]]] = {} + try: + with open(schematic_path, "r", encoding="utf-8") as f: + content = f.read() + sexp = sexpdata.loads(content) + except Exception as e: + logger.warning(f"Could not parse {schematic_path} for hierarchical labels: {e}") + return result + + for item in sexp: + if not isinstance(item, list) or not item: + continue + if item[0] != Symbol("hierarchical_label"): + continue + if len(item) < 2: + continue + name = str(item[1]).strip('"') + for sub in item: + if isinstance(sub, list) and sub and sub[0] == Symbol("at") and len(sub) >= 3: + pt = _to_iu(float(sub[1]), float(sub[2])) + result.setdefault(name, []).append(pt) + break + return result + + +def _process_single_sheet( + schematic: Any, + schematic_path: str, + net_name: str, +) -> List[Dict]: + """Find pins connected to *net_name* on a single schematic sheet. + + Handles label, global_label, hierarchical_label, and power symbols. + All wire and label data is parsed directly from the raw .kicad_sch file + via sexpdata for maximum reliability. + """ + try: + sexp = _load_sexp(schematic_path) + except Exception as e: + logger.warning(f"Could not load sexp for {schematic_path}: {e}") + return [] + + all_wires = _parse_wires_sexp(sexp) + logger.debug(f"Parsed {len(all_wires)} wires from {schematic_path}") + + adjacency: List[Set[int]] = [] + iu_to_wires: Dict[Tuple[int, int], Set[int]] = {} + if all_wires: + adjacency, iu_to_wires = _build_adjacency(all_wires) + + point_to_label, label_to_points = _parse_virtual_connections( + schematic, schematic_path, sexp=sexp + ) + + seed_positions = label_to_points.get(net_name, []) + if not seed_positions: + logger.debug(f"No label positions found for net '{net_name}' in {schematic_path}") + return [] + + logger.debug( + f"Net '{net_name}': {len(seed_positions)} seed position(s) — " + f"{[f'({p[0]/10000},{p[1]/10000})' for p in seed_positions]}" + ) + + net_points: Set[Tuple[int, int]] = set() + + for seed_pt in seed_positions: + net_points.add(seed_pt) + if not all_wires: + continue + visited, pts = _find_connected_wires( + seed_pt[0] / _IU_PER_MM, + seed_pt[1] / _IU_PER_MM, + all_wires, + iu_to_wires, + adjacency, + point_to_label=point_to_label, + label_to_points=label_to_points, + ) + if pts: + logger.debug( + f"BFS from seed ({seed_pt[0]/10000},{seed_pt[1]/10000}) " + f"found {len(pts)} points via {len(visited) if visited else 0} wires" + ) + net_points.update(pts) + else: + logger.debug( + f"BFS from seed ({seed_pt[0]/10000},{seed_pt[1]/10000}) " + f"found NO connected wires" + ) + + logger.debug(f"Net '{net_name}': total {len(net_points)} IU points in net after BFS") + + return _find_pins_on_net(net_points, schematic_path, schematic, sexp=sexp) + + +def get_connections_for_net(schematic: Any, schematic_path: str, net_name: str) -> List[Dict]: + """Find all component pins connected to a named net across all schematic sheets. + + Recursively discovers sub-sheets, processes each sheet independently, and + merges results. Handles label, global_label, hierarchical_label, and + power symbol connections. + + Returns a list of {"component": ref, "pin": pin_num} dicts. + """ + from skip import Schematic as SkipSchematic + + seen: Set[Tuple[str, str]] = set() + all_pins: List[Dict] = [] + + def _collect(pins: List[Dict]) -> None: + for pin in pins: + key = (pin["component"], pin["pin"]) + if key not in seen: + seen.add(key) + all_pins.append(pin) + + _collect(_process_single_sheet(schematic, schematic_path, net_name)) + + sub_sheets = _discover_sub_sheets(schematic_path) + for sub_path in sub_sheets: + try: + sub_sch = SkipSchematic(sub_path) + _collect(_process_single_sheet(sub_sch, sub_path, net_name)) + except Exception as e: + logger.warning(f"Error processing sub-sheet {sub_path}: {e}") + + return all_pins diff --git a/python/kicad_interface.py b/python/kicad_interface.py index 717d927..73beba1 100644 --- a/python/kicad_interface.py +++ b/python/kicad_interface.py @@ -2286,13 +2286,15 @@ class KiCADInterface: """List all nets in a schematic with their connections""" logger.info("Listing schematic nets") try: - from pathlib import Path - from commands.wire_connectivity import ( _build_adjacency, + _discover_sub_sheets, + _load_sexp, + _parse_labels_sexp, _parse_virtual_connections, _parse_wires, count_pins_on_net, + get_connections_for_net, ) schematic_path = params.get("schematicPath") @@ -2303,16 +2305,39 @@ class KiCADInterface: 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) + # Collect net names from the top-level sheet using sexpdata. + # Falls back to kicad-skip's label collections when the file + # cannot be read (e.g. mocked schematics in unit tests). + net_names: set = set() + sexp_loaded = False + try: + sexp = _load_sexp(schematic_path) + sexp_loaded = True + _, label_to_points = _parse_labels_sexp(sexp) + net_names.update(label_to_points.keys()) + except Exception as e: + logger.debug( + f"Could not parse labels from {schematic_path} via sexp ({e}); " + "falling back to kicad-skip label collections" + ) + for attr in ("label", "global_label"): + if not hasattr(schematic, attr): + continue + for label in getattr(schematic, attr): + if hasattr(label, "value"): + net_names.add(label.value) + + # Collect net names from all sub-sheets (only when the parent + # sheet was readable; fake/mock paths skip recursion entirely). + if sexp_loaded: + sub_sheets = _discover_sub_sheets(schematic_path) + for sub_path in sub_sheets: + try: + sub_sexp = _load_sexp(sub_path) + _, sub_label_to_points = _parse_labels_sexp(sub_sexp) + net_names.update(sub_label_to_points.keys()) + except Exception as e: + logger.warning(f"Error reading sub-sheet {sub_path}: {e}") # Pre-build shared wire graph structures for efficiency all_wires = _parse_wires(schematic) @@ -2324,9 +2349,7 @@ class KiCADInterface: nets = [] for net_name in sorted(net_names): - connections = ConnectionManager.get_net_connections( - schematic, net_name, Path(schematic_path) - ) + connections = get_connections_for_net(schematic, schematic_path, net_name) pin_count = count_pins_on_net( schematic, schematic_path, @@ -2931,6 +2954,8 @@ class KiCADInterface: """Get all connections for a named net""" logger.info("Getting net connections") try: + from commands.wire_connectivity import get_connections_for_net + schematic_path = params.get("schematicPath") net_name = params.get("netName") @@ -2941,7 +2966,7 @@ class KiCADInterface: if not schematic: return {"success": False, "message": "Failed to load schematic"} - connections = ConnectionManager.get_net_connections(schematic, net_name) + connections = get_connections_for_net(schematic, schematic_path, net_name) return {"success": True, "connections": connections} except Exception as e: logger.error(f"Error getting net connections: {str(e)}")