Fusion - the Fusion 360 Bridge

"""Record real Fusion window pixels while submitting one new copper item per call."""
import argparse
import json
from pathlib import Path
import time
from call import call


def require(r):
    if not r.get('success',r.get('ok',False)):
        raise RuntimeError(json.dumps(r))
    return r


def main():
    p=argparse.ArgumentParser()
    p.add_argument('--target',default='AdomLapper');p.add_argument('--document',default='fusion-routing')
    p.add_argument('--hwnd',type=int,required=True);p.add_argument('--output',type=Path,required=True)
    p.add_argument('--plan',type=Path,default=Path(__file__).parent/'fixture/plan.json')
    p.add_argument('--record',action='store_true');p.add_argument('--per-net',action='store_true')
    a=p.parse_args();a.output.mkdir(parents=True,exist_ok=True)
    counter=0
    def invoke(verb,args):
        nonlocal counter
        t=time.monotonic();r=call(verb,args,a.target)
        counter+=1
        (a.output/f'{counter:03d}-{verb}.json').write_text(json.dumps({'elapsedSeconds':time.monotonic()-t,'args':args,'result':r},indent=2)+'\n')
        return require(r)
    s=invoke('fusion_routing_state',{'expectDocument':a.document})
    if s['segmentCount'] or s['viaCount']:raise RuntimeError('Demo requires an unrouted board; existing copper is never erased by this script')
    guard=lambda s:{'expectDocument':a.document,'boardId':s['boardId'],'expectedRevision':s['revision']}
    plans=[]
    for row in json.loads(a.plan.read_text()):
        r=invoke('fusion_route_net',{**guard(s),**row,'dryRun':True})
        plans.append((row,r['plan']))
    invoke('fusion_execute_text_command',{'command':'Electron.ImmediateRepaint /full','expectDocument':a.document})
    recording=None;t0=time.monotonic()
    try:
        if a.record:
            recording=invoke('desktop_record_window_start',{'hwnd':a.hwnd,'fps':30,'reason':'Record real-time Fusion trace creation for the Adom routing demo'})
            time.sleep(1)
        for row,plan in plans:
            if a.per_net:
                s=invoke('fusion_route_net',{**guard(s),**row})
            else:
                for item in plan['items']:
                    if item['kind']=='segment':
                        route={'net':item['net'],'width':item['width'],'layer':item['layer'],'points':[item['start'],item['end']]}
                    else:
                        xy=item['at'];route={'net':item['net'],'layer':1,'viaDiameter':item['diameter'],'viaDrill':item['drill'],
                                           'points':[xy,{'x':xy[0],'y':xy[1],'layer':16}]}
                    s=invoke('fusion_route_net',{**guard(s),**route})
                    print(json.dumps({'net':row['net'],'segments':s['segmentCount'],'vias':s['viaCount'],'remainingNets':len(s['remainingNets'])}),flush=True)
                    # Deliberate readable pacing, included in the real-time take.
                    time.sleep(.2)
            s=invoke('fusion_routing_drc',guard(s))
            if s['drc']['errorCount']:raise RuntimeError('Native DRC found a physical violation')
        final=s
        if final['remainingNets'] or not final['drc']['clean']:raise RuntimeError('The board is not completely connected and DRC clean')
        (a.output/'result.json').write_text(json.dumps({'success':True,'elapsedSeconds':time.monotonic()-t0,'mode':'per-net' if a.per_net else 'per-trace','final':final},indent=2)+'\n')
        time.sleep(2)
    finally:
        if recording:
            rid=recording.get('recordingId') or recording.get('data',{}).get('recordingId')
            if not rid:raise RuntimeError('Recorder response has no ID: '+json.dumps(recording))
            invoke('desktop_record_window_stop',{'recordingId':rid,'reason':'Finalize the Fusion routing demo take'})

if __name__=='__main__':main()