KiCad - the KiCad Bridge

@@ -0,0 +1,401 @@+"""Live, undoable copper edits through the official KiCad IPC API (KiCad 10.0.1+).++The caller plans the path. KiCad checks candidate copper using a snapshot before+we commit it. No File/Revert, background clicks, or writes to the open board file.+"""+from __future__ import annotations++from collections import Counter+from contextlib import contextmanager+import hashlib+import json+import math+import os+from pathlib import Path+import shutil+import subprocess+import sys+import tempfile+import threading++from handlers import route+from parsers.pcb import parse_pcb_text++_LOCK = threading.RLock()+_DEPS = Path(__file__).resolve().parents[1] / "routing_deps"+if _DEPS.is_dir():+    sys.path.insert(0, str(_DEPS))+++class RoutingError(Exception):+    def __init__(self, code, message, **detail):+        super().__init__(message)+        self.code, self.detail = code, detail+++def _error(exc):+    return {"success": False, "errorCode": getattr(exc, "code", "ipc_unavailable"),+            "error": str(exc), **getattr(exc, "detail", {}),+            "_hint": "Read kicad_routing_state with the exact filePath and socketPath. "+                     "Enable Preferences > Plugins > Enable IPC API server in KiCad 10.0.1+. "+                     "Install requirements-routing.txt with the bridge's Python. "+                     "After a mutation timeout inspect the live board; never blindly replay it."}+++def _path(value):+    if not isinstance(value, str) or not Path(value).is_absolute():+        raise RoutingError("invalid_board_path", "filePath must be an absolute .kicad_pcb path")+    if Path(value).suffix.lower() != ".kicad_pcb":+        raise RoutingError("invalid_board_path", "filePath must name a .kicad_pcb")+    return os.path.normcase(os.path.realpath(value))+++@contextmanager+def _connect(args):+    from kipy import KiCad+    from kipy.board import Board+    from kipy.proto.common.types.base_types_pb2 import DocumentType+    wanted = _path(args.get("filePath"))+    socket = args.get("socketPath")+    if socket is not None and (not isinstance(socket, str) or not socket.startswith("ipc://")):+        raise RoutingError("invalid_socket", "socketPath must be a local ipc:// socket")+    api = KiCad(socket_path=socket, client_name="adom-routing-" + os.urandom(8).hex(), timeout_ms=5000)+    try:+        version = api.get_version()+        if (version.major, version.minor, version.patch) < (10, 0, 1):+            raise RoutingError("unsupported_kicad", "Live routing requires KiCad 10.0.1 or newer")+        docs = api.get_open_documents(DocumentType.DOCTYPE_PCB)+        def document_path(d):+            filename = Path(d.board_filename)+            return str(filename if filename.is_absolute() else Path(d.project.path) / filename)+        matches = [d for d in docs if _path(document_path(d)) == wanted]+        if len(matches) != 1:+            raise RoutingError("board_mismatch", "This IPC endpoint does not expose exactly the requested board",+                               openBoards=[document_path(d) for d in docs])+        yield Board(api._client, matches[0])+    finally:+        # kicad-python 0.8 has no public close/context-manager API.+        if api._client.connected:+            api._client._conn.close()+++def _snapshot(board, args):+    text = board.get_as_string()+    data = parse_pcb_text(text, args["filePath"])+    return text, data, hashlib.sha256(text.encode()).hexdigest()+++def _revision(args, revision):+    if args.get("expectedRevision") != revision:+        raise RoutingError("stale_board", "expectedRevision must match the current live board",+                           currentRevision=revision, mutated=False)+++def _number(value, label, positive=False):+    if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):+        raise RoutingError("invalid_geometry", f"{label} must be a finite number")+    if abs(value) > 2000 or (positive and value <= 0):+        raise RoutingError("invalid_geometry", f"{label} is outside the supported board range")+    rounded = round(value, 6)+    if positive and rounded <= 0:+        raise RoutingError("invalid_geometry", f"{label} is below KiCad nanometre precision")+    return rounded+++def _plan(data, args):+    """Pure planning/validation; every waypoint is checked before any mutation."""+    netname = args.get("net")+    if not isinstance(netname, str) or not netname:+        raise RoutingError("missing_net", "net must be an exact, nonempty net name")+    nets = [n for n in data["nets"] if n["name"] == netname]+    if len(nets) != 1:+        raise RoutingError("unknown_net", f"No unique net named {netname!r}")+    net = {"name": netname, "number": nets[0].get("number"), "format": data["net_format"]}+    pads = route._pads(data)+    copper = route._copper_layers(data)+    width = _number(args.get("width", 0.25), "width", True)+    size = _number(args.get("viaSize", 0.6), "viaSize", True)+    drill = _number(args.get("viaDrill", 0.3), "viaDrill", True)+    if drill >= size:+        raise RoutingError("invalid_geometry", "viaDrill must be smaller than viaSize")+    paths = args.get("paths", [args.get("points")])+    if not isinstance(paths, list) or not paths or len(paths) > 128:+        raise RoutingError("invalid_path", "Supply points or up to 128 paths for one net")+    segments, vias, blocks = [], [], []+    for path in paths:+        if not isinstance(path, list) or not 2 <= len(path) <= 512:+            raise RoutingError("invalid_path", "Each path needs 2..512 waypoints")+        resolved = []+        for spec in path:+            coordinates = None+            if isinstance(spec, (list, tuple)):+                if len(spec) != 2:+                    raise RoutingError("invalid_geometry", "Coordinates require exactly [x, y]")+                coordinates = spec+            elif isinstance(spec, dict) and not spec.get("pad"):+                coordinates = [spec.get("x"), spec.get("y")]+            if coordinates is not None:+                for value in coordinates:+                    _number(value, "coordinate")+            point, pad, err = route._resolve_point(spec, pads)+            if err:+                raise RoutingError(err["errorCode"], err["error"])+            point = [_number(v, "coordinate") for v in point]+            if pad and pad["netName"] != netname:+                raise RoutingError("net_mismatch", f"Pad {pad['name']} is not on {netname}")+            layer = spec.get("layer") if isinstance(spec, dict) else None+            if layer is not None and layer not in copper:+                raise RoutingError("unknown_layer", f"Unknown copper layer: {layer}")+            resolved.append((point, pad, layer))+        layer = args.get("layer", resolved[0][2] or (resolved[0][1] or {}).get("side") or "F.Cu")+        if layer not in copper:+            raise RoutingError("unknown_layer", f"Unknown copper layer: {layer}")+        for i, (point, pad, explicit_layer) in enumerate(resolved):+            next_layer = explicit_layer or layer+            if i:+                previous = resolved[i - 1][0]+                if previous != point:+                    segments.append({"start": previous, "end": point, "layer": layer, "width": width})+                    blocks.append(route._segment_sexpr(previous, point, width, layer, net))+            # A layer marker changes the outgoing layer AT this point, including+            # the final waypoint (where a via can reach a back-side pad).+            if next_layer != layer:+                vias.append({"at": point, "layers": ["F.Cu", "B.Cu"], "size": size, "drill": drill})+                blocks.append(route._via_sexpr(point, size, drill, ["F.Cu", "B.Cu"], net))+                layer = next_layer+            if pad and not (layer in pad["layers"] or "*.Cu" in pad["layers"]):+                raise RoutingError("pad_layer_mismatch", f"Pad {pad['name']} has no copper on {layer}")+    if not segments and not vias:+        raise RoutingError("zero_length", "No nonzero segments or layer transitions")+    if len(segments) + len(vias) > 1024:+        raise RoutingError("route_too_large", "Limit each call to 1024 copper items")+    return {"netName": netname, "segments": segments, "vias": vias, "blocks": blocks}+++def _drc(info, args, text):+    exe = info.get("kicad_cli_exe")+    if not exe or not Path(exe).is_file():+        raise RoutingError("drc_unavailable", "kicad-cli is required for routing validation")+    source = Path(args["filePath"])+    with tempfile.TemporaryDirectory(prefix="adom-routing-drc-") as tmp:+        candidate = Path(tmp) / source.name+        candidate.write_text(text, encoding="utf-8")+        # Preserve the actual project's rules and exclusions; never save/revert+        # the live document merely to get a headless DRC snapshot.+        copied = []+        for suffix in (".kicad_pro", ".kicad_dru"):+            sibling = source.with_suffix(suffix)+            if sibling.exists():+                shutil.copy2(sibling, candidate.with_suffix(suffix))+                copied.append(str(sibling))+        report = Path(tmp) / "drc.json"+        proc = subprocess.run([exe, "pcb", "drc", "--format", "json", "--severity-all",+                               "--all-track-errors", "--output", str(report), str(candidate)],+                              capture_output=True, text=True, timeout=90)+        if proc.returncode or not report.is_file():+            raise RoutingError("drc_failed", proc.stderr or proc.stdout or "No DRC report")+        result = json.loads(report.read_text(encoding="utf-8"))+    violations = result.get("violations", [])+    unconnected = result.get("unconnected_items", [])+    return {"errors": sum(v.get("severity") == "error" for v in violations),+            "warnings": sum(v.get("severity") == "warning" for v in violations),+            "unconnected": len(unconnected), "violations": violations,+            "unconnectedItems": unconnected,+            "clean": not violations and not unconnected,+            "source": "live-editor-snapshot", "projectRulesCopied": copied}+++def _fingerprint(v):+    return json.dumps([v.get("type"), v.get("severity"), v.get("description"),+                       sorted(json.dumps(i.get("pos"), sort_keys=True) for i in v.get("items", []))], sort_keys=True)+++def _new_errors(before, after):+    existing = Counter(_fingerprint(v) for v in before["violations"] if v.get("severity") == "error")+    added = []+    for v in after["violations"]:+        if v.get("severity") != "error":+            continue+        key = _fingerprint(v)+        if existing[key]:+            existing[key] -= 1+        else:+            added.append(v)+    return added+++def _items(plan, board):+    from kipy.board_types import Track, Via+    from kipy.geometry import Vector2+    from kipy.proto.board.board_types_pb2 import BoardLayer+    net = next(n for n in board.get_nets() if n.name == plan["netName"])+    def vector(point):+        return Vector2.from_xy(*(round(x * 1_000_000) for x in point))+    items = []+    for row in plan["segments"]:+        item = Track()+        item.start, item.end = vector(row["start"]), vector(row["end"])+        item.width = round(row["width"] * 1_000_000)+        item.layer = BoardLayer.Value("BL_" + row["layer"].replace(".", "_"))+        item.net = net+        items.append(item)+    for row in plan["vias"]:+        item = Via()+        item.position = vector(row["at"])+        item.diameter = round(row["size"] * 1_000_000)+        item.drill_diameter = round(row["drill"] * 1_000_000)+        item.net = net+        items.append(item)+    return items+++def _transaction(board, operation, message):+    commit = board.begin_commit()+    pushing = False+    try:+        result = operation()+        pushing = True+        board.push_commit(commit, message)+        return result+    except Exception as exc:+        try:+            board.drop_commit(commit)+        except Exception as rollback:+            raise RoutingError("mutation_outcome_unknown", str(exc), rollbackError=str(rollback),+                               mutated=None) from exc+        if pushing:+            raise RoutingError("mutation_outcome_unknown", str(exc), mutated=None,+                               rollbackAttempted=True) from exc+        raise RoutingError("mutation_rolled_back", str(exc), mutated=False) from exc+++def handle_routing_state(info, args):+    try:+        with _LOCK, _connect(args) as board:+            text, data, revision = _snapshot(board, args)+            pads = route._pads(data)+            nets = []+            live_pads = board.get_pads()+            for name in sorted(n["name"] for n in data["nets"] if n["name"]):+                net_pads = [p for p in live_pads if p.net.name == name]+                remaining = {p.id.value: p for p in net_pads}+                groups = []+                while remaining:+                    seed = next(iter(remaining.values()))+                    seen, frontier = {seed.id.value}, [seed]+                    while frontier:+                        neighbors = board.get_connected_items(frontier)+                        frontier = [n for n in neighbors if n.id.value not in seen]+                        seen.update(n.id.value for n in frontier)+                        if len(seen) > 100000:+                            raise RoutingError("board_too_large", "Connectivity walk exceeds 100000 items")+                    group = [key for key in remaining if key in seen]+                    for key in group:+                        remaining.pop(key)+                    groups.append(group)+                nets.append({"name": name, "padCount": len(net_pads), "padGroups": groups,+                             "connected": len(groups) <= 1, "missingConnections": max(0, len(groups) - 1)})+            current = _snapshot(board, args)[2]+            if current != revision:+                raise RoutingError("stale_board", "Board changed during connectivity inspection; read again", currentRevision=current)+            return {"success": True, "boardPath": args["filePath"], "revision": revision,+                    "source": "live-editor", "pads": pads, "nets": nets,+                    "netsRemaining": [n["name"] for n in nets if not n["connected"]],+                    "segments": data["segments"], "vias": data["vias"],+                    "copperLayers": route._copper_layers(data),+                    "_hint": "Use this revision as expectedRevision in kicad_route_net. Coordinates are mm, +y down. "+                             "Connectivity is measured by KiCad; kicad_routing_validate checks clearance and unconnected items."}+    except Exception as exc:+        return _error(exc)+++def handle_route_net(info, args):+    try:+        with _LOCK, _connect(args) as board:+            text, data, revision = _snapshot(board, args)+            _revision(args, revision)+            plan = _plan(data, args)+            check = None+            if args.get("validate", True) is not False:+                before = _drc(info, args, text)+                check = _drc(info, args, route._splice(text, plan["blocks"]))+                added = _new_errors(before, check)+                if added:+                    raise RoutingError("drc_rejected", "Candidate route adds DRC errors; no live copper was changed",+                                       violations=added, mutated=False)+            public = {k: v for k, v in plan.items() if k != "blocks"}+            if args.get("dryRun") is True:+                return {"success": True, "dryRun": True, "revision": revision, **public,+                        "drc": check, "_hint": "Plan validated without editing. Submit with dryRun:false and the same revision."}+            # DRC runs outside KiCad. Re-read immediately before begin_commit so+            # another caller/user's edits during validation invalidate this plan.+            _revision(args, _snapshot(board, args)[2])+            items = _items(plan, board)+            def create():+                created = board.create_items(items)+                if len(created) != len(items) or any(not i.id.value for i in created):+                    raise RuntimeError("KiCad did not create every requested copper item")+                return created+            created = _transaction(board, create, "Adom route " + plan["netName"])+            result = {"success": True, "mutated": True, "source": "live-editor", **public,+                      "itemIds": [i.id.value for i in created], "undoSteps": 1,+                      "drc": check, "saved": False,+                      "_hint": "Copper committed live as one KiCad Undo step. Inspect routing_state before the next route. "+                               "The file is saved only with save:true; routing_validate checks the unsaved live board."}+            # Commit succeeded. A subsequent read/save failure must NEVER be+            # reported as a failed route inviting the caller to add it twice.+            try:+                result["revision"] = _snapshot(board, args)[2]+                if args.get("save") is True:+                    board.save()+                    result["saved"] = True+            except Exception as exc:+                result["postCommitError"] = str(exc)+            return result+    except Exception as exc:+        return _error(exc)+++def handle_remove_route(info, args):+    try:+        with _LOCK, _connect(args) as board:+            _, _, revision = _snapshot(board, args)+            _revision(args, revision)+            ids = args.get("itemIds")+            if not isinstance(ids, list) or not ids or len(ids) > 1024 or any(not isinstance(i, str) for i in ids):+                raise RoutingError("invalid_item_ids", "Supply 1..1024 copper itemIds returned by route_net")+            wanted = set(ids)+            copper = list(board.get_tracks()) + list(board.get_vias())+            items = [i for i in copper if i.id.value in wanted]+            if len(items) != len(wanted) or any(i.locked for i in items):+                raise RoutingError("invalid_item_ids", "Some IDs are missing, locked, or are not copper; nothing removed")+            _revision(args, _snapshot(board, args)[2])+            def remove():+                board.remove_items(items)+                remaining = {i.id.value for i in list(board.get_tracks()) + list(board.get_vias())}+                if wanted & remaining:+                    raise RuntimeError("KiCad did not remove every requested copper item")+            _transaction(board, remove, "Adom remove route")+            result = {"success": True, "mutated": True, "removedIds": ids, "undoSteps": 1,+                      "_hint": "Copper removed live. Use KiCad Undo to restore it, or inspect routing_state and route again."}+            try:+                result["revision"] = _snapshot(board, args)[2]+            except Exception as exc:+                result["postCommitError"] = str(exc)+            return result+    except Exception as exc:+        return _error(exc)+++def handle_routing_validate(info, args):+    try:+        with _LOCK, _connect(args) as board:+            text, _, revision = _snapshot(board, args)+            result = _drc(info, args, text)+            current = _snapshot(board, args)[2]+            return {"success": True, "revision": revision, "currentRevision": current,+                    "stale": current != revision, **result,+                    "_hint": "This is KiCad DRC on the live editor snapshot with sibling project rules. "+                             "A stale result is superseded by newer edits. clean requires no violations or unconnected items."}+    except Exception as exc:+        return _error(exc)

@@ -0,0 +1,618 @@+"""Handlers for the copper-drawing verbs: add_track, add_via, route, board_pads.++DETERMINISTIC routing -- the same s-expression approach `place_footprint` proved,+pointed at the user's REAL board instead of a preview. A track is+`(segment (start ..) (end ..) (width ..) (layer ..) (net ..) (uuid ..))` spliced+in before the board's final paren; a via is `(via (at ..) (size ..) (drill ..)+(layers ..) (net ..) (uuid ..))`. No GUI, no protobuf, no autorouter.++The AI is the router. `board_pads` hands it pad centres in BOARD coordinates+with their nets; it picks the path; `add_track` (one segment at a time, for a+demo you can watch) or `route` (a whole polyline in one write) lays the copper.++Coordinates are millimetres in KiCad's board frame: +x right, **+y DOWN**.++Open boards are refused before writes. Use kicad_route_net for live editor+transactions; these file verbs never invoke File > Revert.+"""++from __future__ import annotations++import math+import json+import os+import shutil+import sys+import tempfile+import uuid+from collections import Counter+from pathlib import Path++sys.path.insert(0, str(Path(__file__).parent.parent))+from parsers.pcb import parse_pcb_text  # noqa: E402++IS_WINDOWS = sys.platform == "win32"++DEFAULT_WIDTH = 0.2+DEFAULT_VIA_SIZE = 0.6+DEFAULT_VIA_DRILL = 0.3+DEFAULT_LAYER = "F.Cu"+++# ── numbers / geometry ───────────────────────────────────────────────++def _num(v) -> str:+    """KiCad-style trimmed float ('1.5', not '1.500000')."""+    s = f"{float(v):.6f}".rstrip("0").rstrip(".")+    return "0" if s in ("", "-", "-0") else s+++def _dist(a, b) -> float:+    return math.hypot(b[0] - a[0], b[1] - a[1])+++# ── board loading ────────────────────────────────────────────────────++def _load(args: dict, *, read_only=False):+    """(path, text, board, err). `err` is a ready-to-return dict on failure."""+    raw = args.get("filePath") or args.get("boardPath") or ""+    if not raw:+        return None, "", None, {+            "success": False, "error": "No filePath specified", "errorCode": "missing_arg",+            "_hint": "Pass the absolute path to a .kicad_pcb, e.g. "+                     '{"filePath":"C:/designs/foo/foo.kicad_pcb"}. '+                     "Call kicad_board_pads on it first to see pads, nets and copper layers."}+    path = Path(raw)+    if not read_only and (_lock_file(path) or _editor_hwnd(path)):+        return None, "", None, {"success": False, "errorCode": "board_open",+            "error": "This board is open in KiCad; file routing would conflict with live edits",+            "_hint": "Use kicad_routing_state and kicad_route_net for live edits, or close this board before offline routing."}+    if path.suffix.lower() != ".kicad_pcb":+        return None, "", None, {+            "success": False, "error": f"not a board file: {path.name}", "errorCode": "not_a_board",+            "_hint": "These verbs edit a .kicad_pcb. For a schematic there is nothing to route."}+    if not path.exists():+        return None, "", None, {+            "success": False, "error": f"board not found: {path}", "errorCode": "board_not_found",+            "_hint": "Absolute path, no %VAR% expansion. kicad_place_footprint can make a scratch "+                     "board to route on if you have no project yet."}+    try:+        text = path.read_text(encoding="utf-8")+    except OSError as e:+        return None, "", None, {"success": False, "error": f"could not read board: {e}",+                                "errorCode": "board_unreadable", "_hint": "Is the file locked by another process?"}+    if not text.lstrip().startswith("(kicad_pcb"):+        return None, "", None, {+            "success": False, "error": "file does not start with '(kicad_pcb'",+            "errorCode": "not_a_board",+            "_hint": "A legacy .brd or a truncated file. kicad_format_upgrade converts old formats."}+    try:+        board = parse_pcb_text(text, path)+    except Exception as e:  # pylint: disable=broad-except+        return None, "", None, {+            "success": False, "error": f"could not parse board: {e}", "errorCode": "board_unparsable",+            "_hint": "Our reader is not KiCad's parser. kicad_lint_board gets KiCad's own verdict."}+    return path, text, board, None+++def _copper_layers(board: dict) -> list[str]:+    return [l.get("name", "") for l in board.get("layers", []) if l.get("name", "").endswith(".Cu")]+++# ── nets ─────────────────────────────────────────────────────────────++def _net_key(board: dict, number, name):+    """The identity of a net on THIS board: its number pre-KiCad-10, else its name."""+    return number if board.get("net_format") == "number" else (name or "")+++def _resolve_net(board: dict, spec) -> tuple[dict | None, dict | None]:+    """Resolve a net name or number to a reference this board can carry.++    Never invents one: a track on a net the board does not know is copper KiCad+    will not connect to anything.+    """+    nets = board.get("nets", [])+    fmt = board.get("net_format", "number")+    names = [n.get("name", "") for n in nets]++    # Strings are always exact names, including numeric names such as "123".+    if spec is None:+        spec = ""+    if isinstance(spec, str):+        for n in nets:+            if n.get("name", "") == spec:+                return {"format": fmt, "number": n.get("number"), "name": spec,+                        "key": _net_key(board, n.get("number"), spec)}, None+    elif not isinstance(spec, bool) and isinstance(spec, (int, float)) and math.isfinite(spec) and spec == int(spec):+        if fmt == "name" and spec == 0:+            return {"format": fmt, "number": None, "name": "", "key": ""}, None+        if fmt == "number":+            for n in nets:+                if n.get("number") == spec:+                    return {"format": fmt, "number": int(spec), "name": n.get("name", ""), "key": int(spec)}, None+    return None, {"success": False, "errorCode": "unknown_net", "error": f"No exact net matching {spec!r}",+                  "nets": names[:60], "_hint": "Use an exact net name. Integer net codes are supported only for pre-KiCad-10 boards."}+++def _net_sexpr(net: dict) -> str:+    """The (net ...) line in the dialect this board speaks -- or nothing at all+    for deliberately unconnected copper on a KiCad 10 board, which has no net 0+    to name."""+    if net["format"] == "number":+        return f'\t\t(net {int(net["number"] or 0)})\n'+    if not net["name"]:+        return ""+    return f'\t\t(net {json.dumps(net["name"], ensure_ascii=False)})\n'+++def _default_width(board: dict, net_key=None) -> float:+    """Match the board's own copper: this net's width, else the commonest width."""+    on_net = [s["width"] for s in board.get("segments", [])+              if s.get("width") and (net_key is None+                                     or _net_key(board, s.get("net"), s.get("net_name")) == net_key)]+    if on_net:+        return Counter(on_net).most_common(1)[0][0]+    any_w = [s["width"] for s in board.get("segments", []) if s.get("width")]+    if any_w:+        return Counter(any_w).most_common(1)[0][0]+    return DEFAULT_WIDTH+++# ── pads in board coordinates ────────────────────────────────────────++def _pads(board: dict) -> list[dict]:+    """Every pad, transformed from footprint-local into BOARD coordinates.++    KiCad rotates a footprint's pads by the footprint orientation in a y-DOWN+    frame (trigo.cpp RotatePoint): x' = x·cos+y·sin, y' = -x·sin+y·cos.+    """+    out = []+    for fp in board.get("footprints", []):+        pos = fp.get("position") or {}+        fx, fy = float(pos.get("x", 0.0)), float(pos.get("y", 0.0))+        ang = math.radians(float(pos.get("rotation", 0.0) or 0.0))+        cos_a, sin_a = math.cos(ang), math.sin(ang)+        ref = fp.get("reference", "") or ""+        for pad in fp.get("pads", []):+            if "x" not in pad or "y" not in pad:+                continue+            px, py = float(pad["x"]), float(pad["y"])+            out.append({+                "ref": ref,+                "pad": pad.get("number", ""),+                "name": f"{ref}.{pad.get('number', '')}",+                "x": round(fx + px * cos_a + py * sin_a, 6),+                "y": round(fy - px * sin_a + py * cos_a, 6),+                "net": pad.get("net_number"),+                "netName": pad.get("net_name", ""),+                "netKey": _net_key(board, pad.get("net_number"), pad.get("net_name")),+                "type": pad.get("type", ""),+                "layers": pad.get("layers", []),+                "side": fp.get("layer", ""),+                "footprint": fp.get("footprint", ""),+            })+    return out+++def _pad_reaches(pad: dict | None, layer: str) -> bool:+    """Is this pad actually on that copper layer?++    An SMD pad lives on ONE side; a track that ends on it from the other side+    looks connected on screen and is a dangling track to DRC. Through-hole pads+    carry the "*.Cu" wildcard and are reachable from every layer.+    """+    if not pad:+        return True+    layers = pad.get("layers") or []+    if not layers:+        return True  # nothing to judge it by; let DRC have the last word+    for l in layers:+        if l == layer or l == "*.Cu" or (l.endswith("*") and layer.startswith(l[:-1])):+            return True+    return not any(l.endswith(".Cu") or l == "*.Cu" for l in layers)+++def _pad_net(board: dict, pad: dict | None) -> dict | None:+    """The net a pad sits on, as a reference the writers can emit -- or None for+    a pad with no net (a mounting hole, a fiducial)."""+    if not pad or pad.get("netKey") in (None, 0, ""):+        return None+    return {"format": board.get("net_format", "number"), "number": pad.get("net"),+            "name": pad.get("netName") or "", "key": pad.get("netKey")}+++def _find_pad(pads: list[dict], spec: str) -> tuple[dict | None, dict | None]:+    """'R1.2' / 'R1-2' / 'R1 2' -> that pad."""+    key = str(spec).strip().replace("-", ".").replace(" ", ".").upper()+    for p in pads:+        if p["name"].upper() == key:+            return p, None+    ref = key.split(".")[0]+    near = [p["name"] for p in pads if p["ref"].upper() == ref]+    return None, {"success": False, "error": f"no pad '{spec}' on this board",+                  "errorCode": "unknown_pad",+                  "candidates": near[:20] or [p["name"] for p in pads][:20],+                  "_hint": "Pads are addressed 'REF.PAD' (e.g. 'U1.7'). "+                           "kicad_board_pads lists every pad with its board coordinates."}+++def _resolve_point(spec, pads: list[dict]) -> tuple[tuple | None, dict | None, dict | None]:+    """(x, y), the pad it came from (or None), error. Accepts [x,y], {x,y},+    'REF.PAD' or {"pad":"REF.PAD"}."""+    if isinstance(spec, str):+        pad, err = _find_pad(pads, spec)+        if err:+            return None, None, err+        return (pad["x"], pad["y"]), pad, None+    if isinstance(spec, dict):+        if spec.get("pad"):+            pad, err = _find_pad(pads, spec["pad"])+            if err:+                return None, None, err+            return (pad["x"], pad["y"]), pad, None+        if "x" in spec and "y" in spec:+            try:+                return (float(spec["x"]), float(spec["y"])), None, None+            except (TypeError, ValueError):+                pass+    if isinstance(spec, (list, tuple)) and len(spec) >= 2:+        try:+            return (float(spec[0]), float(spec[1])), None, None+        except (TypeError, ValueError):+            pass+    return None, None, {"success": False, "error": f"could not read a point from {spec!r}",+                        "errorCode": "bad_point",+                        "_hint": 'A point is [x,y] in mm, {"x":..,"y":..}, or a pad name like "R1.2". '+                                 "+y is DOWN in KiCad board coordinates."}+++# ── s-expression emission ────────────────────────────────────────────++def _segment_sexpr(a, b, width, layer, net: dict) -> str:+    return ('\t(segment\n'+            f'\t\t(start {_num(a[0])} {_num(a[1])})\n'+            f'\t\t(end {_num(b[0])} {_num(b[1])})\n'+            f'\t\t(width {_num(width)})\n'+            f'\t\t(layer "{layer}")\n'+            + _net_sexpr(net) ++            f'\t\t(uuid "{uuid.uuid4()}")\n'+            '\t)')+++def _via_sexpr(at, size, drill, layers, net: dict) -> str:+    return ('\t(via\n'+            f'\t\t(at {_num(at[0])} {_num(at[1])})\n'+            f'\t\t(size {_num(size)})\n'+            f'\t\t(drill {_num(drill)})\n'+            f'\t\t(layers "{layers[0]}" "{layers[1]}")\n'+            + _net_sexpr(net) ++            f'\t\t(uuid "{uuid.uuid4()}")\n'+            '\t)')+++def _splice(text: str, blocks: list[str]) -> str:+    """Insert board items before the closing paren (where KiCad keeps copper)."""+    idx = text.rstrip().rfind(")")+    return text[:idx].rstrip() + "\n" + "\n".join(blocks) + "\n)\n"+++def _write(path: Path, new_text: str, backup: bool, expected: str = None) -> tuple[str | None, dict | None]:+    """Validate copper, refuse live/changed boards, and replace using a unique temp file."""+    bak, tmp = None, None+    try:+        data = parse_pcb_text(new_text, path)+        original = parse_pcb_text(expected, path) if expected is not None else {"segments": [], "vias": []}+        old_ids = {item.get("uuid") for key in ("segments", "vias") for item in original[key]}+        for segment in data["segments"]:+            if segment.get("uuid") in old_ids:+                continue+            values = [segment["width"], *segment["start"].values(), *segment["end"].values()]+            if not all(math.isfinite(v) for v in values) or segment["width"] <= 0:+                raise ValueError("Track coordinates must be finite and width must be positive")+        for via in data["vias"]:+            if via.get("uuid") in old_ids:+                continue+            values = [via["size"], via["drill"], *via["position"].values()]+            if not all(math.isfinite(v) for v in values) or not 0 < via["drill"] < via["size"]:+                raise ValueError("Via geometry must be finite, with 0 < drill < size")+            if set(via["layers"]) != {"F.Cu", "B.Cu"}:+                raise ValueError("Only through vias spanning F.Cu to B.Cu are supported")+        if _lock_file(path) or _editor_hwnd(path):+            raise ValueError("Board was opened while planning; use live routing")+        if expected is not None and path.read_text(encoding="utf-8") != expected:+            raise ValueError("Board changed while planning; inspect it and retry")+        if backup:+            bak = str(path) + ".adom-bak"+            shutil.copy2(path, bak)+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent,+                                         prefix=path.name + ".", suffix=".adom-tmp", delete=False) as f:+            tmp = Path(f.name)+            f.write(new_text)+            f.flush()+            os.fsync(f.fileno())+        os.replace(tmp, path)+    except (OSError, ValueError, KeyError) as e:+        return None, {"success": False, "error": f"Board write refused: {e}",+                      "errorCode": "board_write_refused",+                      "_hint": "No board replacement was performed. Reinspect geometry and the live editor before retrying."}+    finally:+        if tmp is not None:+            tmp.unlink(missing_ok=True)+    return bak, None+++# ── the running pcbnew ───────────────────────────────────────────────++def _lock_file(path: Path) -> Path | None:+    """KiCad's own open-file marker, `~<name>.kicad_pcb.lck` beside the board.++    It is the one signal that works with no window list at all, and it is what+    tells us a write is about to race a live editor.+    """+    lck = path.parent / f"~{path.name}.lck"+    return lck if lck.exists() else None+++def _editor_hwnd(path: Path) -> int | None:+    """The PCB Editor window showing THIS board, if one is up."""+    try:+        from handlers import kicad_windows+        for row in kicad_windows.find(fresh=True):+            title = (row.get("title") or "")+            if path.stem.lower() in title.lower() and "pcb editor" in title.lower():+                return int(row.get("hwnd") or 0) or None+    except Exception:  # pylint: disable=broad-except+        pass+    return None+++def _after_write(path: Path, args: dict, result: dict) -> dict:+    """File writes never reload or revert an editor (which can lose user edits)."""+    result["source"] = "file"+    result["reloaded"] = False+    result["_hint"] += " Open the saved board to inspect it. For live editing use kicad_route_net."+    return result+++# ── verbs ────────────────────────────────────────────────────────────++def handle_board_pads(kicad_info: dict, args: dict) -> dict:+    """Read-only routing intelligence: pads in board coordinates, nets, layers."""+    path, _text, board, err = _load(args, read_only=True)+    if err:+        return err++    pads = _pads(board)+    net_filter = args.get("net")+    if net_filter not in (None, ""):+        net, nerr = _resolve_net(board, net_filter)+        if nerr:+            return nerr+        pads = [p for p in pads if p.get("netKey") == net["key"]]+    ref_filter = args.get("reference")+    if ref_filter:+        pads = [p for p in pads if p["ref"].upper() == str(ref_filter).upper()]++    by_net: dict[str, list[str]] = {}+    for p in _pads(board):+        if p.get("netKey") not in (None, 0, ""):+            by_net.setdefault(p.get("netName") or f"net{p['net']}", []).append(p["name"])++    segs = board.get("segments", [])+    routed = {_net_key(board, s.get("net"), s.get("net_name")) for s in segs}+    nets = [{"number": n.get("number"), "name": n.get("name", ""),+             "pads": len(by_net.get(n.get("name", ""), [])),+             "hasCopper": _net_key(board, n.get("number"), n.get("name")) in routed}+            for n in board.get("nets", []) if n.get("name")]+    needs_copper = sorted(n["name"] for n in nets if n["pads"] > 1 and not n["hasCopper"])++    return {+        "success": True,+        "boardPath": str(path),+        "pads": pads,+        "padCount": len(pads),+        "nets": nets,+        "copperLayers": _copper_layers(board),+        "netFormat": board.get("net_format"),+        "defaultWidth": _default_width(board),+        "existingTracks": len(segs),+        "existingVias": len(board.get("vias", [])),+        "boardDimensions": board.get("dimensions"),+        "netsWithoutTracks": needs_copper[:60],+        "source": "file", "connectivityChecked": False,+        "output": f"{len(pads)} pads, {len(nets)} nets, {len(segs)} tracks on "+                  f"{len(_copper_layers(board))} copper layers",+        "_hint": "Pad x/y are BOARD millimetres (+y is DOWN) with footprint rotation applied -- "+                 "feed them straight to kicad_route or kicad_add_track. Route a net by name and "+                 "pick your own waypoints. hasCopper means a track exists, not that the net is connected. "+                 "Use kicad_routing_state for live connectivity. "+                 "kicad_run_drc is the authority on clearance.",+    }+++def handle_add_track(kicad_info: dict, args: dict) -> dict:+    """Write one segment to a closed board file."""+    path, text, board, err = _load(args)+    if err:+        return err+    pads = _pads(board)++    start_spec = args.get("start", args.get("from"))+    end_spec = args.get("end", args.get("to"))+    if start_spec is None or end_spec is None:+        return {"success": False, "error": "add_track needs both start and end",+                "errorCode": "missing_arg",+                "_hint": 'kicad_add_track {"filePath":"...","start":"R1.2","end":[120.5,90],'+                         '"net":"GND","layer":"F.Cu"}. Points are pad names or [x,y] in mm.'}+    a, pad_a, err = _resolve_point(start_spec, pads)+    if err:+        return err+    b, pad_b, err = _resolve_point(end_spec, pads)+    if err:+        return err++    net_spec = args.get("net")+    if net_spec in (None, ""):+        net = _pad_net(board, pad_a) or _pad_net(board, pad_b)+        if net is None:+            return {"success": False, "error": "no net given and neither endpoint is a pad on a net",+                    "errorCode": "missing_net",+                    "_hint": 'Pass "net":"GND" (a name from kicad_board_pads), or "net":0 for '+                             "deliberately unconnected copper. Copper on the wrong net is a DRC error."}+    else:+        net, nerr = _resolve_net(board, net_spec)+        if nerr:+            return nerr++    for pad in (pad_a, pad_b):+        pad_net = _pad_net(board, pad)+        if pad_net and pad_net["key"] != net["key"]:+            return {"success": False,+                    "error": f"pad {pad['name']} is on net '{pad.get('netName')}', "+                             f"not the net you asked to route ('{net['name']}')",+                    "errorCode": "net_mismatch",+                    "_hint": "Routing a track to a pad on a different net is a short. Check the "+                             "pad's net with kicad_board_pads, or drop the net arg and let the "+                             "endpoints decide it."}++    if _dist(a, b) < 1e-6:+        return {"success": False, "error": "start and end are the same point",+                "errorCode": "zero_length",+                "_hint": "A zero-length track is a DRC error in KiCad. Give the segment a real span, "+                         "or use kicad_add_via for a layer change at one point."}++    layer = args.get("layer") or (pad_a or {}).get("side") or DEFAULT_LAYER+    copper = _copper_layers(board)+    if layer not in copper:+        return {"success": False, "error": f"'{layer}' is not a copper layer on this board",+                "errorCode": "unknown_layer", "copperLayers": copper,+                "_hint": "Tracks only live on copper. This board's copper layers are listed above; "+                         "a 2-layer board has only F.Cu and B.Cu."}+    width = float(args.get("width", _default_width(board, net["key"])))++    block = _segment_sexpr(a, b, width, layer, net)+    bak, werr = _write(path, _splice(text, [block]), args.get("backup", True) is not False, expected=text)+    if werr:+        return werr++    warnings = [f"pad {p['name']} is not on {layer} (it is on {', '.join(p.get('layers') or []) or 'no copper'}) "+                f"-- this end of the track is dangling, not connected"+                for p in (pad_a, pad_b) if not _pad_reaches(p, layer)]++    net_name = net["name"]+    result = {+        "success": True, "boardPath": str(path), "backupPath": bak, "warnings": warnings,+        "start": list(a), "end": list(b), "layer": layer, "width": width,+        "net": net["number"], "netName": net_name,+        "lengthMm": round(_dist(a, b), 4),+        "tracksOnBoard": len(board.get("segments", [])) + 1,+        "output": f"Track on {net_name or 'no net'}: ({_num(a[0])}, {_num(a[1])}) -> "+                  f"({_num(b[0])}, {_num(b[1])}) on {layer}, {_num(width)} mm wide",+        "_hint": "One segment written. Call again for the next one -- that is the trace-by-trace "+                 "loop. For a whole path in a single write use kicad_route. Nothing here checks "+                 "clearance: run kicad_run_drc when the net is done.",+    }+    return _after_write(path, args, result)+++def handle_add_via(kicad_info: dict, args: dict) -> dict:+    """One via -- how a route changes layer."""+    path, text, board, err = _load(args)+    if err:+        return err+    pads = _pads(board)++    at_spec = args.get("at", args.get("position"))+    if at_spec is None:+        return {"success": False, "error": "add_via needs `at`", "errorCode": "missing_arg",+                "_hint": 'kicad_add_via {"filePath":"...","at":[120,90],"net":"GND"}. '+                         "Place it where the track changes layer."}+    at, pad, err = _resolve_point(at_spec, pads)+    if err:+        return err++    net_spec = args.get("net")+    if net_spec in (None, ""):+        net = _pad_net(board, pad)+        if net is None:+            return {"success": False, "error": "no net given and `at` is not a pad on a net",+                    "errorCode": "missing_net",+                    "_hint": 'Pass "net":"GND". A via must carry the net of the track it joins.'}+    else:+        net, nerr = _resolve_net(board, net_spec)+        if nerr:+            return nerr++    copper = _copper_layers(board)+    layers = args.get("layers") or [copper[0] if copper else "F.Cu",+                                    copper[-1] if copper else "B.Cu"]+    if len(layers) < 2 or any(l not in copper for l in layers[:2]):+        return {"success": False, "error": f"via layers {layers} are not both copper on this board",+                "errorCode": "unknown_layer", "copperLayers": copper,+                "_hint": 'A via spans two copper layers, e.g. "layers":["F.Cu","B.Cu"].'}++    size = float(args.get("size", DEFAULT_VIA_SIZE))+    drill = float(args.get("drill", DEFAULT_VIA_DRILL))+    if drill >= size:+        return {"success": False, "error": f"drill {drill} is not smaller than pad size {size}",+                "errorCode": "bad_via_geometry",+                "_hint": "A via's drill must be smaller than its pad, or there is no annular ring."}++    block = _via_sexpr(at, size, drill, layers[:2], net)+    bak, werr = _write(path, _splice(text, [block]), args.get("backup", True) is not False, expected=text)+    if werr:+        return werr++    result = {+        "success": True, "boardPath": str(path), "backupPath": bak,+        "at": list(at), "size": size, "drill": drill, "layers": layers[:2],+        "net": net["number"], "netName": net["name"],+        "viasOnBoard": len(board.get("vias", [])) + 1,+        "output": f"Via at ({_num(at[0])}, {_num(at[1])}) joining {layers[0]}->{layers[1]} "+                  f"on {net['name'] or 'no net'}",+        "_hint": "The via is copper on its own -- the tracks either side of it still have to be "+                 "drawn (kicad_add_track), or let kicad_route place vias for you by giving a "+                 "waypoint a different layer. Check the size against the fab's rules: "+                 "kicad_list_design_rules.",+    }+    return _after_write(path, args, result)+++def handle_route(kicad_info: dict, args: dict) -> dict:+    """Offline counterpart of route_net. A closed board and explicit path are required."""+    from handlers.live_routing import _plan, RoutingError+    path, text, board, err = _load(args)+    if err:+        return err+    raw = args.get("points", args.get("path", []))+    if not isinstance(raw, list):+        return {"success": False, "errorCode": "invalid_path", "error": "points must be an array",+                "_hint": "Use points:[REF.PAD,[x,y],...] in board millimetres."}+    raw = list(raw)+    if args.get("from") is not None:+        raw.insert(0, args["from"])+    if args.get("to") is not None:+        raw.append(args["to"])+    net_spec = args.get("net")+    if net_spec is None:+        for spec in raw:+            _, pad, _ = _resolve_point(spec, _pads(board))+            if pad and pad.get("netName"):+                net_spec = pad["netName"]+                break+    net, err = _resolve_net(board, net_spec)+    if err:+        return err+    try:+        plan = _plan(board, {**args, "net": net["name"], "points": raw})+    except RoutingError as e:+        return {"success": False, "errorCode": e.code, "error": str(e),+                "_hint": "Inspect kicad_board_pads and correct the route before retrying."}+    bak, err = _write(path, _splice(text, plan["blocks"]), args.get("backup", True) is not False, expected=text)+    if err:+        return err+    return _after_write(path, args, {"success": True, "boardPath": str(path), "backupPath": bak,+        "netName": net["name"], "segments": plan["segments"], "vias": plan["vias"],+        "segmentCount": len(plan["segments"]), "viaCount": len(plan["vias"]),+        "_hint": "Copper written to the closed board. Run kicad_run_drc for clearance and connectivity."})

@@ -8,7 +8,26 @@ from __future__ import annotations  from pathlib import Path -from .sexpr import parse_file, find_node, find_nodes, node_value+from .sexpr import parse, parse_file, find_node, find_nodes, node_value+++def _net_ref(node: list | None, name_based: bool = False) -> tuple[int | None, str | None]:+    """Read a (net ...) reference in either board format.++    KiCad <=9 numbers its nets and declares them at the top of the board:+    `(net 3 "GND")` on a pad, `(net 3)` on a segment. KiCad 10 (file format+    20260206) dropped the table and the numbers entirely -- every reference is+    just `(net "GND")`. Both shapes appear in the wild, so read both.+    """+    if not node or len(node) < 2:+        return None, None+    first = node[1]+    if name_based:+        return None, first if isinstance(first, str) else None+    if isinstance(first, str) and first.lstrip("-").isdigit():+        name = node[2] if len(node) >= 3 and isinstance(node[2], str) else None+        return int(first), name+    return None, first if isinstance(first, str) else None   def parse_pcb(filepath: str | Path) -> dict:@@ -17,8 +36,47 @@ def parse_pcb(filepath: str | Path) -> dict:     return _extract_pcb(tree, filepath)  +def parse_pcb_text(text: str, filepath: str | Path = "") -> dict:+    """Same as parse_pcb, but for board text already in memory.++    The routing verbs splice new copper into the board TEXT, so they hold the+    string anyway; re-reading the file to inspect it would race their own write.+    """+    return _extract_pcb(parse(text), filepath)++ def _extract_pcb(tree: list, filepath: str | Path) -> dict:     """Extract all relevant data from a parsed PCB tree."""+    data = _extract_pcb_raw(tree, filepath)+    return _resolve_net_format(data)+++def _resolve_net_format(data: dict) -> dict:+    """Say which net dialect the board speaks, and give it a net list either way.++    A KiCad 10 board has no net table to read, so the only inventory of its nets+    is the names its pads and copper mention. Callers that write copper need+    this: the (net ...) they emit has to match the shape of the file.+    """+    numbered = bool(data.get("nets")) or any(+        p.get("net_number") is not None+        for fp in data.get("footprints", []) for p in fp.get("pads", []))+    data["net_format"] = "number" if numbered else "name"+    if not data.get("nets"):+        names, seen = [], set()+        for name in ([p.get("net_name") for fp in data.get("footprints", [])+                      for p in fp.get("pads", [])]+                     + [s.get("net_name") for s in data.get("segments", [])]+                     + [v.get("net_name") for v in data.get("vias", [])]+                     + [z.get("net_name") for z in data.get("zones", [])]):+            if name and name not in seen:+                seen.add(name)+                names.append(name)+        data["nets"] = [{"number": None, "name": n} for n in names]+    return data+++def _extract_pcb_raw(tree: list, filepath: str | Path) -> dict:     return {         "file": str(filepath),         "version": node_value(find_node(tree, "version")),@@ -85,17 +143,12 @@ def _extract_layers(tree: list) -> list[dict]:   def _extract_nets(tree: list) -> list[dict]:-    """Extract net definitions."""+    """Net definitions from the board's own table (KiCad <=9 only)."""     nets = []     for node in find_nodes(tree, "net"):-        if len(node) >= 3:-            try:-                nets.append({-                    "number": int(node[1]),-                    "name": node[2] if isinstance(node[2], str) else "",-                })-            except (ValueError, TypeError):-                pass+        num, name = _net_ref(node, int(node_value(find_node(tree, "version")) or 0) >= 20260206)+        if num is not None:+            nets.append({"number": num, "name": name or ""})     return nets  @@ -192,13 +245,13 @@ def _extract_footprints(tree: list) -> list[dict]:                         pad_info["height"] = float(size_node[2])                     except (ValueError, TypeError):                         pass-                net_node = find_node(pad, "net")-                if net_node and len(net_node) >= 3:-                    try:-                        pad_info["net_number"] = int(net_node[1])-                        pad_info["net_name"] = net_node[2] if isinstance(net_node[2], str) else ""-                    except (ValueError, TypeError):-                        pass+                pad_layers = find_node(pad, "layers")+                if pad_layers:+                    pad_info["layers"] = [l for l in pad_layers[1:] if isinstance(l, str)]+                num, name = _net_ref(find_node(pad, "net"), int(node_value(find_node(tree, "version")) or 0) >= 20260206)+                if num is not None or name is not None:+                    pad_info["net_number"] = num+                    pad_info["net_name"] = name or ""                 pad_list.append(pad_info)         fp["pads"] = pad_list @@ -232,12 +285,11 @@ def _extract_segments(tree: list) -> list[dict]:         layer = find_node(node, "layer")         if layer:             seg["layer"] = node_value(layer)-        net = find_node(node, "net")-        if net:-            try:-                seg["net"] = int(node_value(net))-            except (ValueError, TypeError):-                pass+        num, name = _net_ref(find_node(node, "net"), int(node_value(find_node(tree, "version")) or 0) >= 20260206)+        if num is not None:+            seg["net"] = num+        if name is not None:+            seg["net_name"] = name         segments.append(seg)     return segments @@ -274,12 +326,11 @@ def _extract_vias(tree: list) -> list[dict]:         layers = find_node(node, "layers")         if layers:             via["layers"] = [l for l in layers[1:] if isinstance(l, str)]-        net = find_node(node, "net")-        if net:-            try:-                via["net"] = int(node_value(net))-            except (ValueError, TypeError):-                pass+        num, name = _net_ref(find_node(node, "net"), int(node_value(find_node(tree, "version")) or 0) >= 20260206)+        if num is not None:+            via["net"] = num+        if name is not None:+            via["net_name"] = name         vias.append(via)     return vias @@ -289,12 +340,11 @@ def _extract_zones(tree: list) -> list[dict]:     zones = []     for node in find_nodes(tree, "zone"):         zone = {"uuid": node_value(find_node(node, "uuid"))}-        net = find_node(node, "net")-        if net:-            try:-                zone["net"] = int(node_value(net))-            except (ValueError, TypeError):-                pass+        num, name = _net_ref(find_node(node, "net"), int(node_value(find_node(tree, "version")) or 0) >= 20260206)+        if num is not None:+            zone["net"] = num+        if name is not None:+            zone["net_name"] = name         net_name = find_node(node, "net_name")         if net_name:             zone["net_name"] = node_value(net_name)

@@ -0,0 +1,109 @@+import copy+from pathlib import Path+import sys+import tempfile+import unittest+from unittest.mock import patch++sys.path.insert(0, str(Path(__file__).resolve().parents[1]))+from handlers import live_routing as live, route+++def board():+    return {'net_format':'name', 'nets':[{'name':'GND'}, {'name':'VCC'}, {'name':'123'}],+        'layers':[{'name':s} for s in ['F.Cu','In1.Cu','B.Cu']],+        'footprints':[{'reference':'J1','layer':'F.Cu','position':{'x':10,'y':20,'rotation':90},+            'pads':[{'number':'1','x':2,'y':0,'net_name':'GND','layers':['*.Cu'],'type':'thru_hole'},+                    {'number':'2','x':0,'y':0,'net_name':'VCC','layers':['F.Cu'],'type':'smd'}]},+            {'reference':'J2','layer':'B.Cu','position':{'x':30,'y':20},+             'pads':[{'number':'1','x':0,'y':0,'net_name':'GND','layers':['B.Cu'],'type':'smd'}]}]}+++class Planning(unittest.TestCase):+    def test_rotated_pad_coordinates_and_final_layer_transition(self):+        p=live._plan(board(), {'net':'GND','points':['J1.1',{'pad':'J2.1','layer':'B.Cu'}]})+        self.assertEqual(p['segments'][0]['start'],[10,18])+        self.assertEqual(p['vias'][0]['at'],[30,20])+        self.assertEqual(p['vias'][0]['layers'],['F.Cu','B.Cu'])++    def test_branches_on_one_net(self):+        p=live._plan(board(), {'net':'GND','paths':[['J1.1',[20,18]],[[20,18],[20,25]]]})+        self.assertEqual(len(p['segments']),2)++    def test_reject_wrong_net_and_layer(self):+        for args,code in [({'net':'GND','points':['J1.1','J1.2']},'net_mismatch'),+                          ({'net':'GND','points':['J1.1','J2.1']},'pad_layer_mismatch'),+                          ({'net':'GND','points':['J1.1',{'x':20,'y':20,'layer':'F.SilkS'}]},'unknown_layer')]:+            with self.subTest(code=code),self.assertRaises(live.RoutingError) as cm: live._plan(board(),args)+            self.assertEqual(cm.exception.code,code)++    def test_invalid_geometry_is_rejected_before_mutation(self):+        for args in [{'width':0},{'width':-1},{'width':float('nan')},{'width':True},{'points':[[True,0],[1,1]]},{'points':[[0,0,9],[1,1]]},+                     {'viaDrill':0.8,'viaSize':0.6},{'points':[[0,0],[float('inf'),0]]},+                     {'points':[[0,0],[0,0]]}]:+            with self.subTest(args=args),self.assertRaises(live.RoutingError):+                live._plan(board(),{'net':'GND','points':[[0,0],[1,1]],**args})++    def test_numeric_net_name_in_kicad_10_file(self):+        from parsers.pcb import parse_pcb_text+        data=parse_pcb_text('(kicad_pcb (version 20260206) (footprint "X" (layer "F.Cu") (pad "1" thru_hole circle (net "123"))))')+        self.assertEqual(data["net_format"],"name")+        self.assertEqual(data["nets"],[{"number":None,"name":"123"}])++    def test_exact_numeric_net_name(self):+        self.assertEqual(live._plan(board(),{'net':'123','points':[[0,0],[1,1]]})['netName'],'123')++    def test_quoted_net_serialization(self):+        self.assertIn('"a\\"b\\\\c"',route._net_sexpr({'format':'name','name':'a"b\\c'}))++    def test_stale_revision(self):+        for rev in [None,'old']:+            with self.assertRaises(live.RoutingError): live._revision({'expectedRevision':rev},'new')++    def test_existing_error_does_not_mask_additional_same_error(self):+        v={'type':'clearance','severity':'error','description':'collision','items':[]}+        self.assertEqual(live._new_errors({'violations':[v]},{'violations':[v,v]}),[v])++    def test_open_file_refused_even_with_private_arg(self):+        with patch.object(route,'_lock_file',return_value=Path('lock')):+            self.assertEqual(route._load({'filePath':'test.kicad_pcb','_readOnly':True})[-1]['errorCode'],'board_open')+++class Transactions(unittest.TestCase):+    class Board:+        def __init__(self): self.ops=[]+        def begin_commit(self): self.ops.append('begin'); return 'commit'+        def push_commit(self,c,m): self.ops.append('push')+        def drop_commit(self,c): self.ops.append('drop')++    def test_success_is_one_undo_step(self):+        b=self.Board()+        self.assertEqual(live._transaction(b,lambda:42,'route'),42)+        self.assertEqual(b.ops,['begin','push'])++    def test_partial_create_failure_rolls_back(self):+        b=self.Board()+        def fail(): raise RuntimeError('partial create')+        with self.assertRaises(live.RoutingError) as cm: live._transaction(b,fail,'route')+        self.assertEqual(cm.exception.code,'mutation_rolled_back')+        self.assertEqual(b.ops,['begin','drop'])++    def test_push_timeout_is_unknown_even_if_drop_acknowledged(self):+        b = self.Board()+        def push(*args): raise TimeoutError('commit response lost')+        b.push_commit = push+        with self.assertRaises(live.RoutingError) as cm:+            live._transaction(b, lambda: 42, 'route')+        self.assertEqual(cm.exception.code, 'mutation_outcome_unknown')+        self.assertIsNone(cm.exception.detail['mutated'])++    def test_failed_rollback_is_unknown_not_safe_to_retry(self):+        b=self.Board()+        def fail(*a): raise RuntimeError('lost IPC')+        b.drop_commit=fail+        with self.assertRaises(live.RoutingError) as cm: live._transaction(b,fail,'route')+        self.assertEqual(cm.exception.code,'mutation_outcome_unknown')+        self.assertIsNone(cm.exception.detail['mutated'])+++if __name__=='__main__': unittest.main()

@@ -0,0 +1,134 @@+"""Exercise public handlers with real kipy copper types and a simulated editor.++IPC transport and DRC execution are mocked here; Windows live tests remain required.+"""+from contextlib import contextmanager+from pathlib import Path+from types import SimpleNamespace+import tempfile+import unittest+from unittest.mock import patch+import sys+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))+from handlers import live_routing as live, route+from demo.routing.make_fixture import make+try:+    from kipy.board_types import Net+except ImportError:+    Net = None++[email protected](Net is None, 'Install requirements-routing.txt for protocol tests')+class IPCHandlers(unittest.TestCase):+    def setUp(self):+        self.tmp = tempfile.TemporaryDirectory()+        self.path = make(self.tmp.name)+        text = self.path.read_text()+        self.args = {'filePath': str(self.path), 'net': 'NET_1', 'points': ['J1.1', 'J2.1'], 'validate': False}+        class Board:+            def __init__(b):+                b.text = text; b.items = []; b.ops = []; b.partial = False; b.fail_read = False; b.fail_save = False; b.keep_removed = False+            def get_as_string(b):+                if b.fail_read: raise RuntimeError('read failed after commit')+                return b.text+            def get_nets(b):+                n = Net(); n.name = 'NET_1'; return [n]+            def begin_commit(b): b.ops.append('begin'); return 'commit'+            def create_items(b, items):+                for i, item in enumerate(items): item.id.value = 'created-'+str(i)+                b.items += items+                return [] if b.partial else items+            def push_commit(b, *_): b.ops.append('push'); b.text += '\n'+            def drop_commit(b, *_): b.ops.append('drop'); b.items = []+            def save(b):+                if b.fail_save: raise RuntimeError('save failed')+                b.ops.append('save')+            def get_tracks(b): return b.items+            def get_vias(b): return []+            def remove_items(b, items):+                if not b.keep_removed: b.items = []+        self.board = Board()+        @contextmanager+        def connect(args): yield self.board+        self.patch = patch.object(live, '_connect', connect); self.patch.start()+        self.args['expectedRevision'] = live._snapshot(self.board, self.args)[2]++    def tearDown(self): self.patch.stop(); self.tmp.cleanup()++    def test_copper_protocol_units_net_and_layer(self):+        result = live.handle_route_net({}, self.args)+        self.assertTrue(result['success'], result)+        self.assertEqual(self.board.items[0].width, 250000)+        self.assertEqual(self.board.items[0].start.x, 105000000)+        self.assertEqual(self.board.items[0].net.name, 'NET_1')+        self.assertEqual(self.board.ops, ['begin', 'push'])+        self.assertNotEqual(result['revision'], self.args['expectedRevision'])+        self.assertFalse(result['saved'])++    def test_dry_run_does_not_start_transaction(self):+        result = live.handle_route_net({}, {**self.args, 'dryRun': True})+        self.assertTrue(result['success'], result)+        self.assertEqual(self.board.ops, [])++    def test_drc_rejection_and_edit_during_preflight_do_not_mutate(self):+        ok = {'violations': []}+        bad = {'violations': [{'severity': 'error', 'type': 'clearance'}]}+        with patch.object(live, '_drc', side_effect=[ok, bad]):+            r = live.handle_route_net({}, {**self.args, 'validate': True})+        self.assertEqual(r['errorCode'], 'drc_rejected')+        def drc(*a): self.board.text += '\n'; return ok+        with patch.object(live, '_drc', side_effect=drc):+            r = live.handle_route_net({}, {**self.args, 'validate': True})+        self.assertEqual(r['errorCode'], 'stale_board')+        self.assertEqual(self.board.ops, [])++    def test_partial_create_is_rolled_back(self):+        self.board.partial = True+        r = live.handle_route_net({}, self.args)+        self.assertEqual(r['errorCode'], 'mutation_rolled_back')+        self.assertEqual(self.board.ops, ['begin', 'drop'])+        self.assertFalse(self.board.items)++    def test_post_commit_save_failure_is_not_failed_mutation(self):+        self.board.fail_save = True+        r = live.handle_route_net({}, {**self.args, 'save': True})+        self.assertTrue(r['success'], r); self.assertTrue(r['mutated'])+        self.assertFalse(r['saved']); self.assertIn('save failed', r['postCommitError'])+        self.assertEqual(self.board.ops, ['begin', 'push'])++    def test_replay_rejected(self):+        self.assertTrue(live.handle_route_net({}, self.args)['success'])+        self.assertEqual(live.handle_route_net({}, self.args)['errorCode'], 'stale_board')+        self.assertEqual(self.board.ops, ['begin', 'push'])++    def test_remove_readback_detects_ignored_deletion(self):+        r = live.handle_route_net({}, self.args)+        self.board.keep_removed = True+        r = live.handle_remove_route({}, {**self.args, 'expectedRevision': r['revision'], 'itemIds': r['itemIds']})+        self.assertEqual(r['errorCode'], 'mutation_rolled_back')+        self.assertEqual(self.board.ops[-2:], ['begin', 'drop'])++    def test_wrong_or_locked_item_never_deleted(self):+        r = live.handle_remove_route({}, {**self.args, 'itemIds': ['not-copper']})+        self.assertEqual(r['errorCode'], 'invalid_item_ids')+        self.assertEqual(self.board.ops, [])+++class OfflineTests(unittest.TestCase):+    def test_numeric_net_name_and_fractional_code(self):+        b = {'nets': [{'number': 1, 'name': '123'}], 'net_format': 'number'}+        self.assertEqual(route._resolve_net(b, '123')[0]['number'], 1)+        for spec in [1.5, float('nan'), True, '123 '.strip()+'x']:+            self.assertIsNotNone(route._resolve_net(b, spec)[1])++    def test_atomic_file_edit_and_open_board_refusal(self):+        with tempfile.TemporaryDirectory() as d:+            path = make(d); before = path.read_bytes()+            r = route.handle_route({}, {'filePath':str(path),'net':'NET_1','points':['J1.1','J2.1']})+            self.assertTrue(r['success'], r)+            self.assertEqual(Path(str(path)+'.adom-bak').read_bytes(), before)+            path.with_name('~'+path.name+'.lck').write_text('open')+            after = path.read_bytes()+            r = route.handle_add_track({}, {'filePath':str(path),'net':'NET_1','start':[1,1],'end':[2,2]})+            self.assertEqual(r['errorCode'], 'board_open')+            self.assertEqual(path.read_bytes(), after)

@@ -0,0 +1,91 @@+"""Apply the routing development overlay to an existing PINNED bridge cache.++Run with the bridge stopped: python install_routing_dev.py CACHE OVERLAY.zip.+Only the routing files and registration entries change; unrelated newer bridge+fixes stay installed. A timestamped backup beside the cache permits restoration.+"""+import ast+from datetime import datetime, timezone+import json+from pathlib import Path+import shutil+import sys+import zipfile++FILES = ['handlers/live_routing.py', 'handlers/route.py', 'parsers/pcb.py', 'requirements-routing.txt']+NAMES = ['routing_state', 'route_net', 'remove_route', 'routing_validate', 'board_pads', 'add_track', 'add_via', 'route']+++def after_assignment(source, name, addition):+    node = next(n for n in ast.walk(ast.parse(source)) if isinstance(n, ast.Assign)+                and any(isinstance(t, ast.Name) and t.id == name for t in n.targets))+    lines = source.splitlines(keepends=True)+    indentation = ' ' * node.col_offset+    lines.insert(node.end_lineno, '\n' + '\n'.join(indentation + line for line in addition.splitlines()) + '\n')+    return ''.join(lines)+++def install(cache, overlay):+    cache = Path(cache)+    if not (cache / '.dev-pin').is_file():+        raise ValueError('Refusing: use ab bridge_dev_mode {name:kicad,on:true} first')+    # Read and compile the entire overlay before touching the running cache.+    with zipfile.ZipFile(overlay) as z:+        contents = {name: z.read(name) for name in FILES}+        catalog = json.loads(z.read('routing-catalog.json'))+    if set(catalog) != set(NAMES):+        raise ValueError('Incomplete routing verb catalog')+    for name, data in contents.items():+        if name.endswith('.py'):+            compile(data, name, 'exec')+    source = (cache / 'server.py').read_text(encoding='utf-8')+    # Remove only our delimited registrations so reinstallation updates help too.+    import re+    source = re.sub(r'\n[ \t]*# ADOM_LIVE_ROUTING_DEV_BEGIN.*?# ADOM_LIVE_ROUTING_DEV_END\n', '\n', source, flags=re.S)+    if '# ADOM_LIVE_ROUTING_DEV' in source:+        raise ValueError('Old development overlay found; restore server.py from its backup before upgrading')+    def marked(text):+        return '# ADOM_LIVE_ROUTING_DEV_BEGIN\n' + text + '\n# ADOM_LIVE_ROUTING_DEV_END'+    source = after_assignment(source, '_VERB_CATALOG', marked('_VERB_CATALOG.update(' + repr(catalog) + ')'))+    addition = ('from handlers.live_routing import handle_routing_state, handle_route_net, handle_remove_route, handle_routing_validate\n'+                'from handlers.route import handle_board_pads, handle_add_track, handle_add_via, handle_route\n'+                'COMMAND_HANDLERS.update({' + ', '.join(repr(n) + ': handle_' + n for n in NAMES) + '})')+    source = after_assignment(source, 'COMMAND_HANDLERS', marked(addition))+    source = after_assignment(source, '_NO_GUI', marked('_NO_GUI.update(' + repr(set(NAMES)) + ')'))+    compile(source, str(cache / 'server.py'), 'exec')+    contents['server.py'] = source.encode('utf-8')+    manifest = json.loads((cache / 'bridge.json').read_text())+    for name in NAMES:+        if 'kicad_' + name not in manifest['verbs']:+            manifest['verbs'].append('kicad_' + name)+    manifest['version'] = '0.10.0-dev.per-net.1'+    contents['bridge.json'] = (json.dumps(manifest, indent=2) + '\n').encode()+    contents['BRIDGE_VERSION'] = manifest['version'].encode()+    backup = cache.parent / ('kicad-routing-backup-' + datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%f'))+    backup.mkdir()+    originals = {}+    for name in contents:+        original = cache / name+        originals[name] = original.read_bytes() if original.is_file() else None+        if originals[name] is not None:+            target = backup / name+            target.parent.mkdir(parents=True, exist_ok=True)+            target.write_bytes(originals[name])+    try:+        for name, data in contents.items():+            target = cache / name+            target.parent.mkdir(parents=True, exist_ok=True)+            target.write_bytes(data)+    except Exception:+        for name, data in originals.items():+            if data is None:+                (cache / name).unlink(missing_ok=True)+            else:+                (cache / name).write_bytes(data)+        raise+    result = {'installed': str(cache), 'backup': str(backup), 'version': manifest['version']}+    print(json.dumps(result))+    return result++if __name__ == '__main__':+    install(*sys.argv[1:])