KiCad - the KiCad Bridge

"""Replay explicit net waypoints through ab in a live KiCad editor.

Require --target and ADOM_API for a remote test. Never mutates anything unless
routing_state proves the correct board and the expected dev build is installed.
"""
import argparse
import json
import os
from pathlib import Path
import subprocess
import time


def call(target, verb, args):
    proc=subprocess.run(['adom-bridge','--ai-thread','KiCad per-net routing','--target',target,
                         verb,json.dumps(args)],capture_output=True,text=True,timeout=240)
    result=json.loads(proc.stdout)
    if result.get('success') is False or result.get('status') == 'error':
        raise RuntimeError(f'{verb}: {result.get("errorCode")}: {result.get("error")}')
    return result


def main():
    p=argparse.ArgumentParser(description=__doc__)
    p.add_argument('--target',required=True)
    p.add_argument('--board',required=True)
    p.add_argument('--plan',required=True)
    p.add_argument('--output',default='routing-evidence')
    p.add_argument('--per-trace',action='store_true')
    p.add_argument('--socket')
    p.add_argument('--save',action='store_true')
    p.add_argument('--expect-version',default='',help='refuse unless bridgeVersion starts with this (default: any release that reports one)')
    a=p.parse_args()
    out=Path(a.output);out.mkdir(parents=True,exist_ok=True)
    common={'filePath':a.board}
    if a.socket:common['socketPath']=a.socket
    records=[]
    def invoke(verb,args):
        r=call(a.target,verb,{**common,**args})
        records.append({'verb':verb,'result':r})
        (out/'report.json').write_text(json.dumps(records,indent=2))
        return r
    state=invoke('kicad_routing_state',{})
    if not state.get('revision') or not state.get('bridgeVersion','').startswith(a.expect_version):
        raise RuntimeError('Bridge reported no revision or an unexpected bridgeVersion; refusing to continue')
    if state['segments'] or state['vias']:
        raise RuntimeError('Demo requires an unrouted scratch board; will not remove existing copper')
    (out/'before.json').write_text(json.dumps(state,indent=2))
    plans=json.loads(Path(a.plan).read_text())
    for index,plan in enumerate(plans):
        revision=state['revision']
        preview=invoke('kicad_route_net',{**plan,'expectedRevision':revision,'dryRun':True})
        work=[plan]
        if a.per_trace:
            work=[{'net':plan['net'],'width':s['width'],'layer':s['layer'],'points':[s['start'],s['end']]}
                  for s in preview['segments']]
            work.extend({'net':plan['net'],'layer':'F.Cu','viaSize':v['size'],'viaDrill':v['drill'],
                         'points':[v['at'],{'x':v['at'][0],'y':v['at'][1],'layer':'B.Cu'}]}
                        for v in preview['vias'])
        for item_index,item in enumerate(work):
            # The complete net was DRC-checked above. Replay its exact steps
            # against consecutive revisions, then check final board DRC.
            result=invoke('kicad_route_net',{**item,'expectedRevision':revision,
                'validate':not a.per_trace,
                'save':a.save and index==len(plans)-1 and item_index==len(work)-1})
            if not result.get('mutated') or not result.get('revision'):
                raise RuntimeError('Mutation outcome incomplete; inspect the board before retrying')
            revision=result['revision']
            print(f"{plan['net']}: step {item_index+1}/{len(work)}, {len(result['itemIds'])} copper items committed",flush=True)
            time.sleep(0.4)
        state=invoke('kicad_routing_state',{})
        print('Nets remaining:',len(state['netsRemaining']),flush=True)
    final=invoke('kicad_routing_validate',{})
    (out/'after.json').write_text(json.dumps(state,indent=2))
    if state['netsRemaining'] or final['unconnected'] or final['errors'] or final['stale']:
        raise RuntimeError('Final connectivity/DRC failed: '+str({k:final[k] for k in ['errors','warnings','unconnected','stale']}))
    print(json.dumps({k:final[k] for k in ['errors','warnings','unconnected','clean']}),flush=True)

if __name__=='__main__':main()