KiCad - the KiCad Bridge

#!/usr/bin/env python3
"""Run every kicad-bridge edit verb for real, on scratch copies, and confirm each change with a separate read.

The battery (run_verb_tests.py) proves a verb answers. Many of its edit checks stop at success:true or a dry run,
so a verb that answers and changes nothing, or changes the wrong thing, passes. This drill runs the edit, then asks
a different verb, or KiCad itself through kicad-cli's netlist, whether the change happened.

  python3 edit_drill.py --target DESKTOP-BAR6S0F [--part sheet,library,file,live] [--out drill.json]

Exit 1 when any step fails. Files live under Documents/adom-kicad-drill/<stamp> on the target. The live part opens
KiCad, so take the laptop in turn; the other parts touch only files.
"""
import argparse
import json
import pathlib
import subprocess
import sys
import tempfile
import time

AI_THREAD = "kicad-lane"
REPO = pathlib.Path(__file__).resolve().parents[2]
ROUTE_FIXTURE = REPO / "rust/crates/kicad-core/tests/fixtures/konnect-route"
LIVE_BOARD = REPO / "demo/placement/esc-g431-unrouted.kicad_pcb"


class Drill:
    def __init__(self, target):
        self.target = target
        self.results = []

    def call(self, verb, args=None, timeout=240):
        cmd = ["adom-bridge", "--ai-thread", AI_THREAD, "--target", self.target, verb if verb.startswith(("kicad_", "send_", "desktop_")) else "kicad_" + verb, json.dumps(args or {})]
        try:
            p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
            return json.loads(p.stdout.strip() or p.stderr.strip())
        except subprocess.TimeoutExpired:
            return {"success": False, "error": f"no reply in {timeout}s"}
        except Exception as e:
            return {"success": False, "error": f"unreadable reply: {e}"}

    def check(self, verb, what, reply, good, seen=None):
        """Record one step: the edit's reply and what the separate read saw."""
        passed = bool(good)
        self.results.append({"verb": verb, "what": what, "pass": passed, "seen": seen, "reply": None if passed else {k: reply.get(k) for k in ("success", "errorCode", "error", "_hint") if k in reply}})
        print(f"  {'PASS' if passed else 'FAIL'}  {verb:32} {what}" + ("" if passed else f"\n        seen: {json.dumps(seen)[:300]}\n        reply: {json.dumps(reply.get('error') or reply.get('errorCode'))[:300]}"), flush=True)
        return passed


def near(a, b, tol=0.02):
    if a is None or b is None or len(a) != len(b) or None in a or None in b:
        return False
    return all(abs(x - y) <= tol for x, y in zip(a, b))


def same_net(nets, *pins):
    return any(all(p in members for p in pins) for members in nets.values())


def sheet_part(d, base):
    print("[sheet] sch_* verbs on a new project")
    pd = base + "/sheet/dr"
    pro, sch = pd + "/dr.kicad_pro", pd + "/dr.kicad_sch"
    r = d.call("project_create", {"parentDir": base + "/sheet", "name": "dr", "layers": 2})
    if not d.check("project_create", "the project's sheet reads", r, d.call("sch_items", {"filePath": sch}).get("success") is True):
        return

    def nets():
        return d.call("netlist", {"filePath": sch}, 120).get("nets") or {}

    def items(*kinds):
        return d.call("sch_items", {"filePath": sch, "kinds": list(kinds)})

    def comp(ref):
        for c in d.call("sch_components", {"filePath": sch}).get("components") or []:
            if c.get("reference") == ref:
                return c
        return {}

    def field(c, name):
        return next((f.get("value") for f in c.get("fields") or [] if f.get("name") == name), None)

    r = d.call("sch_page", {"filePath": sch, "paper": "A3", "title": "drill title"})
    pg = d.call("sch_page", {"filePath": sch})
    d.check("sch_page", "paper A3 reads back", r, (pg.get("page") or {}).get("paper") == "A3", pg.get("page"))

    # The anchors sit on the grid. Device:R's pins are 3.81 mm above and below its anchor.
    for ref, x in (("R1", 50.8), ("R2", 76.2)):
        r = d.call("sch_place_symbol", {"filePath": sch, "libId": "Device:R", "reference": ref, "at": [x, 50.8]})
        c = comp(ref)
        d.check("sch_place_symbol", f"{ref} is at its point", r, near((c.get("units") or [{}])[0].get("at"), [x, 50.8]), c.get("units"))

    r = d.call("sch_connect_pins", {"filePath": sch, "from": "R1.2", "to": "R2.2"})
    n = nets()
    d.check("sch_connect_pins", "KiCad's netlist joins R1.2 and R2.2", r, same_net(n, "R1.2", "R2.2"), n)

    r = d.call("sch_split_wire", {"filePath": sch, "at": [63.5, 54.61], "junction": False})
    w = items("wires").get("wires") or []
    d.check("sch_split_wire", "one wire is now two", r, len(w) == 2, w)

    r = d.call("sch_junction", {"filePath": sch, "at": [63.5, 54.61]})
    j = items("junctions").get("junctions") or []
    d.check("sch_junction", "the junction is there", r, len(j) == 1 and near(j[0].get("at"), [63.5, 54.61]), j)
    if j:
        r = d.call("sch_junction", {"filePath": sch, "uuid": j[0].get("uuid") or j[0].get("id"), "remove": True})
        j2 = items("junctions").get("junctions") or []
        d.check("sch_junction", "remove by id takes it away", r, not j2, j2)

    r = d.call("sch_delete_wire", {"filePath": sch, "at": [57.15, 54.61]})
    n = nets()
    d.check("sch_delete_wire", "the netlist no longer joins R1.2 and R2.2", r, not same_net(n, "R1.2", "R2.2"), n)

    r = d.call("sch_wire", {"filePath": sch, "points": [[50.8, 54.61], [63.5, 54.61]]})
    n = nets()
    d.check("sch_wire", "a drawn wire joins them again", r, same_net(n, "R1.2", "R2.2"), n)

    r = d.call("sch_connect_to_net", {"filePath": sch, "pin": "R1.1", "net": "VIN"})
    n = nets()
    d.check("sch_connect_to_net", "R1.1 is on VIN", r, "R1.1" in (n.get("VIN") or n.get("/VIN") or []), n)

    lab = next((x for x in items("labels").get("labels") or [] if x.get("text") == "VIN"), {})
    r = d.call("sch_edit_label", {"filePath": sch, "net": "VIN", "rename": "VIN2"})
    n = nets()
    d.check("sch_edit_label", "rename: the net is VIN2", r, "R1.1" in (n.get("VIN2") or n.get("/VIN2") or []) and "VIN" not in n and "/VIN" not in n, n)
    r = d.call("sch_edit_label", {"filePath": sch, "net": "VIN2", "by": [2.54, 0]})
    lab2 = next((x for x in items("labels").get("labels") or [] if x.get("text") == "VIN2"), {})
    d.check("sch_edit_label", "by: the label moved 2.54 mm in x", r, lab.get("at") and lab2.get("at") and abs(lab2["at"][0] - lab["at"][0] - 2.54) < 0.02, [lab.get("at"), lab2.get("at")])

    r = d.call("sch_label", {"filePath": sch, "text": "SIGX", "at": [57.15, 54.61], "kind": "local"})
    n = nets()
    d.check("sch_label", "the wire's net is named SIGX", r, same_net({k: v for k, v in n.items() if "SIGX" in k}, "R1.2"), n)
    r = d.call("sch_delete", {"filePath": sch, "label": "SIGX", "at": [57.15, 54.61]})
    n = nets()
    d.check("sch_delete", "the label is gone from the netlist", r, not any("SIGX" in k for k in n), n)

    r = d.call("sch_no_connect", {"filePath": sch, "pin": "R2.1"})
    nc = items("noConnects").get("noConnects") or []
    d.check("sch_no_connect", "a marker on R2.1's end", r, len(nc) == 1 and near(nc[0].get("at"), [76.2, 46.99]), nc)
    if nc:
        r = d.call("sch_no_connect", {"filePath": sch, "uuid": nc[0].get("uuid") or nc[0].get("id"), "remove": True})
        nc2 = items("noConnects").get("noConnects") or []
        d.check("sch_no_connect", "remove by id takes it away", r, not nc2, nc2)

    # the standard's shape of the same verbs: sheet and id for filePath and uuid, remove on sch_label
    r = d.call("sch_label", {"sheet": sch, "text": "STD1", "at": [57.15, 54.61]})
    n = nets()
    d.check("sch_label", "standard args: the net is STD1", r, any("STD1" in k for k in n), list(n)[:8])
    r = d.call("sch_label", {"sheet": sch, "text": "STD1", "at": [57.15, 54.61], "remove": True})
    n = nets()
    d.check("sch_label", "standard args: remove takes it away", r, not any("STD1" in k for k in n), list(n)[:8])
    r = d.call("sch_no_connect", {"sheet": sch, "at": [76.2, 46.99]})
    nc = items("noConnects").get("noConnects") or []
    if nc:
        r2 = d.call("sch_no_connect", {"sheet": sch, "id": nc[0].get("id"), "remove": True})
        d.check("sch_no_connect", "standard args: a point, then remove by id", r2, len(nc) == 1 and not items("noConnects").get("noConnects"), nc)
    else:
        d.check("sch_no_connect", "standard args: a marker at a point", r, False, nc)
    w0 = len(items("wires").get("wires") or [])
    r = d.call("sch_wire", {"sheet": sch, "points": [[30.48, 25.4], [35.56, 25.4]]})
    d.check("sch_wire", "standard args: one more wire", r, len(items("wires").get("wires") or []) == w0 + 1, w0)
    wid = (items("wires").get("wires") or [{}])[-1].get("id")
    d.call("sch_delete_wire", {"filePath": sch, "from": [30.48, 25.4], "to": [35.56, 25.4]})

    r = d.call("sch_text", {"filePath": sch, "text": "drill note", "at": [30.48, 30.48]})
    tx = [t for t in items("texts").get("texts") or [] if t.get("text") == "drill note"]
    d.check("sch_text", "the note is there at its point", r, len(tx) == 1 and near(tx[0].get("at"), [30.48, 30.48]), tx)
    if tx:
        r = d.call("sch_delete", {"filePath": sch, "uuid": tx[0].get("uuid") or tx[0].get("id")})
        tx2 = [t for t in items("texts").get("texts") or [] if t.get("text") == "drill note"]
        d.check("sch_delete", "delete by id takes the note away", r, not tx2, tx2)

    r = d.call("sch_bus", {"filePath": sch, "points": [[101.6, 38.1], [101.6, 76.2]], "label": "D[0..1]"})
    b = items("buses").get("buses") or []
    d.check("sch_bus", "the bus runs between its points", r, len(b) == 1, b)
    r = d.call("sch_bus_entry", {"filePath": sch, "at": [99.06, 68.58], "direction": "up-right"})
    be = items("busEntries").get("busEntries") or []
    d.check("sch_bus_entry", "the entry is there", r, len(be) == 1, be)
    r = d.call("sch_connect_to_bus", {"filePath": sch, "pins": ["R2.1"], "prefix": "D", "busLabel": "D[0..1]"})
    n = nets()
    d.check("sch_connect_to_bus", "R2.1 is on the bus member D0", r, any("D0" in k and "R2.1" in v for k, v in n.items()), n)

    before = (comp("R1").get("units") or [{}])[0].get("at")
    r = d.call("sch_move", {"filePath": sch, "reference": "R1", "by": [0, -2.54]})
    after = (comp("R1").get("units") or [{}])[0].get("at")
    d.check("sch_move", "R1 moved by the amount asked", r, before and after and near([after[0], after[1] + 2.54], before[:2]), [before, after])
    r = d.call("sch_move", {"filePath": sch, "reference": "R1", "by": [0, 2.54]})

    r = d.call("sch_rotate", {"filePath": sch, "reference": "R2", "rotation": 90})
    pr = [p for p in d.call("sch_pins", {"filePath": sch, "reference": "R2"}).get("pins") or []]
    d.check("sch_rotate", "R2's pins now lie left and right", r, sorted(p.get("direction") for p in pr) == ["left", "right"], [(p.get("number"), p.get("direction")) for p in pr])
    r = d.call("sch_rotate", {"filePath": sch, "reference": "R2", "rotation": 0})

    wires0 = items("wires").get("wires") or []
    r = d.call("sch_move_region", {"filePath": sch, "box": [40, 40, 90, 60], "by": [0, 2.54]})
    r1 = (comp("R1").get("units") or [{}])[0].get("at")
    d.check("sch_move_region", "R1 inside the box moved", r, r1 and before and near([r1[0], r1[1] - 2.54], before[:2]), [before, r1, len(wires0)])
    d.call("sch_move_region", {"filePath": sch, "box": [40, 40, 90, 65], "by": [0, -2.54]})

    r = d.call("sch_set_property", {"filePath": sch, "reference": "R1", "name": "Value", "value": "4k7"})
    d.check("sch_set_property", "R1's Value is 4k7", r, field(comp("R1"), "Value") == "4k7", comp("R1").get("fields"))

    d.call("sch_place_symbol", {"filePath": sch, "libId": "Device:R", "reference": "R?", "at": [50.8, 88.9]})
    r = d.call("sch_annotate", {"filePath": sch})
    d.check("sch_annotate", "R? became R3", r, comp("R3") and not comp("R?"), [c.get("reference") for c in d.call("sch_components", {"filePath": sch}).get("components") or []])

    r = d.call("sch_replace_symbol", {"filePath": sch, "reference": "R3", "libId": "Device:R_Small"})
    d.check("sch_replace_symbol", "R3 is now Device:R_Small", r, comp("R3").get("libId") == "Device:R_Small", comp("R3").get("libId"))

    r = d.call("sch_set_property", {"filePath": sch, "reference": "R3", "name": "Value", "value": "1k"})
    r = d.call("sch_reset_fields", {"filePath": sch, "references": ["R3"]})
    d.check("sch_reset_fields", "R3 keeps its reference and value", r, r.get("success") is True and field(comp("R3"), "Value") == "1k" and comp("R3"), comp("R3").get("fields"))

    r = d.call("sch_update_from_library", {"filePath": sch})
    d.check("sch_update_from_library", "the parts still read, same symbols", r, r.get("success") is True and comp("R3").get("libId") == "Device:R_Small", {k: r.get(k) for k in ("updated", "changed", "symbols")})

    # a wire that stops 0.2 mm short of R3's pin: fix_connectivity must join it, and the netlist must say so
    pins = {p["number"]: p for p in d.call("sch_pins", {"filePath": sch, "reference": "R3"}).get("pins") or []}
    pa = pins.get("1", {}).get("at") or [50.8, 85.09]
    d.call("sch_wire", {"filePath": sch, "points": [[pa[0] + 0.2, pa[1]], [pa[0] + 12.7, pa[1]]], "junctions": False})
    d.call("sch_connect_to_net", {"filePath": sch, "pin": "R3.2", "net": "FIXNET"})
    d.call("sch_label", {"filePath": sch, "text": "FARNET", "at": [pa[0] + 12.7, pa[1]], "kind": "local"})
    n0 = nets()
    r = d.call("sch_fix_connectivity", {"filePath": sch, "dryRun": False, "tolerance": 0.5})
    n = nets()
    d.check("sch_fix_connectivity", "the 0.2 mm gap is closed: R3.1 on FARNET", r, not any("FARNET" in k and "R3.1" in v for k, v in n0.items()) and any("FARNET" in k and "R3.1" in v for k, v in n.items()), {k: v for k, v in n.items() if "FARNET" in k or "R3" in str(v)})

    r = d.call("sch_sheet", {"filePath": sch, "action": "add", "name": "Child", "file": "child.kicad_sch", "at": [127, 30.48], "size": [25.4, 20.32], "projectPath": pro})
    h = d.call("sch_hierarchy", {"projectPath": pro})
    d.check("sch_sheet", "the hierarchy has Child", r, h.get("sheetCount") == 2 and any(str(f).endswith("child.kicad_sch") for f in h.get("files") or []), h.get("files"))
    r = d.call("sch_sheet_pins", {"filePath": sch, "sheet": "Child", "action": "add", "name": "VIO", "shape": "input", "side": "left"})
    sp = d.call("sch_sheet_pins", {"filePath": sch, "sheet": "Child", "action": "list"})
    d.check("sch_sheet_pins", "the sheet has pin VIO", r, "VIO" in json.dumps(sp), sp.get("pins"))

    had = bool(comp("R3"))
    r = d.call("sch_delete", {"filePath": sch, "reference": "R3"})
    d.check("sch_delete", "R3 was there and is gone", r, had and not comp("R3"))
    # KiCad itself must still read the finished sheet
    e = d.call("run_erc", {"filePath": sch}, 200)
    d.check("run_erc", "kicad-cli still reads the sheet", e, e.get("success") is True, {k: e.get(k) for k in ("errors", "warnings", "error")})


def library_part(d, base):
    print("[library] symbol and footprint verbs on new libraries")
    lib, pretty = base + "/lib/D.kicad_sym", base + "/lib/D.pretty"
    pins = [{"number": 1, "name": "A"}, {"number": 2, "name": "B", "type": "output"}]

    def info(name):
        return d.call("sym_info", {"libraryPath": lib, "name": name})

    def fp(name):
        return d.call("fp_info", {"libraryPath": pretty, "name": name})

    r = d.call("sym_create", {"libraryPath": lib, "name": "P", "pins": pins})
    i = info("P")
    d.check("sym_create", "two pins read back with their types", r, len(i.get("pins") or []) == 2 and any(p.get("type") == "output" for p in i.get("pins") or []), i.get("pins"))
    r = d.call("set_params", {"libraryPath": lib, "symbolName": "P", "params": {"MPN": "X1"}, "comment": "10k", "footprint": "Lib:FP"})
    f = info("P").get("fields") or {}
    d.check("set_params", "MPN, Value and Footprint read back", r, (f.get("MPN"), f.get("Value"), f.get("Footprint")) == ("X1", "10k", "Lib:FP"), f)
    r = d.call("set_visibility", {"libraryPath": lib, "symbolName": "P", "param": "MPN", "visible": True})
    d.check("set_visibility", "the verb says MPN is shown, and the symbol still reads", r, r.get("success") is True and info("P").get("success") is True, r.get("visible"))
    r = d.call("clone_symbol", {"libraryPath": lib, "from": "P", "to": "Q"})
    d.check("clone_symbol", "Q exists with P's pins", r, len(info("Q").get("pins") or []) == 2)
    r = d.call("rename_symbol", {"libraryPath": lib, "from": "Q", "to": "Q2"})
    d.check("rename_symbol", "Q2 exists, Q is gone", r, info("Q2").get("success") is True and info("Q").get("success") is not True)
    r = d.call("sym_delete", {"libraryPath": lib, "name": "Q2"})
    d.check("sym_delete", "Q2 is gone, P stays", r, info("Q2").get("success") is not True and info("P").get("success") is True)
    r = d.call("set_models", {"libraryPath": lib, "symbolName": "P", "footprint": "Lib:FP2"})
    d.check("set_models", "the footprint link was replaced", r, (info("P").get("fields") or {}).get("Footprint") == "Lib:FP2" and r.get("removed") == 1, r.get("removed"))
    r = d.call("add_models", {"libraryPath": lib, "symbolName": "P", "models": [{"name": "FP3"}], "description": "d"})
    f = info("P").get("fields") or {}
    d.check("add_models", "the filter and description read back", r, f.get("ki_fp_filters") == "FP3" and f.get("Description") == "d", f)

    pads = [{"number": 1, "size": [1, 1.5], "at": [-1, 0]}, {"number": 2, "size": [1, 1.5], "at": [1, 0]}]
    r = d.call("fp_create", {"libraryPath": pretty, "name": "F", "pads": pads, "bodySize": [3, 2]})
    i = fp("F")
    d.check("fp_create", "two pads read back at their points", r, len(i.get("pads") or []) == 2, i.get("pads"))
    r = d.call("fp_edit", {"libraryPath": pretty, "name": "F", "pads": [{"number": "1", "set": {"size": [2, 2]}}]})
    p1 = next((p for p in fp("F").get("pads") or [] if str(p.get("number")) == "1"), {})
    d.check("fp_edit", "pad 1 is now 2 by 2", r, near(p1.get("size"), [2, 2]), p1)
    r = d.call("copy_footprint", {"libraryPath": pretty, "from": "F", "to": "G"})
    d.check("copy_footprint", "G exists", r, fp("G").get("success") is True)
    r = d.call("rename_footprint", {"libraryPath": pretty, "from": "G", "to": "G2"})
    d.check("rename_footprint", "G2 exists, G is gone", r, fp("G2").get("success") is True and fp("G").get("success") is not True)
    r = d.call("xcopy_footprint", {"fromLib": pretty, "from": "G2", "toLib": base + "/lib/E.pretty", "to": "H"})
    d.check("xcopy_footprint", "H exists in the other library", r, d.call("fp_info", {"libraryPath": base + "/lib/E.pretty", "name": "H"}).get("success") is True)
    r = d.call("remove_footprint", {"libraryPath": pretty, "footprintName": "G2"})
    d.check("remove_footprint", "G2 is gone, F stays", r, fp("G2").get("success") is not True and fp("F").get("success") is True)
    r = d.call("footprint_from_land_pattern", {"libraryPath": pretty, "name": "S8", "family": "dual", "pins": 8, "pitch": 1.27, "span": 5.4, "padLength": 1.55, "padWidth": 0.6})
    d.check("footprint_from_land_pattern", "eight pads read back", r, len(fp("S8").get("pads") or []) == 8, len(fp("S8").get("pads") or []))
    # a thru-hole pad with no plating: plate_vias must plate it
    d.call("fp_create", {"libraryPath": pretty, "name": "V", "pads": [{"number": 1, "type": "thru_hole", "shape": "circle", "size": 1, "drill": 0.5, "at": [0, 0]}], "bodySize": [2, 2]})
    r = d.call("plate_vias", {"libraryPath": pretty, "footprintName": "V"})
    d.check("plate_vias", "the verb reports and the footprint still reads", r, r.get("success") is True and fp("V").get("success") is True, {k: r.get(k) for k in ("changed", "plated", "footprints", "pads")})


def file_part(d, base):
    print("[file] closed-board edits on a copy")
    bdir = base + "/board"
    src = [str(ROUTE_FIXTURE / n) for n in ("route-fixture.kicad_pcb", "route-fixture.kicad_pro", "route-fixture.net", "fp-lib-table")]
    d.call("send_files", {"filePaths": src, "dest": bdir}, 200)
    d.call("send_files", {"filePaths": [str(LIVE_BOARD)], "dest": bdir}, 200)
    d.call("send_files", {"filePaths": [str(p) for p in (ROUTE_FIXTURE / "Fixture.pretty").iterdir()], "dest": bdir + "/Fixture.pretty"}, 200)
    b = bdir + "/esc-g431-unrouted.kicad_pcb"
    rf = bdir + "/route-fixture.kicad_pcb"

    def part(path, ref):
        s = d.call("routing_state", {"filePath": path, "source": "file", "model": "standard"})
        return next((c for c in s.get("components") or [] if c.get("ref") == ref), {})

    def layers(path):
        return d.call("layers", {"filePath": path, "source": "file"})

    r = d.call("set_layer_count", {"filePath": rf, "count": 4})
    l = layers(rf)
    d.check("set_layer_count", "the board now has 4 copper layers", r, r.get("countAfter") == 4 and "In1.Cu" in json.dumps(l), {k: l.get(k) for k in ("copper", "copperLayers", "count")})

    # wiki #138: a library footprint with its reference broken over lines kept REF** and read as missing
    odd = pathlib.Path(tempfile.mkdtemp()) / "R_0603_1608Metric.kicad_mod"
    odd.write_text((ROUTE_FIXTURE / "Fixture.pretty/R_0603_1608Metric.kicad_mod").read_text().replace('(property "Reference" "REF**"', '(property "Reference"\n\t\t"REF**"'))
    d.call("send_files", {"filePaths": [str(odd)], "dest": bdir + "/Fixture.pretty"}, 200)
    r = d.call("update_pcb_from_schematic", {"projectPath": bdir + "/route-fixture.kicad_pro", "filePath": bdir + "/route-fixture.kicad_pcb", "netlistPath": bdir + "/route-fixture.net", "execute": True, "allowRemovals": True})
    p = d.call("update_pcb_from_schematic", {"filePath": bdir + "/route-fixture.kicad_pcb", "netlistPath": bdir + "/route-fixture.net"})
    d.check("update_pcb_from_schematic", "execute: a second compare says in sync, R2 has its reference", r, r.get("executed") is True and (p.get("plan") or {}).get("inSync") is True and part(bdir + "/route-fixture.kicad_pcb", "R2"), {"exec": r.get("error"), "after": (p.get("plan") or {}).get("changeCount"), "changes": p.get("changeOrder")})


def cancel_part(d, base):
    print("[cancel] stop a Freerouting job")
    bdir = base + "/cancel"
    d.call("send_files", {"filePaths": [str(LIVE_BOARD)], "dest": bdir}, 200)
    b = bdir + "/esc-g431-unrouted.kicad_pcb"
    rev0 = d.call("routing_state", {"filePath": b, "source": "file", "model": "standard"}).get("revision")
    j = d.call("autoroute", {"filePath": b, "engine": "freerouting", "job": True, "passes": 50}, 120)
    if j.get("errorCode") == "not_installed" or "not installed" in str(j.get("error")):
        print("  SKIP  cancel_run                       Freerouting is not installed here, so a real stop is NOT proven")
        return
    if j.get("success") is not True:
        d.check("autoroute", "a job started", j, False, j.get("error"))
        return
    for _ in range(20):
        p = d.call("autoroute", {"job": j["job"]})
        if "freerouting" in str(p.get("stepLabel")).lower() or p.get("status") != "running":
            break
        time.sleep(3)
    c = d.call("cancel_run", {"runId": j["job"]})
    d.check("cancel_run", "the running job takes the request", c, c.get("cancelRequested") is True and c.get("runId") == j["job"], {k: c.get(k) for k in ("cancelRequested", "stage", "reason")})
    p = {}
    for _ in range(40):
        p = d.call("autoroute", {"job": j["job"]})
        if p.get("status") != "running":
            break
        time.sleep(3)
    rev1 = d.call("routing_state", {"filePath": b, "source": "file", "model": "standard"}).get("revision")
    d.check("cancel_run", "the job ended cancelled, the board unchanged", p, p.get("errorCode") == "cancelled" and rev0 == rev1, {"status": p.get("status"), "code": p.get("errorCode"), "same": rev0 == rev1})


def prefs_part(d, base):
    print("[prefs] KiCad closed: a setting really changes, then back")
    d.call("close", {"discardChanges": True}, 120)
    key = "kicad_common.adom_drill_marker"
    r = d.call("preferences", {"set": {"kicad_common.adom_drill_marker": 41}})
    g = d.call("preferences", {"key": key})
    d.check("preferences", "set reads back", r, (g.get("current") or {}).get(key) == 41, r.get("applied"))


def live_part(d, base):
    print("[live] edits on a copy open in the PCB editor")
    ldir = base + "/live"
    d.call("send_files", {"filePaths": [str(LIVE_BOARD)], "dest": ldir}, 200)
    b = ldir + "/esc-g431-unrouted.kicad_pcb"
    def file_tracks():
        return [t for t in d.call("routing_state", {"filePath": b, "source": "file", "model": "standard"}).get("tracks") or [] if t.get("net") == "GND"]
    t0 = file_tracks()
    r = d.call("add_track", {"filePath": b, "net": "GND", "layer": "F.Cu", "width": 0.2, "start": [100, 60], "end": [104, 60]})
    t1 = file_tracks()
    d.check("add_track", "closed file: one more GND track", r, len(t1) == len(t0) + 1, [len(t0), len(t1), r.get("errorCode"), r.get("error")])
    o = d.call("open_board", {"filePath": b}, 240)
    if o.get("success") is not True:
        d.check("open_board", "the copy opened", o, False, o.get("error"))
        return
    st = {}
    for _ in range(10):
        st = d.call("routing_state", {"filePath": b})
        if st.get("revision"):
            break
        time.sleep(5)
    rev = [st.get("revision")]

    def state():
        s = d.call("routing_state", {"filePath": b, "model": "standard"})
        if s.get("revision"):
            rev[0] = s["revision"]
        return s

    def comp(ref):
        return next((c for c in state().get("components") or [] if c.get("ref") == ref), {})

    def outline():
        o = d.call("read_board", {"filePath": b, "sections": "outline"}).get("data", {}).get("outline", {})
        pts = [p for e in o.get("edges") or [] for p in (e.get("start"), e.get("end")) if p]
        return (max(p[0] for p in pts) - min(p[0] for p in pts), max(p[1] for p in pts) - min(p[1] for p in pts)) if pts else None

    def edit(verb, **args):
        return d.call(verb, dict({"filePath": b, "expectedRevision": rev[0], "validate": False}, **args))

    # edits take KiCad's frame (y down), the standard model reads y up
    M = "C7"
    c0 = comp(M)
    r = edit("move_footprint", ref=M, x=c0.get("x", 0) + 1, y=-c0.get("y", 0), allowOverlap=True, allowOutside=True)
    c1 = comp(M)
    d.check("move_footprint", f"live: {M} moved 1 mm in x", r, near([c1.get("x"), c1.get("y")], [c0.get("x", 0) + 1, c0.get("y")], 0.01), [c0.get("x"), c1.get("x"), r.get("errorCode"), r.get("error")])

    r = edit("align_footprints", refs=[M, "C12"], mode="top", by="anchor", allowOverlap=True, allowOutside=True)
    a, c = comp(M), comp("C12")
    d.check("align_footprints", f"live: {M} and C12 share a y", r, a.get("y") is not None and abs(a.get("y") - c.get("y")) < 0.01, [a.get("y"), c.get("y"), r.get("errorCode"), r.get("error")])

    pool = [c.get("ref") for c in state().get("components") or [] if str(c.get("ref")).startswith("C") and c.get("ref") not in (M, "C12")]
    ta, tb = pool[0], pool[1]
    t0 = comp(ta)
    r = edit("replicate_layout", source=[M, "C12"], targets=[[ta, tb]], anchor=M, allowOverlap=True, allowOutside=True)
    t2, t3, s0, s1 = comp(ta), comp(tb), comp(M), comp("C12")
    # the target group keeps the source's shape, turned by the difference between the anchors' rotations
    dist = lambda p, q: round(((p.get("x", 0) - q.get("x", 0)) ** 2 + (p.get("y", 0) - q.get("y", 0)) ** 2) ** 0.5, 2)
    d.check("replicate_layout", f"live: {ta} and {tb} are as far apart as {M} and C12", r, r.get("success") is True and abs(dist(s0, s1) - dist(t2, t3)) < 0.02, [dist(s0, s1), dist(t2, t3), t0.get("x"), t2.get("x"), s0.get("rotation"), t2.get("rotation"), r.get("movedRefs"), r.get("errorCode"), r.get("error")])

    before = {c.get("ref"): (c.get("x"), c.get("y")) for c in state().get("components") or []}
    r = edit("place_decoupling", ic="U5", dryRun=False, allowOverlap=True)
    after = {c.get("ref"): (c.get("x"), c.get("y")) for c in state().get("components") or []}
    moved = [k for k in before if before[k] != after.get(k)]
    d.check("place_decoupling", "live: the capacitors really moved", r, r.get("success") is True and moved, [moved[:6], r.get("errorCode"), r.get("error"), r.get("mutated")])

    r = edit("flip_footprint", ref=M)
    a = comp(M)
    d.check("flip_footprint", f"live: {M} is on the back", r, a.get("side") == "bottom", [a.get("side"), r.get("errorCode"), r.get("error")])

    r = edit("edit_footprint", ref=M, value="drillval")
    a = comp(M)
    d.check("edit_footprint", f"live: {M}'s value reads back", r, a.get("value") == "drillval", [a.get("value"), r.get("errorCode"), r.get("error")])

    def silk_fields():
        return [f for c in d.call("silk_state", {"filePath": b}).get("parts") or [] for f in c.values() if isinstance(f, dict) and f.get("id") and f.get("at")]
    fld = next((x for x in silk_fields() if x.get("layer") == "F.SilkS"), None)
    if fld:
        nx = round(fld["at"][0] + 0.5, 3)
        r = edit("silk_fields_batch", update=[{"id": fld["id"], "x": nx, "y": -fld["at"][1], "layer": "F.SilkS", "height": fld.get("heightMm") or 0.8, "stroke": fld.get("thicknessMm") or 0.12}])
        fld2 = next((x for x in silk_fields() if x["id"] == fld["id"]), {})
        moved = fld2.get("at") and abs(fld2["at"][0] - nx) < 0.01
        refused = r.get("errorCode") in ("native_field_update_unsupported", "drc_rejected") and fld2.get("at") == fld["at"]
        d.check("silk_fields_batch", "live: the field moved 0.5 mm, or a named refusal with nothing changed", r, moved or refused, [fld["at"], fld2.get("at"), r.get("errorCode"), r.get("error")])
    else:
        d.check("silk_fields_batch", "live: a field to move", {}, False, None)

    n0 = len(state().get("components") or [])
    hx, hy = c0.get("x", 0) - 20, -c0.get("y", 0) - 30
    r = edit("add_mounting_hole", x=hx, y=hy, drill=2.2)
    s1 = state()
    d.check("add_mounting_hole", "live: one more part on the board", r, len(s1.get("components") or []) == n0 + 1, [n0, len(s1.get("components") or []), r.get("errorCode"), r.get("error")])

    g0 = [t for t in state().get("tracks") or [] if t.get("net") == "GND"]
    r = edit("copy_routing", source={"minX": 99, "minY": 59, "maxX": 105, "maxY": 61}, offset={"dx": 0, "dy": 3}, keepUnmappedNets=True)
    g1 = [t for t in state().get("tracks") or [] if t.get("net") == "GND"]
    d.check("copy_routing", "live: the GND track was copied", r, len(g1) == len(g0) + 1, [len(g0), len(g1), r.get("errorCode"), r.get("error")])

    nets = [n.get("name") if isinstance(n, dict) else n for n in state().get("nets") or []]
    nets = [n for n in nets if n and n not in ("GND", "")]
    pa, pb = nets[0], nets[1]
    p0 = len([t for t in state().get("tracks") or [] if t.get("net") in (pa, pb)])
    r = edit("route_diff_pair", netP=pa, netN=pb, centerline=[[100, 70], [106, 70]], width=0.2, gap=0.2)
    p1 = len([t for t in state().get("tracks") or [] if t.get("net") in (pa, pb)])
    d.check("route_diff_pair", "live: both nets got copper", r, p1 >= p0 + 2, [pa, pb, p0, p1, r.get("errorCode"), r.get("error")])

    g1 = [t for t in state().get("tracks") or [] if t.get("net") == "GND"]
    if g1:
        tid = g1[0].get("id") or g1[0].get("uuid")
        r = d.call("edit_tracks", {"filePath": b, "expectedRevision": rev[0], "validate": False, "tracks": [{"id": tid, "width": 0.33}]})
        g2 = next((t for t in state().get("tracks") or [] if (t.get("id") or t.get("uuid")) == tid), {})
        d.check("edit_tracks", "live: the track is 0.33 wide", r, abs((g2.get("width") or 0) - 0.33) < 0.001, [g2.get("width"), r.get("errorCode"), r.get("error")])
    r = d.call("place_parts", {"filePath": b, "expectedRevision": rev[0], "parts": [{"library": "Resistor_SMD", "footprint": "R_0603_1608Metric", "ref": "R900", "x": 130, "y": 60, "rotation": 90}, {"library": "Resistor_SMD", "footprint": "R_0603_1608Metric", "ref": "R901", "x": 134, "y": 60, "side": "bottom"}], "validate": False}, 300)
    a0, a1 = comp("R900"), comp("R901")
    d.check("place_parts", "live: both parts landed, position, rotation and side", r, near([a0.get("x"), a0.get("y")], [130, -60], 0.01) and abs((a0.get("rotation") or 0) - 90) < 0.01 and a1.get("side") == "bottom", [a0.get("x"), a0.get("y"), a0.get("rotation"), a1.get("side"), r.get("results")])
    r = d.call("place_parts", {"filePath": b, "parts": [{"library": "Resistor_SMD", "footprint": "R_0603_1608Metric", "ref": "TP", "x": 138, "y": 60}], "validate": False}, 300)
    got = (r.get("placed") or [{}])[0]
    d.check("place_parts", "live: a reference KiCad numbers itself reads back", r, str(got.get("ref")).startswith("TP") and near([got.get("x"), got.get("y")], [138, -60], 0.01), r.get("placed"))
    r = d.call("select_items", {"filePath": b, "refs": ["R900", "NOPE9"]})
    g = d.call("get_selection", {"filePath": b})
    d.check("select_items", "live: R900 selected, NOPE9 named as failed", r, "R900" in (g.get("references") or []) and (r.get("failed") or [{}])[0].get("ref") == "NOPE9" and r.get("verified") is False, [g.get("references"), r.get("failed"), r.get("verified")])
    r = d.call("select_items", {"filePath": b, "clear": True})
    g = d.call("get_selection", {"filePath": b})
    d.check("select_items", "live: clear empties the selection", r, not g.get("references"), g.get("references"))
    r = d.call("select_items", {"filePath": b, "where": "sheet", "refs": ["R1"]})
    d.check("select_items", "where:sheet is refused by name", r, r.get("errorCode") == "not_supported", r.get("errorCode"))
    r = d.call("model_check", {"filePath": b})
    d.check("model_check", "per part: found and geometry", r, r.get("checked", 0) > 3 and all("found" in m and "hasGeometry" in m for q in r.get("parts") or [] for m in q.get("models") or []) and any(q.get("models") for q in r.get("parts") or []), {"checked": r.get("checked"), "ok": r.get("ok"), "error": r.get("error")})
    state()
    r = edit("ripup_nets", nets=["GND"], dryRun=False)
    t2 = [t for t in state().get("tracks") or [] if t.get("net") == "GND"]
    d.check("ripup_nets", "live: GND has no tracks left", r, not t2 and len(g1) > 0, [len(g1), len(t2), r.get("errorCode"), r.get("error")])

    r = edit("board_outline", shape="rectangle", width=90, height=80, x=90, y=50, replace=True)
    ol = outline()
    d.check("board_outline", "live: the outline is 90 by 80", r, ol and near(list(ol), [90, 80], 0.1), [ol, r.get("errorCode"), r.get("error")])

    r = d.call("save_board", {"filePath": b})
    fs = d.call("routing_state", {"filePath": b, "source": "file", "model": "standard"})
    cf = next((c for c in fs.get("components") or [] if c.get("ref") == M), {})
    d.check("save_board", f"the file holds the live edits ({M} on the back)", r, cf.get("side") == "bottom" and cf.get("value") == "drillval", [cf.get("side"), cf.get("value"), r.get("error")])

    r = d.call("teardrops", {"filePath": b})
    d.check("teardrops", "the reply counts, and the board saved", r, r.get("success") is True and r.get("saved") is True and isinstance(r.get("teardropsAfter"), int), {k: r.get(k) for k in ("teardropsBefore", "teardropsAfter", "saved", "errorCode", "error")})

    d.call("close", {"discardChanges": True}, 120)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--target", required=True)
    ap.add_argument("--part", default="sheet,library,file,cancel,live")
    ap.add_argument("--out")
    a = ap.parse_args()
    d = Drill(a.target)
    diag = d.call("readiness", {})
    user = None
    for inst in diag.get("installs") or []:
        parts = str(inst.get("baseDir") or "").split("/")
        if len(parts) > 2 and parts[1].lower() == "users":
            user = parts[2]
    print(f"bridge {diag.get('bridgeVersion')} on {a.target}, KiCad {diag.get('kicadVersion') or (diag.get('installs') or [{}])[0].get('version')}")
    if not user:
        print("FATAL: could not learn the Windows user from kicad_readiness")
        return 2
    base = f"C:/Users/{user}/Documents/adom-kicad-drill/{time.strftime('%m%d%H%M%S')}"
    parts = {"sheet": sheet_part, "library": library_part, "file": file_part, "live": live_part, "cancel": cancel_part, "prefs": prefs_part}
    for name in a.part.split(","):
        parts[name.strip()](d, base)
    fails = [r for r in d.results if not r["pass"]]
    print(f"\n{len(d.results) - len(fails)} passed, {len(fails)} failed")
    for f in fails:
        print(f"  FAIL {f['verb']}: {f['what']}")
    if a.out:
        pathlib.Path(a.out).write_text(json.dumps(d.results, indent=1))
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main())