AI Flow

#!/usr/bin/env python3
"""Offline test of tools/kicad-cli-remote against a stand-in adom-bridge (no desktop, no KiCad).

The mock bridge plays a Linux desktop (run_script) or a Windows one (shell_execute with cmd lines),
copies files for send_files / pull_file, and resolves a relative pull_file saveTo against a different
folder than the caller's, the way the real bridge does (#61, #72). A stand-in kicad-cli writes what
the native one would. Checks: every verb on both OSes (the STEP export with its model files and its
native output passed through), relative outputs come back, `sch erc` pulls its
report, remote job folders are removed, and the Linux job folder sits under the home folder. The
project's sym-lib-table, fp-lib-table and a ${KIPRJMOD} .pretty folder reach the job folder in their
folders for the board and the schematic jobs (#96), and `pcb drc --schematic-parity` reaches kicad-cli
with the schematic hierarchy beside the board (the board's own, or KICAD_REMOTE_SCHEMATIC renamed)."""
import json, os, shutil, stat, subprocess, sys, tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
WRAPPER = os.path.join(HERE, "kicad-cli-remote")

MOCK = r'''#!/usr/bin/env python3
import base64, json, os, re, shutil, subprocess, sys
a = sys.argv[1:]
if "--target" in a:
    i = a.index("--target"); del a[i:i + 2]
if "--ai-thread" in a:
    i = a.index("--ai-thread"); del a[i:i + 2]
verb = a[0]
args = json.loads(a[1]) if len(a) > 1 and a[1].startswith("{") else {}
OS = os.environ["MOCK_OS"]; HOME = os.environ["MOCK_HOME"]; CLI = os.environ["MOCK_CLI"]
with open(os.environ["MOCK_LOG"], "a") as f: f.write(json.dumps([verb, args]) + "\n")
def out(d): print(json.dumps(d)); sys.exit(0)
def p(x): return x.replace("\\", "/")
if verb == "targets":
    out({"targets": [{"name": "box", "os": {"linux": "Linux", "mac": "macOS", "windows": "Windows"}[OS]}]})
if verb == "kicad_status":
    out({"install": {"kicadCli": "C:" + CLI.replace("/", "\\") if OS == "windows" else CLI}, "platform": {"os": "darwin" if OS == "mac" else OS}})
if verb == "send_files" and OS == "mac":
    # Adom Bridge on macOS (issue #195): dest is ignored, the files land in ~/Downloads
    d = os.path.join(HOME, "Downloads"); os.makedirs(d, exist_ok=True)
    for f in args["filePaths"]: shutil.copy(f, d)
    out({"success": True, "destinationPaths": [os.path.join(d, os.path.basename(f)) for f in args["filePaths"]]})
if verb == "send_files":
    d = p(args["dest"]); os.makedirs(d, exist_ok=True)
    for f in args["filePaths"]: shutil.copy(f, d)
    out({"success": True})
if verb == "pull_file":
    save = args["saveTo"]
    if not os.path.isabs(save): save = os.path.join(os.environ["MOCK_ELSEWHERE"], save)
    os.makedirs(save, exist_ok=True)
    for f in args["filePaths"]:
        if os.path.isfile(p(f)): shutil.copy(p(f), save)
    out({"success": True})
if verb == "run_script" and OS == "mac":
    out({"success": False, "error": "No command specified"})
if verb == "shell_execute" and OS == "mac":
    m = re.match(r'echo ([A-Za-z0-9+/=]+) \| base64 -d \| bash$', args["command"])
    assert m, args["command"]
    s = base64.b64decode(m.group(1)).decode()
    # macOS has shasum, not sha256sum
    r = subprocess.run(["bash", "-c", s], capture_output=True, text=True, env=dict(os.environ, HOME=HOME))
    out({"success": r.returncode == 0, "stdout": r.stdout, "stderr": r.stderr, "exitCode": r.returncode})
if verb == "run_script":
    assert OS == "linux"
    s = base64.b64decode(args["scriptB64"]).decode()
    r = subprocess.run(["bash", "-c", s], capture_output=True, text=True, env=dict(os.environ, HOME=HOME))
    out({"success": r.returncode == 0, "stdout": r.stdout, "stderr": r.stderr, "exitCode": r.returncode})
if verb == "shell_execute":
    assert OS == "windows"
    c = args["command"]
    if c == "echo %TEMP%": out({"stdout": HOME.replace("/", "\\") + "\\Temp\r\n"})
    m = re.match(r'dir /b "(.*)"$', c)
    if m: out({"stdout": "\r\n".join(sorted(os.listdir(p(m.group(1))))) + "\r\n"})
    m = re.match(r'rmdir /s /q "(.*)"$', c)
    if m: shutil.rmtree(p(m.group(1)), ignore_errors=True); out({"stdout": ""})
    m = re.match(r'mkdir "(.*?)" 2>nul & (.*)$', c)
    if m: os.makedirs(p(m.group(1)), exist_ok=True); c = m.group(2)
    env = dict(os.environ)
    while True:   # set "NAME=value" && ... (the remote environment, #177)
        m = re.match(r'set "([^=]+)=([^"]*)" && (.*)$', c)
        if not m: break
        v = p(m.group(2)); env[m.group(1)] = v[2:] if v.startswith("C:") else v; c = m.group(3)
    toks = re.findall(r'"([^"]*)"|(\S+)', c)
    argv = [p(q or r) for q, r in toks]
    assert argv[0].startswith("C:"), argv
    argv[0] = argv[0][2:]
    argv = [x[2:] if x.startswith("C:") else x for x in argv]
    r = subprocess.run(argv, capture_output=True, text=True, env=env)
    out({"stdout": r.stdout, "stderr": r.stderr, "exitCode": r.returncode})
out({"success": False, "error": "unknown verb " + verb})
'''

FAKE_CLI = r'''#!/bin/sh
out=""; prev=""; last=""; parity=no
for a in "$@"; do [ "$prev" = "--output" ] && out="$a"; [ "$a" = "--schematic-parity" ] && parity=yes; prev="$a"; last="$a"; done
# the project libraries beside the board or schematic (#96)
libs() { d="$(dirname "$1")"; for f in fp-lib-table sym-lib-table demo.kicad_sym lib/Demo.pretty/A.kicad_mod lib/Demo.pretty/B.kicad_mod; do [ -f "$d/$f" ] || { echo "missing $f" >&2; exit 5; }; done; }
case "$1 $2 $3" in
"version"*) echo "10.0.7";;
"pcb drc"*) [ -f "$last" ] || exit 3; libs "$last"
  if [ $parity = yes ]; then [ -f "${last%.kicad_pcb}.kicad_sch" ] && [ -f "$(dirname "$last")/sub/child.kicad_sch" ] || { echo "no schematic for parity" >&2; exit 6; }; fi
  printf '{"violations":[],"unconnected_items":[],"parity":"%s"}' "$parity" > "$out";;
"sch erc"*) [ -f "$last" ] || exit 3; [ -f "$(dirname "$last")/sub/child.kicad_sch" ] || exit 4; libs "$last"; printf '{"sheets":[{"path":"/","violations":[]}]}' > "$out";;
"sch export netlist") printf '(export (version "E") (components (comp (ref "R1"))) (nets))' > "$out";;
"pcb export gerbers") libs "$last"; for l in F_Cu B_Cu Edge_Cuts; do echo "G04 $l*" > "${out%/}/demo-$l.gbr"; done; echo '{}' > "${out%/}/demo-job.gbrjob";;
"pcb export drill") echo "M48" > "${out%/}/demo-PTH.drl";;
"pcb export pos") echo "Ref,Val" > "$out";;
"pcb export step") [ -f "$(dirname "$last")/models/part.step" ] || exit 4
  # a ${KICAD10_3RD_PARTY} model travelled into the job folder and the variable points there (#177)
  case "$KICAD10_3RD_PARTY" in "$(dirname "$last")"/*) ;; *) echo "3rd-party folder not in the job: $KICAD10_3RD_PARTY" >&2; exit 7;; esac
  [ -f "$KICAD10_3RD_PARTY/3dmodels/x.step" ] || { echo "3rd-party model missing" >&2; exit 8; }; echo "Wire self-interference check failed"; echo "z: 1.585; bounding box: x0: 1; y0: 2; x1: 3; y1: 4" >&2; echo "ISO-10303-21;" > "$out";;
*) echo "unknown $*" >&2; exit 2;;
esac
'''


def run(osname):
    root = tempfile.mkdtemp(prefix="kcr-test-")
    try:
        home, bindir, work, elsewhere = (os.path.join(root, n) for n in ("home", "bin", "work", "elsewhere"))
        for d in (home, bindir, work, elsewhere, os.path.join(work, "sub")): os.makedirs(d)
        for name, body in (("adom-bridge", MOCK), ("kicad-cli", FAKE_CLI)):
            fp = os.path.join(bindir, name)
            open(fp, "w").write(body); os.chmod(fp, os.stat(fp).st_mode | stat.S_IEXEC)
        open(os.path.join(work, "demo.kicad_pcb"), "w").write('(kicad_pcb (footprint "X" (model "${KIPRJMOD}/models/part.step")) (footprint "Y" (model "${KICAD10_3RD_PARTY}/3dmodels/x.step")))')
        os.makedirs(os.path.join(work, "models")); open(os.path.join(work, "models", "part.step"), "w").write("ISO-10303-21;")
        os.makedirs(os.path.join(work, "kicad", "3rdparty", "3dmodels")); open(os.path.join(work, "kicad", "3rdparty", "3dmodels", "x.step"), "w").write("ISO-10303-21;")
        open(os.path.join(work, "demo.kicad_pro"), "w").write("{}")
        open(os.path.join(work, "demo.kicad_sch"), "w").write('(kicad_sch (sheet (property "Sheetfile" "sub/child.kicad_sch")))')
        open(os.path.join(work, "sub", "child.kicad_sch"), "w").write("(kicad_sch)")
        # project libraries: a symbol file and a .pretty footprint folder, both named by ${KIPRJMOD}
        open(os.path.join(work, "sym-lib-table"), "w").write('(sym_lib_table (lib (name "Demo")(type "KiCad")(uri "${KIPRJMOD}/demo.kicad_sym")))')
        open(os.path.join(work, "fp-lib-table"), "w").write('(fp_lib_table (lib (name "Demo")(type "KiCad")(uri "${KIPRJMOD}/lib/Demo.pretty")))')
        open(os.path.join(work, "demo.kicad_sym"), "w").write("(kicad_symbol_lib)")
        os.makedirs(os.path.join(work, "lib", "Demo.pretty"))
        for n in ("A", "B"): open(os.path.join(work, "lib", "Demo.pretty", n + ".kicad_mod"), "w").write("(footprint)")
        # a run snapshot: another folder, another name, no lib tables beside it
        os.makedirs(os.path.join(work, "run")); shutil.copy(os.path.join(work, "demo.kicad_pcb"), os.path.join(work, "run", "board-1.kicad_pcb"))
        log = os.path.join(root, "log")
        env = dict(os.environ, PATH=bindir + ":" + os.environ["PATH"], MOCK_OS=osname, MOCK_HOME=home,
                   MOCK_CLI=os.path.join(bindir, "kicad-cli"), MOCK_LOG=log, MOCK_ELSEWHERE=elsewhere,
                   KICAD_REMOTE_TARGET="box", KICAD_REMOTE_THREAD="test")
        for k in ("KICAD_REMOTE_CLI", "KICAD_REMOTE_OS", "KICAD_REMOTE_TMP"): env.pop(k, None)
        def kc(*a, full=False):
            r = subprocess.run(["bash", WRAPPER, *a], cwd=work, env=env, capture_output=True, text=True)
            assert r.returncode == 0, f"{osname} {a}: {r.returncode} {r.stdout} {r.stderr}"
            return r if full else r.stdout
        assert kc("version").strip() == "10.0.7"
        kc("pcb", "drc", "--refill-zones", "--format", "json", "--output", "rel/drc.json", "demo.kicad_pcb")
        assert json.load(open(os.path.join(work, "rel/drc.json")))["parity"] == "no"
        # --schematic-parity reaches kicad-cli with demo.kicad_sch and its child sheet in the job folder
        kc("pcb", "drc", "--schematic-parity", "--format", "json", "--output", "rel/drc-p.json", "demo.kicad_pcb")
        assert json.load(open(os.path.join(work, "rel/drc-p.json")))["parity"] == "yes"
        # a snapshot with no schematic beside it: KICAD_REMOTE_SCHEMATIC, sent as board-1.kicad_sch, and the
        # schematic project's lib tables
        env["KICAD_REMOTE_SCHEMATIC"] = os.path.join(work, "demo.kicad_sch")
        kc("pcb", "drc", "--refill-zones", "--schematic-parity", "--output", "rel/drc-s.json", "run/board-1.kicad_pcb")
        assert json.load(open(os.path.join(work, "rel/drc-s.json")))["parity"] == "yes"
        del env["KICAD_REMOTE_SCHEMATIC"]
        kc("sch", "erc", "--format", "json", "--severity-all", "--output", "rel/erc.json", "demo.kicad_sch")
        assert json.load(open(os.path.join(work, "rel/erc.json")))["sheets"][0]["path"] == "/"
        kc("sch", "export", "netlist", "--format", "kicadsexpr", "--output", "rel/n.net", "demo.kicad_sch")
        kc("pcb", "export", "gerbers", "--output", "run/mfg/gerbers/", "--layers", "F.Cu,B.Cu,Edge.Cuts", "demo.kicad_pcb")
        kc("pcb", "export", "drill", "--output", "run/mfg/gerbers/", "--format", "excellon", "demo.kicad_pcb")
        kc("pcb", "export", "pos", "--output", "run/mfg/demo-cpl.csv", "--format", "csv", "demo.kicad_pcb")
        # the board twin's STEP: the ${KIPRJMOD} model travels, the native output passes through
        env["KICAD10_3RD_PARTY"] = os.path.join(work, "kicad", "3rdparty")   # what step-export sets (#177)
        st = kc("pcb", "export", "step", "--subst-models", "--include-tracks", "--include-silkscreen", "--output", "twin/demo.step", "demo.kicad_pcb", full=True)
        assert "Wire self-interference check failed" in st.stdout and "bounding box" in st.stderr, (st.stdout, st.stderr)
        assert open(os.path.join(work, "twin/demo.step")).read().startswith("ISO-10303-21"), "the STEP did not come back"
        g = sorted(os.listdir(os.path.join(work, "run/mfg/gerbers")))
        assert g == ["demo-B_Cu.gbr", "demo-Edge_Cuts.gbr", "demo-F_Cu.gbr", "demo-PTH.drl", "demo-job.gbrjob"], g
        for f in ("rel/drc.json", "rel/n.net", "run/mfg/demo-cpl.csv"): assert os.path.isfile(os.path.join(work, f)), f
        assert not os.listdir(elsewhere), "a relative saveTo was handed to the bridge"
        calls = [json.loads(l) for l in open(log)]
        sent = [os.path.relpath(f, work) for c in calls if c[0] == "send_files" for f in c[1]["filePaths"]]
        for f in ("fp-lib-table", "sym-lib-table", "lib/Demo.pretty/A.kicad_mod", "lib/Demo.pretty/B.kicad_mod"):
            assert sent.count(f) == 8, (f, sent.count(f))  # 3 drc, erc, netlist, gerbers, drill, pos
        dests = [c[1]["dest"].replace("\\", "/") for c in calls if c[0] == "send_files"]
        if osname == "linux":
            assert all(d.startswith(home + "/.adom-aiflow-kicad/aiflow-") for d in dests), dests
            assert not os.path.exists(os.path.join(home, ".adom-aiflow-kicad")), "the Linux job folder was left behind"
            assert not any(c[0] == "shell_execute" for c in calls)
        elif osname == "mac":
            # issue #195: bash goes through shell_execute, and the files sent to ~/Downloads were moved into the job
            assert all(d.startswith(home + "/.adom-aiflow-kicad/aiflow-") for d in dests), dests
            assert not os.path.exists(os.path.join(home, ".adom-aiflow-kicad")), "the macOS job folder was left behind"
            assert os.listdir(os.path.join(home, "Downloads")) == [], os.listdir(os.path.join(home, "Downloads"))
            assert not any(c[0] == "run_script" for c in calls)
        else:
            assert all(d.startswith(home + "/Temp/aiflow-") for d in dests), dests
            assert os.listdir(os.path.join(home, "Temp")) == [], os.listdir(os.path.join(home, "Temp"))
            assert not any(c[0] == "run_script" for c in calls)
        # sch erc with no report coming back fails
        os.remove(os.path.join(work, "sub", "child.kicad_sch"))
        r = subprocess.run(["bash", WRAPPER, "sch", "erc", "--output", "rel/erc2.json", "demo.kicad_sch"], cwd=work, env=env, capture_output=True, text=True)
        assert r.returncode == 1 and "no erc report came back" in r.stderr, (r.returncode, r.stderr)
        print(f"ok: {osname}")
    finally:
        shutil.rmtree(root, ignore_errors=True)


if __name__ == "__main__":
    for o in sys.argv[1:] or ["linux", "mac", "windows"]:
        run(o)