feat: multi-sheet net connectivity + sexp-based parsing reliability

Adds robust multi-sheet (hierarchical) net connectivity for KiCad
schematics and switches the wire/label parsing to a direct sexpdata
pipeline that bypasses kicad-skip's collection iteration, which was
silently dropping wires, labels, and symbol instances on real-world
schematics.

python/commands/wire_connectivity.py
  - New sexpdata helpers: _load_sexp, _parse_wires_sexp,
    _parse_labels_sexp, _parse_symbol_instances_sexp,
    _parse_hierarchical_labels_sexp, _discover_sub_sheets.
  - _build_adjacency now detects T-junctions (endpoint landing on
    another wire's interior segment) so adjacency captures connections
    KiCad doesn't represent as separate wire segments.
  - _find_connected_wires gains an interior-segment fallback so labels
    placed mid-wire still seed BFS correctly.
  - _parse_virtual_connections gathers label / global_label /
    hierarchical_label and power-symbol pin positions, with a
    kicad-skip fallback for unit tests that mock the schematic.
  - _find_pins_on_net rebuilds pin positions from sexpdata symbol
    instances (with mirror_x/mirror_y/rotation handling) and uses a
    plus/minus 1 IU tolerance for floating-point edge cases.
  - get_connections_for_net walks the top sheet plus every recursively
    discovered sub-sheet, deduping pins across sheets.

python/commands/pin_locator.py
  - lib_id matching now falls back to a bare-name + unit-suffix match
    so instances like "stat-tis-custom:BAT_18650" resolve to
    lib_symbols entries like "BAT_18650_3".
  - Pin position math now y-negates lib_symbols coords, applies
    mirror_x/mirror_y in local coords before rotation, and propagates
    the same transform into get_pin_orientation so downstream callers
    get a correct outward angle for mirrored symbols.

python/commands/connection_schematic.py
  - generate_netlist now collects nets from both label and
    global_label and routes them through get_connections_for_net so
    netlists reflect cross-sheet connectivity instead of single-sheet
    label-only nets.

python/kicad_interface.py
  - list_schematic_nets aggregates net names across the top sheet and
    all sub-sheets via the sexp helpers, then resolves connections
    using get_connections_for_net.
  - get_net_connections delegates to get_connections_for_net for
    consistent multi-sheet results.
This commit is contained in:
William Viana
2026-04-19 23:42:15 -07:00
parent 5b7434fdef
commit 046f33d876
4 changed files with 599 additions and 82 deletions

View File

@@ -420,7 +420,9 @@ class ConnectionManager:
} }
""" """
try: try:
netlist = {"nets": [], "components": []} from commands.wire_connectivity import get_connections_for_net
netlist: Dict[str, Any] = {"nets": [], "components": []}
# Gather all components # Gather all components
if hasattr(schematic, "symbol"): if hasattr(schematic, "symbol"):
@@ -438,20 +440,19 @@ class ConnectionManager:
} }
netlist["components"].append(component_info) netlist["components"].append(component_info)
# Gather all nets from labels # Gather all nets from labels and global labels
if hasattr(schematic, "label"): net_names: set = set()
net_names = set() for attr_name in ("label", "global_label"):
for label in schematic.label: if hasattr(schematic, attr_name):
if hasattr(label, "value"): for label in getattr(schematic, attr_name):
net_names.add(label.value) if hasattr(label, "value"):
net_names.add(label.value)
# For each net, get connections sch_path_str = str(schematic_path) if schematic_path else ""
for net_name in net_names: for net_name in net_names:
connections = ConnectionManager.get_net_connections( connections = get_connections_for_net(schematic, sch_path_str, net_name)
schematic, net_name, schematic_path if connections:
) netlist["nets"].append({"name": net_name, "connections": connections})
if connections:
netlist["nets"].append({"name": net_name, "connections": connections})
logger.info( logger.info(
f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components" f"Generated netlist with {len(netlist['nets'])} nets and {len(netlist['components'])} components"

View File

@@ -126,16 +126,42 @@ class PinLocator:
logger.error("No lib_symbols section found in schematic") logger.error("No lib_symbols section found in schematic")
return {} return {}
# Find the specific symbol definition # Find the specific symbol definition.
for item in lib_symbols[1:]: # Skip 'lib_symbols' itself # KiCad lib_symbols may use a different name than the instance lib_id:
if isinstance(item, list) and len(item) > 1 and item[0] == Symbol("symbol"): # instance lib_id: "stat-tis-custom:BAT_18650"
symbol_name = str(item[1]).strip('"') # lib_symbols name: "BAT_18650_3" (prefix stripped, unit suffix added)
if symbol_name == lib_id: # Strategy: exact match first, then bare-name prefix match.
# Found the symbol, parse pins bare_name = lib_id.split(":")[-1] if ":" in lib_id else lib_id
pins = self.parse_symbol_definition(item)
self.pin_definition_cache[cache_key] = pins best_match = None
logger.info(f"Extracted {len(pins)} pins from {lib_id}") for item in lib_symbols[1:]:
return pins 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") logger.warning(f"Symbol {lib_id} not found in lib_symbols")
return {} return {}
@@ -228,8 +254,26 @@ class PinLocator:
else: else:
return None 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) 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 absolute_angle = (pin_def_angle + symbol_rotation) % 360
return absolute_angle return absolute_angle
@@ -270,12 +314,25 @@ class PinLocator:
logger.error(f"Symbol {symbol_reference} not found in schematic") logger.error(f"Symbol {symbol_reference} not found in schematic")
return None return None
# Get symbol position and rotation # Get symbol position, rotation, and mirror state
symbol_at = target_symbol.at.value symbol_at = target_symbol.at.value
symbol_x = float(symbol_at[0]) symbol_x = float(symbol_at[0])
symbol_y = float(symbol_at[1]) symbol_y = float(symbol_at[1])
symbol_rotation = float(symbol_at[2]) if len(symbol_at) > 2 else 0.0 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 # Get symbol lib_id
lib_id = target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None lib_id = target_symbol.lib_id.value if hasattr(target_symbol, "lib_id") else None
if not lib_id: if not lib_id:
@@ -283,7 +340,8 @@ class PinLocator:
return None return None
logger.debug( 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 # 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})") 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 # Apply symbol rotation to pin position
if symbol_rotation != 0: if symbol_rotation != 0:
pin_rel_x, pin_rel_y = self.rotate_point(pin_rel_x, pin_rel_y, symbol_rotation) 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 # Calculate absolute position
abs_x = symbol_x + pin_rel_x abs_x = symbol_x + pin_rel_x

View File

@@ -4,13 +4,18 @@ Wire Connectivity Analysis for KiCad Schematics
Traces wire networks from a point and finds connected component pins. 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 Uses KiCad's internal integer unit system (10,000 IU per mm) for exact
coordinate matching, mirroring KiCad's own connectivity algorithm. 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 import logging
from pathlib import Path from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Tuple from typing import Any, Dict, List, Optional, Set, Tuple
import sexpdata
from commands.pin_locator import PinLocator from commands.pin_locator import PinLocator
from sexpdata import Symbol
logger = logging.getLogger("kicad_interface") 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)) 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]]]: def _parse_wires(schematic: Any) -> List[List[Tuple[int, int]]]:
"""Extract wire endpoints from a schematic object as IU tuples.""" """Extract wire endpoints from a kicad-skip schematic object as IU tuples.
all_wires = []
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: for wire in schematic.wire:
if hasattr(wire, "pts") and hasattr(wire.pts, "xy"): if hasattr(wire, "pts") and hasattr(wire.pts, "xy"):
pts = [] pts: List[Tuple[int, int]] = []
for point in wire.pts.xy: for point in wire.pts.xy:
if hasattr(point, "value"): if hasattr(point, "value"):
pts.append(_to_iu(float(point.value[0]), float(point.value[1]))) 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 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( def _build_adjacency(
all_wires: List[List[Tuple[int, int]]], all_wires: List[List[Tuple[int, int]]],
) -> Tuple[List[Set[int]], Dict[Tuple[int, int], Set[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 Wires that share an endpoint are adjacent — this naturally handles
junctions since all wires meeting at the same point get connected. 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: Returns a tuple of:
- adjacency: list of sets, one per wire, containing adjacent wire indices - 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 - 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: for pt in pts:
iu_to_wires.setdefault(pt, set()).add(i) 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))] adjacency: List[Set[int]] = [set() for _ in range(len(all_wires))]
for wire_set in iu_to_wires.values(): for wire_set in iu_to_wires.values():
wire_list = list(wire_set) wire_list = list(wire_set)
@@ -68,9 +176,17 @@ def _build_adjacency(
def _parse_virtual_connections( 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]]]]: ) -> 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: Returns a tuple of:
- point_to_label: Dict[Tuple[int,int], str] — IU position → label name - 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] = {} point_to_label: Dict[Tuple[int, int], str] = {}
label_to_points: Dict[str, List[Tuple[int, int]]] = {} label_to_points: Dict[str, List[Tuple[int, int]]] = {}
if hasattr(schematic, "label"): if sexp is None:
for label in schematic.label: try:
try: sexp = _load_sexp(schematic_path)
if not hasattr(label, "value"): except Exception as e:
continue logger.debug(
name = label.value f"Could not load sexp for {schematic_path} ({e}); "
if not hasattr(label, "at") or not hasattr(label.at, "value"): "falling back to kicad-skip label collection"
continue )
coords = label.at.value sexp = None
pt = _to_iu(float(coords[0]), float(coords[1]))
point_to_label[pt] = name if sexp is not None:
label_to_points.setdefault(name, []).append(pt) point_to_label, label_to_points = _parse_labels_sexp(sexp)
except Exception as e: logger.debug(
logger.warning(f"Error parsing net label: {e}") 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"): if hasattr(schematic, "symbol"):
locator = PinLocator() locator = PinLocator()
@@ -132,12 +267,24 @@ def _find_connected_wires(
) -> Tuple: ) -> Tuple:
"""BFS from query point. Returns (visited wire indices, net IU points) or (None, None). """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) query_iu = _to_iu(x_mm, y_mm)
# Find seed wires: exact IU match on the query endpoint # Find seed wires: exact IU match on the query endpoint
seed_set = iu_to_wires.get(query_iu) 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: if not seed_set:
return (None, None) return (None, None)
seed_indices: Set[int] = set(seed_set) seed_indices: Set[int] = set(seed_set)
@@ -175,44 +322,133 @@ def _find_connected_wires(
return (visited, net_points) 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( def _find_pins_on_net(
net_points: Set[Tuple[int, int]], net_points: Set[Tuple[int, int]],
schematic_path: Any, schematic_path: Any,
schematic: Any, schematic: Any,
sexp: Optional[list] = None,
) -> List[Dict]: ) -> 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. Returns a list of {"component": ref, "pin": pin_num} dicts.
""" """
def _on_net(px_mm: float, py_mm: float) -> bool: 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() locator = PinLocator()
pins = [] instances = _parse_symbol_instances_sexp(sexp)
seen: Set[Tuple] = set() logger.debug(f"Found {len(instances)} symbol instances via sexpdata")
ref = None pins: List[Dict] = []
for symbol in schematic.symbol: seen: Set[Tuple[str, str]] = set()
for inst in instances:
ref = inst["ref"]
try: try:
if not hasattr(symbol, "property") or not hasattr(symbol.property, "Reference"): if ref.startswith("_TEMPLATE") or ref.startswith("#"):
continue 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 continue
all_pins = locator.get_all_symbol_pins(Path(schematic_path), ref)
if not all_pins: sym_x = inst["x"]
continue sym_y = inst["y"]
for pin_num, pin_data in all_pins.items(): sym_rot = inst["rotation"]
if _on_net(pin_data[0], pin_data[1]): 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) key = (ref, pin_num)
if key not in seen: if key not in seen:
seen.add(key) seen.add(key)
pins.append({"component": ref, "pin": pin_num}) pins.append({"component": ref, "pin": pin_num})
except Exception as e: except Exception as e:
logger.warning( logger.warning(f"Error checking pins for {ref}: {e}")
f"Error checking pins for {ref if ref is not None else '<unknown>'}: {e}"
)
return pins return pins
@@ -492,3 +728,189 @@ def get_net_at_point(
return {"net_name": net, "position": position, "source": "wire_endpoint"} return {"net_name": net, "position": position, "source": "wire_endpoint"}
return {"net_name": None, "position": position, "source": None} 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

View File

@@ -1963,13 +1963,15 @@ class KiCADInterface:
"""List all nets in a schematic with their connections""" """List all nets in a schematic with their connections"""
logger.info("Listing schematic nets") logger.info("Listing schematic nets")
try: try:
from pathlib import Path
from commands.wire_connectivity import ( from commands.wire_connectivity import (
_build_adjacency, _build_adjacency,
_discover_sub_sheets,
_load_sexp,
_parse_labels_sexp,
_parse_virtual_connections, _parse_virtual_connections,
_parse_wires, _parse_wires,
count_pins_on_net, count_pins_on_net,
get_connections_for_net,
) )
schematic_path = params.get("schematicPath") schematic_path = params.get("schematicPath")
@@ -1980,16 +1982,39 @@ class KiCADInterface:
if not schematic: if not schematic:
return {"success": False, "message": "Failed to load schematic"} return {"success": False, "message": "Failed to load schematic"}
# Get all net names from labels and global labels # Collect net names from the top-level sheet using sexpdata.
net_names = set() # Falls back to kicad-skip's label collections when the file
if hasattr(schematic, "label"): # cannot be read (e.g. mocked schematics in unit tests).
for label in schematic.label: net_names: set = set()
if hasattr(label, "value"): sexp_loaded = False
net_names.add(label.value) try:
if hasattr(schematic, "global_label"): sexp = _load_sexp(schematic_path)
for label in schematic.global_label: sexp_loaded = True
if hasattr(label, "value"): _, label_to_points = _parse_labels_sexp(sexp)
net_names.add(label.value) 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 # Pre-build shared wire graph structures for efficiency
all_wires = _parse_wires(schematic) all_wires = _parse_wires(schematic)
@@ -2001,9 +2026,7 @@ class KiCADInterface:
nets = [] nets = []
for net_name in sorted(net_names): for net_name in sorted(net_names):
connections = ConnectionManager.get_net_connections( connections = get_connections_for_net(schematic, schematic_path, net_name)
schematic, net_name, Path(schematic_path)
)
pin_count = count_pins_on_net( pin_count = count_pins_on_net(
schematic, schematic,
schematic_path, schematic_path,
@@ -2608,6 +2631,8 @@ class KiCADInterface:
"""Get all connections for a named net""" """Get all connections for a named net"""
logger.info("Getting net connections") logger.info("Getting net connections")
try: try:
from commands.wire_connectivity import get_connections_for_net
schematic_path = params.get("schematicPath") schematic_path = params.get("schematicPath")
net_name = params.get("netName") net_name = params.get("netName")
@@ -2618,7 +2643,7 @@ class KiCADInterface:
if not schematic: if not schematic:
return {"success": False, "message": "Failed to load 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} return {"success": True, "connections": connections}
except Exception as e: except Exception as e:
logger.error(f"Error getting net connections: {str(e)}") logger.error(f"Error getting net connections: {str(e)}")