AI Flow

"""Issue #183 end to end on tests/fixtures/single-layer.kicad_pcb: a CNC-milled one-face board of
through-hole parts. J1 and J2 sit against the board's edges and nets A (J1.1, R1, J2.2) and B (J1.2,
J2.1) must cross, so one face alone cannot close both.

- spec "routeLayers": ["B.Cu"]: route plans B.Cu copper only and no via, the rules-used line says so,
  and the crossing comes back as a finding with suggestions (allow F.Cu jumpers, move a part, widen the
  channel), never a refusal;
- no spec keys, the project's .kicad_dru with (constraint disallow via) and disallow track on F.Cu: the
  same mode, with a note naming the .kicad_dru;
- "routeLayers": ["B.Cu", "F.Cu"], "vias": false: route closes everything, F.Cu copper is wire jumpers
  that change face only on through-hole pads, the report lists them, gate passes and land route sends
  the bridge paths with no layer change inside them (no via);
- a hand-written plan with a via is a gate finding and land route lands nothing; a hand-written layer
  change on a through-hole pad is cut there without a via;
- pour plans no stitching via on this board.

A fake kicad-cli answers DRC with no violations and a fake adom-bridge records every call, so this runs
with no CAD and no desktop. Usage: python3 tests/single-layer-workflow.py <adom-aiflow binary>"""
import json, os, pathlib, subprocess, sys, tempfile

binary = str(pathlib.Path(sys.argv[1]).resolve())
fixture = pathlib.Path(__file__).resolve().parent / 'fixtures' / 'single-layer.kicad_pcb'

FAKE_BRIDGE = r'''#!/usr/bin/env python3
import json, os, sys
a = sys.argv[1:]
i = a.index("--target")
verb, args = a[i + 2], json.loads(a[i + 3])
with open(os.environ["FAKE_LOG"], "a") as f: f.write(json.dumps({"verb": verb, "args": args}) + "\n")
print(json.dumps({"success": True, "status": "ok", "revision": "r1"}))
'''
FAKE_CLI = ('#!/usr/bin/env python3\nimport sys, json\na = sys.argv[1:]\n'
            'if "--help" in a: print("--refill-zones"); sys.exit(0)\n'
            'out = a[a.index("--output") + 1]\njson.dump({"violations": [], "unconnected_items": []}, open(out, "w"))\n')
DRU = '(version 1)\n# milled on phenolic: one copper face, no plated holes\n(rule "no vias"\n\t(constraint disallow via))\n(rule "no top copper"\n\t(layer F.Cu)\n\t(constraint disallow track zone))\n'

passed = 0


def check(cond, what, out=''):
    global passed
    if not cond:
        sys.exit(f'FAIL: {what}\n{out}')
    passed += 1


def in_path_changes(entries):
    """Waypoints after a path's first that name a layer: a layer change (a via) inside the path."""
    return [(e['net'], p) for e in entries for p in e.get('paths', [e.get('points')]) if any(isinstance(w, dict) and 'layer' in w for w in p[1:])]


def path_layer(e, p):
    w = p[0]
    return e.get('layer') or (w.get('layer') if isinstance(w, dict) else None) or 'F.Cu'


with tempfile.TemporaryDirectory(prefix='aiflow-single-') as td:
    p = pathlib.Path(td)
    bindir = p / 'bin'; bindir.mkdir()
    (bindir / 'adom-bridge').write_text(FAKE_BRIDGE); (bindir / 'adom-bridge').chmod(0o755)
    (bindir / 'kicad-cli').write_text(FAKE_CLI); (bindir / 'kicad-cli').chmod(0o755)
    log = p / 'bridge.jsonl'
    env = dict(os.environ, AIFLOW_DASHBOARD='off', AIFLOW_NO_PART_SEARCH='1', AIFLOW_NO_PART_LOOKUP='1',
               ADOM_AIFLOW_KICAD_CLI=str(bindir / 'kicad-cli'), FAKE_LOG=str(log), PATH=f"{bindir}:{os.environ['PATH']}")
    env.pop('KICAD_REMOTE_TARGET', None)

    def call(run, *args):
        return subprocess.run([binary, '--run', str(run), '--ai-thread', 'single-test', *args], env=env, capture_output=True, text=True, timeout=300)

    def start(name, spec, dru=None):
        d = p / name; d.mkdir()
        (d / 'b.kicad_pcb').write_text(fixture.read_text())
        if dru: (d / 'b.kicad_dru').write_text(dru)
        (d / 'spec.json').write_text(json.dumps(spec))
        run = d / 'run'
        r = call(run, 'start', '--board', str(d / 'b.kicad_pcb'), '--spec', str(d / 'spec.json'), '--engine', 'test', '--target', 'fakebox', '--remote-board', f'C:/boards/{name}.kicad_pcb')
        check(r.returncode == 0, f'start {name}', r.stdout + r.stderr)
        return run

    def outcome(run, k):
        return json.loads((run / 'run.json').read_text())['outcomes'][k]

    # ---- 1. B.Cu only (spec): one face, no via, the crossing is a finding with suggestions
    run = start('bonly', {"board": "single-layer fixture", "clearance": 0.2, "routeLayers": ["B.Cu"]})
    r = call(run, 'route', '--passes', '3')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'ERROR' not in out, 'route on one face never refuses', out)
    check('route rules used:' in out and 'layers B.Cu (spec), vias off' in out, 'rules-used line shows the layers and vias', out)
    check('FINDINGS' in out and 'Allow wire jumpers' in out and 'Move or rotate a part' in out and 'Widen the channel' in out, 'the crossing is a finding with the three suggestions', out)
    o = outcome(run, 'route')
    check(o['unrouted'] >= 1, 'A and B cannot both close on one face', json.dumps(o)[:600])
    sf = o.get('singleFace') or {}
    check(sf.get('layers') == ['B.Cu'] and sf.get('viasPlanned') == 0 and sf.get('withJumper') == 0 and sf.get('singleLayer', 0) >= 1, 'singleFace counts', json.dumps(sf))
    check('closed on B.Cu alone' in out, 'route says how many closed on the single layer', out)
    plan = json.loads((run / 'plan-route.json').read_text())
    check(plan.get('vias') is False and plan.get('routeLayers') == ['B.Cu'], 'the plan says its mode', json.dumps(plan)[:300])
    check(not in_path_changes(plan['nets']), 'no layer change inside a path', json.dumps(in_path_changes(plan['nets'])))
    check(all(path_layer(e, pp) == 'B.Cu' for e in plan['nets'] for pp in e['paths']), 'every path on B.Cu', json.dumps(plan['nets']))

    # ---- 2. the project's .kicad_dru decides when the spec is silent (a note, not a refusal)
    run = start('dru', {"board": "single-layer fixture", "clearance": 0.2}, dru=DRU)
    r = call(run, 'route', '--passes', '3')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'ERROR' not in out, 'route with the dru never refuses', out)
    check('NOTE' in out and '(constraint disallow via)' in out and 'disallows tracks on F.Cu' in out, 'notes name the .kicad_dru rules applied', out)
    check('layers B.Cu (project .kicad_dru), vias off (project .kicad_dru)' in out, 'rules-used line says the dru decided', out)
    plan = json.loads((run / 'plan-route.json').read_text())
    check(not in_path_changes(plan['nets']) and all(path_layer(e, pp) == 'B.Cu' for e in plan['nets'] for pp in e['paths']), 'dru mode: B.Cu only, no via', json.dumps(plan['nets']))

    # ---- 3. B.Cu with F.Cu wire jumpers: everything closes, jumpers listed, gate and land route
    spec3 = {"board": "single-layer fixture", "clearance": 0.2, "routeLayers": ["B.Cu", "F.Cu"], "vias": False,
             "pours": [{"net": "A", "layers": ["B.Cu"], "outline": True}], "stitchVias": [{"net": "A", "x": 10, "y": 3}]}
    run = start('jumpers', spec3)
    r = call(run, 'route', '--passes', '3')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'OK: routing closed: 0 unrouted' in out, 'jumpers close the crossing', out)
    check('layers F.Cu + B.Cu (spec), vias off (spec): a layer change only on a through-hole pad' in out and 'wire jumpers' in out, 'rules-used line names the jumper mode', out)
    check('need a jumper' in out and '  jumper ' in out, 'route lists the jumpers', out)
    o = outcome(run, 'route')
    sf = o['singleFace']
    check(sf['jumpers']['count'] >= 1 and sf['withJumper'] >= 1 and sf['singleLayer'] >= 1 and sf['viasPlanned'] == 0 and sf['planIssues'] == [], 'singleFace record with jumpers', json.dumps(sf))
    check(all(j['from'] and j['to'] for j in sf['jumpers']['list']), 'each jumper names its ends', json.dumps(sf['jumpers']))
    check(o['routeQuality'].get('jumpers') == sf['jumpers']['count'], 'route quality counts the jumpers', json.dumps(o['routeQuality']))
    plan = json.loads((run / 'plan-route.json').read_text())
    check(not in_path_changes(plan['nets']), 'jumpers: no layer change inside a path', json.dumps(in_path_changes(plan['nets'])))
    layers = {path_layer(e, pp) for e in plan['nets'] for pp in e['paths']}
    check(layers == {'B.Cu', 'F.Cu'}, 'milled B.Cu and jumper F.Cu paths', json.dumps(plan['nets']))
    rep = json.loads((run / 'plan-route-report.json').read_text())
    check(rep['singleFace']['jumpers']['count'] == sf['jumpers']['count'] and rep['rules'].get('vias') is False, 'report carries the mode and jumpers', json.dumps(rep['singleFace']))
    r = call(run, 'gate')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'gate passed' in out, 'gate passes the jumper plan', out)
    g = json.loads((run / 'gate.json').read_text())
    check(g['singleFace']['issues'] == [] and g['singleFace']['jumpers']['count'] >= 1 and g['copper'].get('vias') in (0, None), 'gate: no single-face issue, jumpers, no via', json.dumps(g.get('singleFace')) + json.dumps(g.get('copper')))
    check(outcome(run, 'gate').get('singleFaceIssues') == 0, 'gate outcome records the single-face issues')
    r = call(run, 'land', 'route', '--pause-ms', '0')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'FINDINGS' not in out, 'land route', out)
    sent = [json.loads(l) for l in log.read_text().splitlines() if l.strip()]
    routes = [s['args'] for s in sent if s['verb'] == 'kicad_route_net']
    check(routes, 'kicad_route_net sent', json.dumps(sent)[:800])
    check(not in_path_changes(routes), 'landed paths change no layer inside a path (no via)', json.dumps(in_path_changes(routes)))
    check(not any('via' in s['verb'] for s in sent), 'no via verb sent', json.dumps([s['verb'] for s in sent]))
    # pours: no stitching via on a board with no vias
    r = call(run, 'pour')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and ', 0 vias' in out and 'stitchVias entry is left out' in out, 'pour plans no via', out)
    check(json.loads((run / 'plan-vias.json').read_text())['nets'] == [], 'plan-vias.json is empty')

    # ---- 4. hand-written plans: a via is a finding (gate, land), a layer change on a PTH pad is cut there
    with_via = {"nets": [{"net": "B", "width": 0.25, "paths": [[{"pad": "J1.2", "layer": "B.Cu"}, [6.0, 4.27], {"x": 6.0, "y": 4.27, "layer": "F.Cu"}, [24.0, 1.73], "J2.1"]]}]}
    (p / 'via-plan.json').write_text(json.dumps(with_via))
    r = call(run, 'gate', '--plan', str(p / 'via-plan.json'))
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'FINDINGS' in out and 'a via at (6.00, 4.27)' in out and 'ERROR' not in out, 'gate names the via as a finding', out)
    before = len(log.read_text().splitlines())
    r = call(run, 'land', 'route', '--plan', str(p / 'via-plan.json'), '--pause-ms', '0')
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'FINDINGS' in out and 'nothing landed' in out, 'land route lands no via', out)
    check(len(log.read_text().splitlines()) == before, 'no bridge call for the via plan')
    at_pad = {"nets": [{"net": "A", "width": 0.25, "paths": [[{"pad": "J1.1", "layer": "B.Cu"}, [12.0, 1.73], {"pad": "R1.1", "layer": "F.Cu"}, [15.0, 3.0], "R1.2"]]}]}
    (p / 'pad-plan.json').write_text(json.dumps(at_pad))
    r = call(run, 'gate', '--plan', str(p / 'pad-plan.json'))
    out = r.stdout + r.stderr
    check(r.returncode == 0 and 'made without a via' in out and 'a via at' not in out, 'a layer change on a PTH pad is cut there, no via', out)

print(f'OK single-layer-workflow: {passed} checks passed')