app
KiCad - the KiCad Bridge
Public Made by Adomby adom
Reference implementation of the KiCad bridge: multi-instance Python server, forward path via kicad-cli, reverse path via in-process plugin. Most complex of the three bundled bridges.
← Commit history
0.9.340: Live routing verbs through KiCad's IPC API (routing_state, route_net, remove_route, routing_validate), offline copper verbs (board_pads, add_track, add_via, route), kicad_ipc_api, KiCad 10 name-based nets, vendored routing deps
4 files changed
+1113
handlers/ipc_api.pyadded+72@@ -0,0 +1,72 @@+"""kicad_ipc_api: read or flip KiCad's IPC API server switch (Preferences > Plugins).++The live routing verbs need `api.enable_server` in kicad_common.json. KiCad reads+that file at start, so a change needs KiCad closed and relaunched; this verb+never restarts KiCad itself and says so in needsRestart.+"""+from __future__ import annotations++import json+import os+import shutil+import sys+from datetime import datetime, timezone+from pathlib import Path+++def _settings_roots() -> list[Path]:+ if sys.platform == "win32":+ base = os.environ.get("APPDATA")+ return [Path(base) / "kicad"] if base else []+ if sys.platform == "darwin":+ return [Path.home() / "Library" / "Preferences" / "kicad"]+ return [Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "kicad"]+++def _common_files() -> list[Path]:+ found = []+ for root in _settings_roots():+ if root.is_dir():+ found += sorted(root.glob("*/kicad_common.json"))+ def key(p):+ try:+ return tuple(int(x) for x in p.parent.name.split("."))+ except ValueError:+ return (0,)+ return sorted(found, key=key, reverse=True)+++def handle_ipc_api(info: dict, args: dict) -> dict:+ files = _common_files()+ if not files:+ return {"success": False, "errorCode": "settings_not_found",+ "error": "No kicad_common.json found; start KiCad once so it writes its settings",+ "_hint": "KiCad creates <settings>/<version>/kicad_common.json on first launch."}+ target = files[0]+ try:+ settings = json.loads(target.read_text(encoding="utf-8"))+ except Exception as exc:+ return {"success": False, "errorCode": "settings_unreadable", "error": f"{target}: {exc}"}+ api = settings.get("api") if isinstance(settings.get("api"), dict) else {}+ before = bool(api.get("enable_server"))+ enable = args.get("enable")+ result = {"success": True, "file": str(target), "kicadSettingsVersion": target.parent.name,+ "enabled": before, "changed": False, "needsRestart": False}+ if enable is not None and not isinstance(enable, bool):+ return {"success": False, "errorCode": "invalid_arg", "error": "enable must be true or false"}+ if enable is None or enable == before:+ result["_hint"] = ("IPC API server is " + ("enabled" if before else "disabled") ++ ". Pass enable:true|false to change it (KiCad must then be restarted).")+ return result+ backup = target.with_name(target.name + ".adom-bak")+ shutil.copy2(target, backup)+ settings.setdefault("api", {})["enable_server"] = enable+ tmp = target.with_name(target.name + ".adom-tmp")+ tmp.write_text(json.dumps(settings, indent=2) + "\n", encoding="utf-8")+ os.replace(tmp, target)+ result.update(enabled=enable, changed=True, needsRestart=True, backup=str(backup),+ changedAt=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),+ _hint="Written to kicad_common.json. KiCad reads it at start: kicad_close, then launch "+ "again, then kicad_routing_state will connect. A running KiCad that saves its "+ "preferences on exit overwrites this change, so close KiCad first when in doubt.")+ return result
handlers/live_routing.pyadded+422@@ -0,0 +1,422 @@+"""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()+# Vendored IPC client (see routing_deps/README.txt): one tree per CPython ABI,+# appended (not prepended) so an environment that already has kipy keeps its own.+_DEPS = Path(__file__).resolve().parents[1] / "routing_deps" / f"cp{sys.version_info.major}{sys.version_info.minor}"+if _DEPS.is_dir() and str(_DEPS) not in sys.path:+ sys.path.append(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 _delete_items(board, items):+ # kipy 0.8 discards DeleteItemsResponse. Validate every result before+ # pushing: pending deletions remain visible to GetItems until commit.+ from kipy.proto.common.commands.editor_commands_pb2 import DeleteItems, DeleteItemsResponse, IDS_OK+ from kipy.proto.common.types.base_types_pb2 import IRS_OK+ request = DeleteItems()+ request.header.document.CopyFrom(board._doc)+ request.item_ids.extend(i.id for i in items)+ response = board._kicad.send(request, DeleteItemsResponse)+ expected = {i.id.value for i in items}+ received = {r.id.value for r in response.deleted_items if r.status == IDS_OK}+ if response.status != IRS_OK:+ raise RuntimeError("KiCad rejected the copper deletion request")+ # KiCad 10.0.5 returns an empty per-item list (upstream MR !2583).+ # We precheck all IDs/locks and verify the board after push in that case.+ if response.deleted_items and (len(response.deleted_items) != len(items) or received != expected):+ raise RuntimeError("KiCad did not accept every requested copper deletion")+++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])+ _transaction(board, lambda: _delete_items(board, items), "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:+ _, after, result["revision"] = _snapshot(board, args)+ remaining = wanted & {i.get("uuid") for key in ("segments", "vias") for i in after[key]}+ if remaining:+ result["removedIds"] = sorted(wanted - remaining)+ result["postCommitError"] = "Some requested copper remains after commit; inspect the live board"+ result["verified"] = not remaining+ 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)
handlers/route.pyadded+618@@ -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."})
requirements-routing.txtadded+1@@ -0,0 +1 @@+kicad-python==0.8.0