Fusion - the Fusion 360 Bridge

@@ -3287,11 +3287,17 @@ def _multipart_body(fields: dict, files: list):     return b"\r\n".join(out), boundary  -def _service_glb_optimize(step_path: str, silk_top: str = None, silk_bottom: str = None,-                          pin: str = "medium", job_name: str = "fusion-board",-                          poll_max_s: int = 240) -> dict:-    """POST a STEP (+ optional silk PNGs) to service-step2glb molecule mode, poll to-    completion, return {ok, glbBytes, stats}. Pure urllib (no requests on the box)."""+# A User-Agent is REQUIRED on every service call. The *.adom.cloud edge WAF 403s the+# default "Python-urllib/x.y" UA (confirmed live 2026-07-14) while a browser/curl UA+# gets 200. Set it on the POST AND both polls or the call fails with an opaque 403.+def _service_ua():+    return "adom-fusion-bridge/%s" % BRIDGE_VERSION+++def _service_glb_submit(step_path: str, silk_top: str = None, silk_bottom: str = None,+                        pin: str = "medium", job_name: str = "fusion-board") -> dict:+    """POST a STEP (+ optional silk PNGs) to service-step2glb molecule mode. Returns+    {ok, jobId} - does NOT wait. Pure urllib (no requests on the box)."""     try:         with open(step_path, "rb") as f:             step_bytes = f.read()@@ -3306,16 +3312,11 @@ def _service_glb_optimize(step_path: str, silk_top: str = None, silk_bottom: str             except Exception:                 pass     body, boundary = _multipart_body({}, files)-    # NOTE: a User-Agent is REQUIRED. The *.adom.cloud edge WAF 403s the default-    # "Python-urllib/x.y" UA (confirmed live 2026-07-14) while a browser/curl UA gets-    # 200. urllib sends the default on EVERY request (POST + both polls), so set it-    # on all three or the whole call fails with an opaque 403.-    _UA = "adom-fusion-bridge/%s" % BRIDGE_VERSION     headers = {         "Content-Type": "multipart/form-data; boundary=" + boundary,         "X-Client": "fusion-bridge/adom",         "X-Job-Name": job_name,-        "User-Agent": _UA,+        "User-Agent": _service_ua(),     }     url = "%s/convert?molecule=true&pin=%s" % (STEP2GLB_URL, urllib.parse.quote(pin))     try:@@ -3328,32 +3329,47 @@ def _service_glb_optimize(step_path: str, silk_top: str = None, silk_bottom: str     job_id = queued.get("job_id")     if not job_id:         return {"ok": False, "error": "service did not return a job_id", "raw": queued}-    # Poll until complete (STEP tessellation of a big board can take a few minutes).-    stats, deadline = {}, _time.time() + poll_max_s-    while _time.time() < deadline:+    return {"ok": True, "jobId": job_id}+++def _service_glb_fetch(job_id: str, out_path: str = None, poll_s: int = 0) -> dict:+    """Poll a service job for up to poll_s seconds; if complete, optionally write the+    GLB to out_path. Returns {ok, status, stats, glbBytes?, wrote?}. poll_s=0 = one check."""+    ua = _service_ua()+    stats, deadline, first = {}, _time.time() + max(0, poll_s), True+    while first or _time.time() < deadline:+        first = False         try:-            preq = urllib.request.Request("%s/jobs/%s" % (STEP2GLB_URL, job_id),-                                          headers={"User-Agent": _UA})+            preq = urllib.request.Request("%s/jobs/%s" % (STEP2GLB_URL, job_id), headers={"User-Agent": ua})             with urllib.request.urlopen(preq, timeout=20) as r:                 stats = json.loads(r.read().decode("utf-8", "replace"))         except Exception:-            _time.sleep(5); continue+            _time.sleep(4); continue         st = stats.get("status")         if st == "complete":             break         if st == "error":-            return {"ok": False, "error": "service job failed: %s" % (stats.get("error") or stats), "jobId": job_id}+            return {"ok": False, "status": "error", "error": stats.get("error") or stats, "stats": stats}+        if _time.time() >= deadline:+            return {"ok": True, "status": st or "processing", "stats": stats}         _time.sleep(6)-    else:-        return {"ok": False, "error": "service job did not finish within %ss" % poll_max_s, "jobId": job_id}+    if stats.get("status") != "complete":+        return {"ok": True, "status": stats.get("status") or "processing", "stats": stats}     try:-        rreq = urllib.request.Request("%s/jobs/%s/result" % (STEP2GLB_URL, job_id),-                                      headers={"User-Agent": _UA})+        rreq = urllib.request.Request("%s/jobs/%s/result" % (STEP2GLB_URL, job_id), headers={"User-Agent": ua})         with urllib.request.urlopen(rreq, timeout=60) as r:             glb = r.read()     except Exception as e:-        return {"ok": False, "error": "could not download result GLB: %s" % e, "jobId": job_id}-    return {"ok": True, "glbBytes": glb, "stats": stats, "jobId": job_id}+        return {"ok": False, "status": "complete", "error": "download failed: %s" % e, "stats": stats}+    wrote = None+    if out_path:+        try:+            with open(out_path, "wb") as f:+                f.write(glb)+            wrote = out_path+        except Exception as e:+            return {"ok": False, "status": "complete", "error": "could not write GLB: %s" % e, "stats": stats}+    return {"ok": True, "status": "complete", "stats": stats, "glbBytes": glb, "wrote": wrote}   def _orchestrate_export_optimized_glb(args: dict) -> dict:@@ -3391,37 +3407,81 @@ def _orchestrate_export_optimized_glb(args: dict) -> dict:             silk_bottom = sb         if not (silk_top or silk_bottom):             silk_note = "silkscreen capture unavailable on this design; GLB built without a baked silk texture."-    # 4) Run the molecule optimizer and write the GLB.+    # 4) Submit to the molecule optimizer (fire-and-return: a big board's tessellation+    #    can run minutes, longer than the AD relay's request timeout, so we do NOT block+    #    the whole time here). Then bounded-wait up to `wait` seconds (default 90) so+    #    small boards still come back complete in one call.     job_name = os.path.splitext(os.path.basename(output_path))[0]-    opt = _service_glb_optimize(step_path, silk_top, silk_bottom, pin=pin, job_name=job_name)-    if not opt.get("ok"):-        return {"success": False, "error": opt.get("error"), "_hint": opt.get("_hint"),-                "stepPath": step_path, "note": "STEP exported OK; optimization failed - you can still convert the STEP another way."}-    try:-        with open(output_path, "wb") as f:-            f.write(opt["glbBytes"])-    except Exception as e:-        return {"success": False, "error": "could not write GLB to %s: %s" % (output_path, e)}-    stats = opt.get("stats") or {}+    sub = _service_glb_submit(step_path, silk_top, silk_bottom, pin=pin, job_name=job_name)+    if not sub.get("ok"):+        return {"success": False, "error": sub.get("error"), "_hint": sub.get("_hint"),+                "stepPath": step_path, "note": "STEP exported OK; optimize submit failed."}+    job_id = sub["jobId"]+    wait_s = int(args.get("wait", 90))+    fetched = _service_glb_fetch(job_id, out_path=output_path, poll_s=wait_s) if wait_s > 0 else {"ok": True, "status": "processing"}+    status_url = "%s/jobs/%s" % (STEP2GLB_URL, job_id)+    result_url = "%s/jobs/%s/result" % (STEP2GLB_URL, job_id)+    if fetched.get("status") == "complete" and fetched.get("wrote"):+        stats = fetched.get("stats") or {}+        size = len(fetched.get("glbBytes") or b"")+        return {+            "success": True, "status": "complete",+            "glbPath": output_path, "stepPath": step_path, "jobId": job_id,+            "silkTop": silk_top, "silkBottom": silk_bottom,+            "meshesBefore": stats.get("meshes_before"), "meshesAfter": stats.get("meshes_after"),+            "sizeBytes": size, "moleculeAnchored": stats.get("molecule_anchored"),+            "silkscreenApplied": stats.get("silkscreen_applied"), "note": silk_note,+            "message": "Optimized GLB written (%d KB, meshes %s->%s, anchored=%s, silk=%s) to %s" % (+                size // 1024, stats.get("meshes_before"), stats.get("meshes_after"),+                stats.get("molecule_anchored"), stats.get("silkscreen_applied"), output_path),+            "_hint": "Pull it with pull_file, then set it as component.parts.model_3d on a wiki component page. "+                     "Same optimizer as the molecule GLBs (Colby's pipeline).",+        }+    if not fetched.get("ok"):+        return {"success": False, "error": fetched.get("error"), "jobId": job_id, "stepPath": step_path,+                "statusUrl": status_url, "resultUrl": result_url}+    # Still processing after the bounded wait - hand back the job so the caller finishes it.     return {-        "success": True,-        "glbPath": output_path,-        "stepPath": step_path,-        "silkTop": silk_top, "silkBottom": silk_bottom,-        "meshesBefore": stats.get("meshes_before"), "meshesAfter": stats.get("meshes_after"),-        "sizeBytes": len(opt["glbBytes"]),-        "sizeBeforeCompress": stats.get("size_before_compress"),-        "moleculeAnchored": stats.get("molecule_anchored"),-        "silkscreenApplied": stats.get("silkscreen_applied"),-        "note": silk_note,-        "message": "Optimized GLB written (%d KB, meshes %s->%s, anchored=%s, silk=%s) to %s" % (-            len(opt["glbBytes"]) // 1024, stats.get("meshes_before"), stats.get("meshes_after"),-            stats.get("molecule_anchored"), stats.get("silkscreen_applied"), output_path),-        "_hint": "Pull it with pull_file, then put it on a wiki component page as component.parts.model_3d. "-                 "This is the same optimizer that produces the gorgeous molecule GLBs (Colby's pipeline).",+        "success": True, "status": fetched.get("status") or "processing",+        "pending": True, "jobId": job_id, "stepPath": step_path,+        "glbPath": output_path, "silkTop": silk_top, "silkBottom": silk_bottom, "note": silk_note,+        "statusUrl": status_url, "resultUrl": result_url,+        "message": "Optimize job %s submitted; still processing after %ds. Finish it with "+                   "fusion_fetch_optimized_glb {\"jobId\":\"%s\",\"outputPath\":\"%s\"} (re-call until complete)." % (+                       job_id, wait_s, job_id, output_path),+        "_hint": "Big boards tessellate for a few minutes. Either re-call fusion_fetch_optimized_glb "+                 "with this jobId (writes the GLB on the box when ready), or GET the resultUrl directly.",     }  +def _orchestrate_fetch_optimized_glb(args: dict) -> dict:+    """Fetch a previously-submitted optimize job's GLB (from fusion_export_optimized_glb's+    jobId). Bounded poll (default 90s); writes to outputPath when complete."""+    job_id = args.get("jobId")+    if not job_id:+        return {"success": False, "error": "No jobId.",+                "_hint": 'Usage: fusion_fetch_optimized_glb {"jobId":"...","outputPath":"C:/tmp/board.glb"}'}+    out_path = args.get("outputPath") or args.get("output_path")+    if out_path and not out_path.lower().endswith(".glb"):+        out_path = out_path + ".glb"+    wait_s = int(args.get("wait", 90))+    res = _service_glb_fetch(job_id, out_path=out_path, poll_s=wait_s)+    status_url = "%s/jobs/%s" % (STEP2GLB_URL, job_id)+    result_url = "%s/jobs/%s/result" % (STEP2GLB_URL, job_id)+    if res.get("status") == "complete" and res.get("wrote"):+        stats = res.get("stats") or {}+        size = len(res.get("glbBytes") or b"")+        return {"success": True, "status": "complete", "glbPath": out_path, "jobId": job_id,+                "sizeBytes": size, "meshesAfter": stats.get("meshes_after"),+                "moleculeAnchored": stats.get("molecule_anchored"), "silkscreenApplied": stats.get("silkscreen_applied"),+                "message": "Optimized GLB written (%d KB) to %s" % (size // 1024, out_path)}+    if not res.get("ok"):+        return {"success": False, "error": res.get("error"), "jobId": job_id, "statusUrl": status_url, "resultUrl": result_url}+    return {"success": True, "status": res.get("status") or "processing", "pending": True, "jobId": job_id,+            "statusUrl": status_url, "resultUrl": result_url,+            "message": "Job %s still %s; re-call fusion_fetch_optimized_glb to finish." % (job_id, res.get("status") or "processing")}++ def dispatch_command(command: str, args: dict) -> dict:     """Dispatch a command to the appropriate handler."""     # Direct handlers (don't need the add-in)@@ -3550,6 +3610,8 @@ def dispatch_command(command: str, args: dict) -> dict:         return _apply_failure_dialogs(_orchestrate_generate_package(args))     if command == "export_optimized_glb":         return _orchestrate_export_optimized_glb(args)+    if command == "fetch_optimized_glb":+        return _orchestrate_fetch_optimized_glb(args)      # Add-in proxy commands     if command in ADDIN_COMMANDS: