Fusion - the Fusion 360 Bridge

"""Container-hosted SSE for the existing Adom activity palette, with DOM acknowledgments."""
import ast,json,threading,time
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
from pathlib import Path
from urllib.request import Request,urlopen
ROOT=Path(__file__).resolve().parents[2]
PORT=8893
URL=f'http://localhost:{PORT}/'

def shell_html():
    # Reuse the bridge's branding/component without importing adsk in the container.
    tree=ast.parse((ROOT/'addin/AdomBridge/commands/activity_palette.py').read_text())
    nodes=[n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='_shell_html' or isinstance(n,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='ADOM_LOGO_SVG' for t in n.targets)]
    ns={};exec(compile(ast.Module(body=nodes,type_ignores=[]),'<activity palette>','exec'),ns)
    return ns['_shell_html'](json.dumps({'title':'Connecting to Astra routing','detail':'Waiting for measured results','tone':'done'}),'events').encode()

class Feed:
    def __init__(self):
        self.state={};self.sequence=0;self.ack=0;self.changed=threading.Condition();self.page=shell_html()
    def update(self,state):
        with self.changed:
            self.sequence+=1;self.state={**state,'sequence':self.sequence};self.changed.notify_all();return self.sequence
feed=Feed()
class Handler(BaseHTTPRequestHandler):
    def log_message(self,*args):pass
    def do_GET(self):
        if self.path=='/events':
            self.send_response(200);self.send_header('Content-Type','text/event-stream');self.send_header('Cache-Control','no-cache');self.end_headers();last=-1
            try:
                while True:
                    with feed.changed:
                        if last==feed.sequence:feed.changed.wait(10)
                        state=feed.state.copy();seq=feed.sequence
                    self.wfile.write(('data: '+json.dumps(state)+'\n\n').encode() if seq!=last else b': heartbeat\n\n');self.wfile.flush();last=seq
            except (BrokenPipeError,ConnectionResetError):pass
            return
        data=json.dumps({'sequence':feed.sequence,'ack':feed.ack,'state':feed.state}).encode() if self.path=='/state' else feed.page
        self.send_response(200);self.send_header('Content-Type','application/json' if self.path=='/state' else 'text/html; charset=utf-8');self.send_header('Content-Length',str(len(data)));self.end_headers();self.wfile.write(data)
    def do_POST(self):
        data=json.loads(self.rfile.read(int(self.headers['Content-Length'])))
        if self.path=='/ack':
            with feed.changed:feed.ack=max(feed.ack,int(data['sequence']));feed.changed.notify_all()
            out={'ack':feed.ack}
        elif self.path=='/state':out={'sequence':feed.update(data)}
        else:self.send_error(404);return
        body=json.dumps(out).encode();self.send_response(200);self.send_header('Content-Length',str(len(body)));self.end_headers();self.wfile.write(body)

def publish(state,wait=True):
    seq=json.load(urlopen(Request(URL+'state',data=json.dumps(state).encode(),headers={'Content-Type':'application/json'}),timeout=3))['sequence']
    if wait:
        end=time.monotonic()+5
        while time.monotonic()<end:
            status=json.load(urlopen(URL+'state',timeout=2))
            if status['ack']>=seq:return {'success':True,'sequence':seq,'domAcknowledged':True}
            time.sleep(.05)
        raise RuntimeError(f'Adom palette did not acknowledge event {seq}; stop before next copper edit')
    return {'success':True,'sequence':seq}

class LivePanel:
    def __init__(self,invoke,document,events_path=None,palette_hwnd=None):self.invoke=invoke;self.document=document;self.state={};self.events=[];self.events_path=events_path;self.palette_hwnd=palette_hwnd
    def update(self,create=False,event=None,**state):
        if self.palette_hwnd is not None:
            windows=self.invoke('desktop_list_windows',{})['windows']
            window=next((w for w in windows if w['hwnd']==self.palette_hwnd),None)
            if window is None or window['minimized']:raise RuntimeError('Adom palette was closed or minimized; stop this demo')
        self.state.update(state)
        if event:self.events.append(event)
        self.state['events']='\n'.join(self.events[-2:])
        payload={'title':'Astra: '+self.state.get('status','Routing'),'detail':'Codex Astra in Adom Hydrogen / bounded two-layer A*','percent':self.state.get('percent',0),'tone':'failed' if self.state.get('failed') else 'done' if self.state.get('percent')==100 else 'working','routing':{k:self.state.get(k,'') for k in ['connection','plan','checks','events']}}
        start=time.time();result=publish(payload)
        if self.events_path:
            with Path(self.events_path).open('a') as f:f.write(json.dumps({'start':start,'end':time.time(),'payload':payload,'result':result})+'\n')
        return result

if __name__=='__main__':
    print(URL,flush=True);ThreadingHTTPServer(('0.0.0.0',PORT),Handler).serve_forever()