From 700ad29bdd474dcefb6acc7ec29f16597095fb50 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Thu, 5 Mar 2026 08:19:06 -0700 Subject: [PATCH] Redesign audit_wiring output for large ICs Group results by net name instead of per-pin, keeping the summary compact enough to stay inline even for 100+ pin components. Add anomaly detection (unconnected pins, high-fanout nets, auto-named nets) and optional pin/net filters. Wire coordinates are now opt-in via include_wires flag to avoid flooding the calling LLM with coordinate noise. --- src/mckicad/tools/schematic_analysis.py | 140 ++++++++++++++++++------ tests/test_schematic_analysis.py | 86 +++++++++++++-- 2 files changed, 185 insertions(+), 41 deletions(-) diff --git a/src/mckicad/tools/schematic_analysis.py b/src/mckicad/tools/schematic_analysis.py index ca91a80..44054b9 100644 --- a/src/mckicad/tools/schematic_analysis.py +++ b/src/mckicad/tools/schematic_analysis.py @@ -1139,22 +1139,34 @@ def export_schematic_pdf( @mcp.tool() -def audit_wiring(schematic_path: str, reference: str) -> dict[str, Any]: - """Trace all wires connected to a component and report per-pin net membership. +def audit_wiring( + schematic_path: str, + reference: str, + pins: list[str] | None = None, + net: str | None = None, + include_wires: bool = False, +) -> dict[str, Any]: + """Audit wiring for a component, grouped by net name with anomaly detection. - For each pin on the specified component, reports the net it belongs to, - the wire segments in that net, and other pins sharing the same net. - Essential for debugging wiring errors — especially overlapping wire - segments that create unintended shorts. + Returns a compact net-grouped summary that fits inline even for 100+ pin + ICs. Each net entry lists which pins of the component belong to it, how + many wire segments exist, and what other components are connected. + + By default, wire segment coordinates are omitted to keep the response + compact. Pass ``include_wires=True`` to get full ``{start, end, uuid}`` + data for each wire (offloaded to a sidecar file when large). Args: schematic_path: Path to a .kicad_sch file. reference: Component reference designator (e.g. ``U8``, ``R1``). + pins: Optional list of pin numbers to filter to (e.g. ``["103", "97"]``). + net: Optional net name to filter to (only pins on this net are included). + include_wires: Include wire segment coordinates and UUIDs. Default False. Returns: - Dictionary with ``reference``, ``pin_nets`` list (per-pin net info - with wire segments and connected pins), and ``detail_file`` when - the result exceeds the inline threshold. + Dictionary with ``net_summary`` (grouped by net), ``anomalies`` + (unconnected pins, high-fanout nets, auto-named nets), and optionally + ``detail_file`` when wire data is offloaded. """ err = _require_sch_api() if err: @@ -1182,14 +1194,13 @@ def audit_wiring(schematic_path: str, reference: str) -> dict[str, Any]: root_to_net: dict[tuple[float, float], str] = {} for n_name, n_pins in net_graph.items(): for p in n_pins: - # Find coordinate for this pin for coord, pin_list in pin_at.items(): for pe in pin_list: if pe["reference"] == p["reference"] and pe["pin"] == p["pin"]: root = find_fn(coord) root_to_net[root] = n_name - # Build wire segments by net root (each wire's endpoints share a root) + # Build wire segments by net root root_to_wires: dict[tuple[float, float], list[dict[str, Any]]] = {} for ws in wire_segments: s_coord = (round(ws["start"]["x"], 2), round(ws["start"]["y"], 2)) @@ -1210,50 +1221,115 @@ def audit_wiring(schematic_path: str, reference: str) -> dict[str, Any]: "schematic_path": schematic_path, } - # Build per-pin audit report - pin_nets: list[dict[str, Any]] = [] + # Apply pin filter early + if pins is not None: + pin_set = set(pins) + target_pins = [(p, c) for p, c in target_pins if p in pin_set] + + # Group target pins by net name + net_groups: dict[str, list[tuple[str, tuple[float, float]]]] = {} for pin_num, coord in target_pins: root = find_fn(coord) net_name = root_to_net.get(root, "unconnected") + net_groups.setdefault(net_name, []).append((pin_num, coord)) + + # Apply net filter + if net is not None: + net_groups = {k: v for k, v in net_groups.items() if k == net} + + # Build net_summary + net_summary: dict[str, dict[str, Any]] = {} + for net_name, group_pins in sorted(net_groups.items()): + pin_nums = sorted([p for p, _ in group_pins]) + + # Collect wires for this net (use root of any pin in the group) + _, first_coord = group_pins[0] + root = find_fn(first_coord) wires = root_to_wires.get(root, []) - # Collect other pins on the same net - connected_pins: list[dict[str, str]] = [] + # Collect other components connected to this net + connected_to: list[dict[str, str]] = [] if net_name in net_graph: for p in net_graph[net_name]: - if not (p["reference"] == reference and p["pin"] == pin_num): - connected_pins.append(p) + if p["reference"] != reference: + connected_to.append(p) - pin_nets.append({ - "pin": pin_num, - "net": net_name, + entry: dict[str, Any] = { + "pins": pin_nums, "wire_count": len(wires), - "wires": wires, - "connected_pins": connected_pins, - }) + "connected_to": connected_to, + } + if include_wires: + entry["wires"] = wires + + net_summary[net_name] = entry + + # Compute anomalies + unconnected_pins = sorted( + [p for p, _ in group_pins] + for net_name, group_pins in net_groups.items() + if net_name == "unconnected" + ) + # Flatten list of lists into a single list + unconnected_flat: list[str] = [] + for pl in unconnected_pins: + unconnected_flat.extend(pl) + + high_fanout_nets: list[dict[str, Any]] = [] + for net_name, entry in net_summary.items(): + if net_name == "unconnected": + continue + pin_count = len(entry["pins"]) + total_connections = len(entry["connected_to"]) + if pin_count > 2 or total_connections > 4: + high_fanout_nets.append({ + "net": net_name, + "pin_count": pin_count, + "total_connections": total_connections, + }) + + auto_named_nets: list[dict[str, Any]] = [ + {"net": net_name, "pins": entry["pins"]} + for net_name, entry in net_summary.items() + if net_name.startswith("Net-(") + ] + + anomalies: dict[str, Any] = { + "unconnected_pins": unconnected_flat, + "high_fanout_nets": high_fanout_nets, + "auto_named_nets": auto_named_nets, + } + + total_pin_count = sum(len(e["pins"]) for e in net_summary.values()) logger.info( - "audit_wiring for %s: %d pins, %d total wire segments", - reference, len(pin_nets), - sum(pn["wire_count"] for pn in pin_nets), + "audit_wiring for %s: %d pins across %d nets", + reference, total_pin_count, len(net_summary), ) result: dict[str, Any] = { "success": True, "reference": reference, - "pin_count": len(pin_nets), + "pin_count": total_pin_count, + "net_count": len(net_summary), + "net_summary": net_summary, + "anomalies": anomalies, "engine": _get_schematic_engine(), "schematic_path": schematic_path, } - if len(pin_nets) > INLINE_RESULT_THRESHOLD: + # Offload wire detail to sidecar only when wires are included and large + if include_wires and len(net_summary) > INLINE_RESULT_THRESHOLD: + wire_detail = { + n: {"wires": e.get("wires", [])} for n, e in net_summary.items() + } detail_path = write_detail_file( - schematic_path, f"audit_{reference}.json", pin_nets, + schematic_path, f"audit_{reference}_wires.json", wire_detail, ) result["detail_file"] = detail_path - result["pin_nets_preview"] = pin_nets[:INLINE_RESULT_THRESHOLD] - else: - result["pin_nets"] = pin_nets + # Strip wires from inline summary to keep it compact + for entry in net_summary.values(): + entry.pop("wires", None) return result diff --git a/tests/test_schematic_analysis.py b/tests/test_schematic_analysis.py index a638b96..c08b551 100644 --- a/tests/test_schematic_analysis.py +++ b/tests/test_schematic_analysis.py @@ -154,8 +154,9 @@ class TestAuditWiring: ) assert result["success"] is True assert result["reference"] == "R1" - assert "pin_nets" in result or "pin_nets_preview" in result + assert "net_summary" in result assert result.get("pin_count", 0) > 0 + assert "net_count" in result def test_audit_nonexistent_component(self, populated_schematic): from mckicad.tools.schematic_analysis import audit_wiring @@ -184,7 +185,8 @@ class TestAuditWiring: ) assert result["success"] is False - def test_audit_pin_net_structure(self, populated_schematic): + def test_audit_net_summary_structure(self, populated_schematic): + """Each net entry has pins, wire_count, connected_to — no wires by default.""" from mckicad.tools.schematic_analysis import audit_wiring result = audit_wiring( @@ -192,13 +194,79 @@ class TestAuditWiring: reference="R1", ) if result["success"]: - pin_nets = result.get("pin_nets", result.get("pin_nets_preview", [])) - for pn in pin_nets: - assert "pin" in pn - assert "net" in pn - assert "wire_count" in pn - assert "wires" in pn - assert "connected_pins" in pn + for _net_name, entry in result["net_summary"].items(): + assert "pins" in entry + assert isinstance(entry["pins"], list) + assert "wire_count" in entry + assert "connected_to" in entry + assert isinstance(entry["connected_to"], list) + # No wire coords by default + assert "wires" not in entry + + def test_audit_include_wires(self, populated_schematic): + """When include_wires=True, each net entry contains a wires list.""" + from mckicad.tools.schematic_analysis import audit_wiring + + result = audit_wiring( + schematic_path=populated_schematic, + reference="R1", + include_wires=True, + ) + if result["success"]: + for _net_name, entry in result["net_summary"].items(): + assert "wires" in entry + assert isinstance(entry["wires"], list) + + def test_audit_pin_filter(self, populated_schematic): + """Filtering by pin number limits the net_summary to matching nets.""" + from mckicad.tools.schematic_analysis import audit_wiring + + result = audit_wiring( + schematic_path=populated_schematic, + reference="R1", + pins=["1"], + ) + if result["success"]: + # Only nets containing pin "1" should appear + for entry in result["net_summary"].values(): + assert "1" in entry["pins"] + + def test_audit_net_filter(self, populated_schematic): + """Filtering by net name limits the summary to that net only.""" + from mckicad.tools.schematic_analysis import audit_wiring + + # First get all nets to pick one + full = audit_wiring( + schematic_path=populated_schematic, + reference="R1", + ) + if full["success"] and full["net_summary"]: + target_net = next(iter(full["net_summary"])) + filtered = audit_wiring( + schematic_path=populated_schematic, + reference="R1", + net=target_net, + ) + assert filtered["success"] is True + assert list(filtered["net_summary"].keys()) == [target_net] + + def test_audit_anomalies_structure(self, populated_schematic): + """Anomalies dict always present with expected keys.""" + from mckicad.tools.schematic_analysis import audit_wiring + + result = audit_wiring( + schematic_path=populated_schematic, + reference="R1", + ) + if result["success"]: + assert "anomalies" in result + anomalies = result["anomalies"] + assert "unconnected_pins" in anomalies + assert isinstance(anomalies["unconnected_pins"], list) + assert "high_fanout_nets" in anomalies + assert isinstance(anomalies["high_fanout_nets"], list) + assert "auto_named_nets" in anomalies + assert isinstance(anomalies["auto_named_nets"], list) @requires_sch_api