KiCad - the KiCad Bridge

"""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 _report_limits(violations, unconnected):
    # KiCad 10.0.3 DRC_ENGINE: 199 per type, extended to 499 for
    # clearance and unconnected items. At the boundary completeness is unknown.
    counts = Counter(v.get("type") for v in violations)
    capped = sorted(str(kind) for kind, count in counts.items()
                    if count >= (499 if kind == "clearance" else 199))
    return {"unconnectedReportMayBeTruncated": len(unconnected) >= 499,
            "unconnectedCountIsLowerBound": len(unconnected) >= 499,
            "violationReportMayBeTruncated": bool(capped),
            "possiblyTruncatedViolationTypes": capped}


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"
        # Copper edits invalidate cached zone fills. Refill the disposable
        # snapshot so connectivity reflects the proposed copper, not old pours.
        proc = subprocess.run([exe, "pcb", "drc", "--format", "json", "--severity-all",
                               "--all-track-errors", "--refill-zones", "--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,
            "zonesRefilled": True, **_report_limits(violations, unconnected)}


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):
    if any(r.get("violationReportMayBeTruncated") for r in (before, after)):
        raise RoutingError("drc_incomplete", "DRC violation lists may be capped; cannot certify the error delta. No live copper was changed",
                           mutated=False)
    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)