"""Exercise tracks B and G once against the live AutoCAD on this machine. Creates NEW documents (never touches the operator's open drawings), then runs the whole track through the same code the tools call: xref source: drawing_new -> a line -> drawing_export_dwg -> close apply_layer_set("mech") -> apply_titleblock (ISO 5457 A3 + ISO 7200) -> build_part -> draw_part -> add_view(section) -> dimension_part -> insert_std_part (ISO 4014 block + ACADMCP_MECH xdata) -> weld_symbol_prims -> extract_records -> rows_from_records -> draw_bom_table -> add_balloon -> xref_attach / xref_manage("list") / xref_manage("detach") -> run_critique(focus=None) -> drawing_export_dwg and prints what came back as JSON. Exit codes: 0 the sheet drew, the critique returned nothing and the .dwg exists 1 it drew but the critique found something or the DWG is missing 2 no live CAD application reachable (pywin32 missing, or the ProgID cannot be created) -- the COM paths then remain fake-tested only The backend's ``connect()`` is lazy on purpose (the server must start before AutoCAD is open), so reachability is read from ``system_status()`` rather than from ``connect()``. Every document this run opens is closed on the way out, also when a step fails, unless ``--keep`` is given. AUTOCAD_MCP_BACKEND=com uv run --frozen python scripts/smoke_mech_com.py uv run --frozen python scripts/smoke_mech_com.py --keep # leave it open """ from __future__ import annotations import asyncio import json import sys import tempfile import time from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) XDATA_APP = "ACADMCP_MECH" #: Every DWG since R13 opens with "AC10xx" (AC1032 for the 2018 format). DWG_MAGIC = b"AC10" #: ``RPC_E_CALL_REJECTED``: AutoCAD's message filter refused the call because #: it was busy (a document still initialising after ``Documents.Add`` is the #: case measured on this seat). A refused call was not executed. RPC_E_CALL_REJECTED = -2147418111 #: How long a refused call is retried before the smoke gives up on it. REJECTED_PATIENCE_S = 90.0 #: Names of the documents this run opened and has not closed yet. _OPENED: list[str] = [] def _patient(backend) -> None: """Retry any backend call AutoCAD refused with ``RPC_E_CALL_REJECTED``. Measured on AutoCAD 2026: the first call after ``Documents.Add`` was refused, the smoke died, and its cleanup - refused as well - left the scratch document open. Refused calls are retried with backoff for up to ``REJECTED_PATIENCE_S``; a document created twice by a retried ``drawing_new`` is still closed, because ``_new_document`` tracks the document list, not a name it assumes. """ run = backend._run async def patient(func, *args, **kwargs): deadline = time.monotonic() + REJECTED_PATIENCE_S pause = 0.5 while True: try: return await run(func, *args, **kwargs) except RuntimeError as exc: cause = exc.__cause__ hr = cause.args[0] if cause is not None and cause.args else None if hr != RPC_E_CALL_REJECTED or time.monotonic() + pause > deadline: raise await asyncio.sleep(pause) pause = min(pause * 2.0, 5.0) backend._run = patient async def _names(backend) -> set[str]: return {row["name"] for row in await backend.document_list()} async def _new_document(backend) -> str: """``drawing_new``, recording every document it added (normally one).""" before = await _names(backend) await backend.drawing_new() added = sorted((await _names(backend)) - before) _OPENED.extend(name for name in added if name not in _OPENED) return added[-1] if added else "" async def _close_leftovers(backend) -> None: open_names = {row["name"] for row in await backend.document_list()} for name in list(_OPENED): if name in open_names: try: await backend.document_close(name, discard=True) except Exception as exc: # best effort on the way out print(f"could not close {name}: {exc}", file=sys.stderr) _OPENED.remove(name) async def _xref_source(backend, workspace: Path) -> Path: """A one-line DWG in its own scratch document, saved and closed. An xref has to be a DWG, and it cannot be the host document itself - ``SaveAs`` rebinds the host to the file it wrote. """ created = await _new_document(backend) await backend.entity_create_line(0.0, 0.0, 50.0, 0.0, layer="0") path = workspace / "xref_source.dwg" before = await _names(backend) await backend.drawing_export_dwg(str(path), "R2018") rebound = await _rebound(backend, created, before) await backend.document_close(rebound, discard=True) _OPENED.remove(rebound) return path async def _rebound(backend, created: str, before: set[str]) -> str: """``SaveAs`` renames the active document to the file it wrote; track that name. Read from the document list rather than ``system_status``: the latter swallows a refused call and answers without ``active_document``. """ after = await _names(backend) renamed = sorted(after - before) rebound = renamed[0] if created not in after and len(renamed) == 1 else created if created in _OPENED: _OPENED.remove(created) if rebound not in _OPENED: _OPENED.append(rebound) return rebound async def _sheet(backend, workspace: Path, xref_path: Path) -> dict: from engineering.critique import run_critique from engineering.layers import apply_layer_set from engineering.mech.annotate import weld_symbol_prims from engineering.mech.draw import add_view, dimension_part, draw_part, draw_prims from engineering.mech.part import build_part from engineering.mech.stdparts import insert_std_part from engineering.sheet.bom import ( add_balloon, draw_bom_table, extract_records, rows_from_records, ) from engineering.sheet.titleblock import ( TB_HEIGHT, TitleBlockMetadata, apply_titleblock, titleblock_origin, ) created = await _new_document(backend) await apply_layer_set(backend, "mech") await apply_titleblock( backend, size="A3", metadata=TitleBlockMetadata(title="COM SMOKE", drawing_no="SMOKE-001", scale="1:1"), projection="first", frame=True, zones=True, marks=True, ) shaft = build_part( { "kind": "revolved", "name": "SHAFT", "material": "steel", "segments": [ {"length": 40.0, "d_outer": 30.0}, {"length": 20.0, "d_outer": 50.0, "d_inner": 20.0}, ], } ) drawn = await draw_part(backend, shaft, at=(125.0, 170.0), views=("front",), dimension=False) section = await add_view( backend, drawn["part_id"], "section", plane={"p1": [0.0, 0.0], "p2": [60.0, 0.0], "label": "A"}, style="full", ) dims = await dimension_part(backend, drawn["part_id"], style="chain") bolt = await insert_std_part(backend, "ISO 4014 - M12x60", at=(250.0, 120.0), view="side") bolt_handle = bolt.get("handle") or (bolt.get("handles") or [None])[0] got = await backend.entity_get_xdata(bolt_handle, XDATA_APP) await draw_prims(backend, weld_symbol_prims(kind="fillet", size=5.0), at=(60.0, 100.0)) records = await extract_records(backend) rows = rows_from_records(records) tb_x, tb_y = titleblock_origin("A3") table = await draw_bom_table(backend, rows, at=(tb_x, tb_y + TB_HEIGHT)) balloon = await add_balloon( backend, item=int(rows[0]["item"]), at=(250.0, 150.0), leader_to=(250.0, 125.0), targets=[str(h) for h in rows[0]["handles"]], ) attached = await backend.xref_attach(str(xref_path), (30.0, 30.0), 1.0, 0.0, "attach") listed = await backend.xref_manage(attached["name"], "list", None) detached = await backend.xref_manage(attached["name"], "detach", None) after = await backend.xref_manage(attached["name"], "list", None) issues = await run_critique(backend, None) dwg_path = workspace / "mech_smoke.dwg" before = await _names(backend) exported = await backend.drawing_export_dwg(str(dwg_path), "R2018") await _rebound(backend, created, before) head = dwg_path.read_bytes()[:6] if dwg_path.exists() else b"" return { "part_id": drawn["part_id"], "views": len(drawn.get("views") or []) + 1, "section_omitted": section.get("omitted"), "dimensions": int(dims.get("count") or len(dims.get("handles") or [])), "bolt_handle": bolt_handle, "xdata_chunks": len(got["xdata"].get(XDATA_APP, [])), "parts_list_rows": len(rows), "parts_list_representation": table.get("representation"), "balloon": balloon.get("handle"), "xref_attached": attached.get("name"), "xref_listed": len(listed.get("xrefs") or []), "xref_detached": bool(detached.get("ok")), "xrefs_after_detach": len(after.get("xrefs") or []), "dwg": { "path": exported.get("path", str(dwg_path)), "exists": dwg_path.exists(), "bytes": dwg_path.stat().st_size if dwg_path.exists() else 0, "magic": head.decode("ascii", "replace"), }, "critique_issues": len(issues), "critique": [issue.to_dict() for issue in issues], } async def main(keep_document: bool = False) -> int: try: from backends.com_backend import ComBackend except ImportError as exc: print(f"pywin32 not available: {exc}") return 2 backend = ComBackend() _patient(backend) try: await backend.connect() except Exception as exc: print(f"no live AutoCAD reachable: {exc}") return 2 try: status = await backend.system_status() if not status.get("connected"): print(f"no live AutoCAD reachable: {status.get('error', status)}") return 2 workspace = Path(tempfile.mkdtemp(prefix="mech_smoke_")) xref_path = await _xref_source(backend, workspace) report = { "autocad": status.get("autocad_version") or status.get("cad_progid"), **await _sheet(backend, workspace, xref_path), } print(json.dumps(report, indent=2)) ok = ( report["critique_issues"] == 0 and report["dwg"]["exists"] and report["dwg"]["magic"].startswith(DWG_MAGIC.decode()) and report["xrefs_after_detach"] == 0 ) return 0 if ok else 1 finally: if not keep_document: await _close_leftovers(backend) await backend.disconnect() if __name__ == "__main__": raise SystemExit(asyncio.run(main(keep_document="--keep" in sys.argv[1:])))