Merge pull request #62 from Mehanik/feature/wire-connectivity-lookup

feat: add get_wire_connections — schematic net traversal from a point

python/commands/wire_connectivity.py

@@ -0,0 +1,278 @@
+"""
+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
+
+
+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 _build_adjacency(
+    all_wires: List[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
+        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)
+
+    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).
+
+    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: 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)
+    queue = list(seed_indices)
+    net_points: Set[Tuple[int, int]] = set()
+    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]:
+            if neighbor_idx not in visited:
+                visited.add(neighbor_idx)
+                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)
+
+
+def _find_pins_on_net(
+    net_points: Set[Tuple[int, int]],
+    schematic_path,
+    schematic,
+) -> List[Dict]:
+    """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:
+        return _to_iu(px_mm, py_mm) 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 '<unknown>'}: {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, net labels, and power symbols.
+
+    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
+    Or None if no wire endpoint found within tolerance of the query point.
+    """
+    all_wires = _parse_wires(schematic)
+    if not all_wires:
+        return {"pins": [], "wires": []}
+
+    adjacency, iu_to_wires = _build_adjacency(all_wires)
+
+    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
+
+    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}

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,53 @@ 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 commands.wire_connectivity import get_wire_connections
+
+            schematic_path = params.get("schematicPath")
+            x = params.get("x")
+            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",
+                }
+
+            try:
+                x, y = float(x), float(y)
+            except (TypeError, ValueError):
+                return {
+                    "success": False,
+                    "message": "Parameters x and y must be numeric",
+                }
+
+            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"}
+
+            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": 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)}
+
     def _handle_run_erc(self, params):
         """Run Electrical Rules Check on a schematic via kicad-cli"""
         logger.info("Running ERC on schematic")

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",

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

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",

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, 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"),
+      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);
+      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",