"""AutoCAD COM backend – live AutoCAD control via pywin32. All COM calls are routed through a single-threaded executor to satisfy AutoCAD's STA (Single-Threaded Apartment) COM requirement. """ from __future__ import annotations import asyncio import io import logging import math import os import sys import time import uuid from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any import config from engineering.measure import is_self_intersecting, polygon_area_perimeter from security import sanitize_macro_argument, sanitize_symbol_name, validate_path from . import ocs from .base import ( AutoCADBackend, BlockInfo, CapabilityMap, DrawingInfo, EntityInfo, FeatureCapability, LayerInfo, UnsupportedCapabilityError, deg2rad, normalize_lineweight, rad2deg, shoelace_area, ) from .contracts.settings import SUMMARY_FIELDS, validate_drawing_properties log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Optional COM imports # --------------------------------------------------------------------------- _WIN32_AVAILABLE = sys.platform == "win32" if _WIN32_AVAILABLE: try: import pythoncom import pywintypes import win32com.client import win32con # noqa: F401 import win32gui import win32ui _COM_IMPORTS_OK = True except ImportError: _COM_IMPORTS_OK = False else: _COM_IMPORTS_OK = False #: What `_run` may catch as a COM failure. A *tuple*, because where pywin32 is #: absent the right value is the empty one: `except ()` catches nothing, which #: is exactly correct when no `com_error` type exists to be raised. #: #: `except pywintypes.com_error` used to be written directly, and Python #: evaluates that name on the way out of every failed call — not only COM ones. #: On any host without pywin32 (all of Linux, and Windows installed without the #: `[com]` extra) it raised `NameError: name 'pywintypes' is not defined` and #: buried the real exception. A placeholder Exception subclass would be worse #: than the bug: it would silently swallow unrelated errors. _COM_ERROR: tuple[type[BaseException], ...] = (pywintypes.com_error,) if _COM_IMPORTS_OK else () #: ``RPC_E_CALL_REJECTED``: the callee's message filter refused the incoming #: call *before* running it, so nothing happened on the seat. AutoCAD answers #: it for a moment after ``Documents.Open`` / ``Add`` / ``Close`` while it is #: still switching documents — measured twice on 2026-09-22 (AutoCAD 2026, #: ``scripts/smoke_settings_com.py --build-dwt``: ``Documents.Count`` right #: after ``Documents.Open`` and right after ``Close(False)``) — and for as long #: as an operator or another client leaves a command prompt open. pywin32 has #: no ``CoRegisterMessageFilter``, the client-side hook that would retry this #: per call, so the read-only reads that follow a switch wait it out themselves. _RPC_E_CALL_REJECTED = -2147418111 def _wait_out_rejected_call(read: Callable[[], Any], *, budget_s: float, first_pause_s: float): """Run a read-only COM callable again while it answers ``RPC_E_CALL_REJECTED``. Only for callables that change nothing on the seat (the rejection means the refused call did not run, but an earlier call inside ``read`` did). Pauses double from ``first_pause_s`` up to one second and stop at ``budget_s``; any other ``com_error`` is re-raised on the first attempt. """ deadline = time.monotonic() + budget_s pause = first_pause_s while True: try: return read() except _COM_ERROR as exc: hr = exc.args[0] if exc.args else 0 if hr != _RPC_E_CALL_REJECTED or time.monotonic() + pause > deadline: raise time.sleep(pause) pause = min(pause * 2, 1.0) try: from PIL import Image as PILImage _PIL_OK = True except ImportError: _PIL_OK = False # --------------------------------------------------------------------------- # COM thread-state (accessed ONLY from the COM executor thread) # --------------------------------------------------------------------------- _COM_STATE: dict[str, Any] = {} # keys: "app" def _com_init(): """COM thread initializer – called once by the ThreadPoolExecutor.""" if _COM_IMPORTS_OK: pythoncom.CoInitialize() def _com_teardown(): """COM thread finalizer — drop the cached app and release the STA apartment. Submitted to the executor (R5) so the CoInitialize done in _com_init is paired with a CoUninitialize when the worker is torn down or a stuck SendCommand eventually unblocks, instead of leaking the apartment forever. """ _COM_STATE.pop("app", None) if _COM_IMPORTS_OK: try: pythoncom.CoUninitialize() except Exception: pass # AcCoordinateSystem, as Utility.TranslateCoordinates takes it (From / To). _AC_WORLD = 0 _AC_UCS = 1 def _apoint(x: float, y: float, z: float = 0.0): """Create a VARIANT double-array point for AutoCAD COM.""" return win32com.client.VARIANT( pythoncom.VT_ARRAY | pythoncom.VT_R8, [float(x), float(y), float(z)], ) def _av(values: list[float]): """Create a VARIANT double-array from a flat list.""" return win32com.client.VARIANT( pythoncom.VT_ARRAY | pythoncom.VT_R8, [float(v) for v in values], ) def _ai(values: list[int]): """Create a VARIANT short-int array.""" return win32com.client.VARIANT( pythoncom.VT_ARRAY | pythoncom.VT_I2, list(values), ) def _avar(values): """Create a VARIANT array of VARIANTs (what SetXRecordData takes for its values).""" return win32com.client.VARIANT( pythoncom.VT_ARRAY | pythoncom.VT_VARIANT, list(values), ) def _solid_3d_capability() -> FeatureCapability: """3D solids are opt-in (ENABLE_3D=true) even on the live COM backend.""" if config.settings.enable_3d: return FeatureCapability(True, "native") return FeatureCapability(False, reason="opt_in_disabled:set ENABLE_3D=true") def _acad_app(): """Return (or lazily create) the CAD Application COM object. Must only be called from the COM executor thread. The ProgID comes from ``CAD_PROGID`` (default ``AutoCAD.Application``) and is honoured on *both* paths deliberately. ``Dispatch`` is not a passive probe — it COM-launches the application and makes it visible — so honouring the setting on ``GetActiveObject`` and then falling back to AutoCAD would start the very application the operator said they were not using. """ if "app" not in _COM_STATE: progid = config.settings.cad_progid try: _COM_STATE["app"] = win32com.client.GetActiveObject(progid) except Exception as exc: log.debug("GetActiveObject(%r) failed, trying Dispatch: %s", progid, exc) try: _COM_STATE["app"] = win32com.client.Dispatch(progid) _COM_STATE["app"].Visible = True except Exception as exc: raise RuntimeError( f"Cannot connect to CAD application {progid!r}: {exc}. Make sure it is " f"installed and running. (Set CAD_PROGID to change it; default " f"{config.DEFAULT_CAD_PROGID!r}.)" ) from exc return _COM_STATE["app"] #: ``AcSaveAsType`` template constants, newest first: ac2018_Template, #: ac2013_Template, ac2010_Template. A seat older than the first refuses it #: with a COM error and the next one is tried. _TEMPLATE_SAVE_FORMATS = (66, 62, 50) def _template_cache_dir() -> Path: import tempfile return Path(tempfile.gettempdir()) / "acadmcp-templates" def _template_as_dwt(app, source: Path) -> tuple[Path, bool]: """A real ``.dwt`` for ``source``, built once through AutoCAD and cached. ``Documents.Add()`` accepts only a genuine DWT: measured on AutoCAD 2026, ``Add(templates/iso_a3_mech.dxf)`` returned the default acadiso drawing (layers ``['0']``, tabs ``Layout1``/``Layout2``) exactly as ``Add()`` and ``Add()`` did, and the DXF bytes renamed ``.dwt`` fared no better. Opening the file and saving it as a template (``SaveAs(path, ac2018_Template)``) is what yields the eleven layers and the ``A3`` tab, so that is done here — once per file content, keyed on the path, size and mtime, under the temp folder — and the ``.dwt`` is what ``Add`` is handed. Returns ``(dwt_path, cached)``. Must only be called from the COM executor thread. The conversion document is opened read-only and closed without saving in a ``finally``; the ``.dwt`` on disk is the only thing it leaves behind. """ import hashlib source = Path(source).resolve() stat = source.stat() digest = hashlib.sha1( f"{source}|{stat.st_size}|{stat.st_mtime_ns}".encode("utf-8", "surrogateescape") ).hexdigest()[:12] cache_dir = _template_cache_dir() cache_dir.mkdir(parents=True, exist_ok=True) dwt = cache_dir / f"{source.stem}-{digest}.dwt" if dwt.is_file() and dwt.stat().st_size > 0: return dwt, True doc = app.Documents.Open(str(source), True) try: last_exc: Exception | None = None for fmt in _TEMPLATE_SAVE_FORMATS: try: doc.SaveAs(str(dwt), fmt) break except _COM_ERROR as exc: last_exc = exc log.debug("SaveAs(%s, %d) refused: %s", dwt, fmt, exc) else: raise RuntimeError( f"could not save {source.name} as a .dwt template: {last_exc}" ) from last_exc finally: try: doc.Close(False) except Exception as exc: # the temp document must never stay open log.warning("template conversion document did not close: %s", exc) if not dwt.is_file() or dwt.stat().st_size == 0: raise RuntimeError(f"AutoCAD reported the template saved but {dwt} is not on disk") return dwt, False def _acad_doc(): """Return active AutoCAD document.""" app = _acad_app() if app.Documents.Count == 0: raise RuntimeError("No drawing is open in AutoCAD. Open or create a .dwg file first.") return app.ActiveDocument def _msp(): """The space new geometry goes into — the active layout's block. This returned ``ModelSpace`` unconditionally, so ``layout_set_current`` reported success and everything drawn afterwards still landed in model space. ``ActiveLayout.Block`` is the same object as ``ModelSpace`` while the Model tab is active, so the model-space path is unchanged; on a paper layout it is the sheet, which is where a title block belongs. """ doc = _acad_doc() try: return doc.ActiveLayout.Block except Exception as exc: # never strand a write over a layout probe log.debug("ActiveLayout.Block unavailable (%s); using ModelSpace", exc) return doc.ModelSpace def _int_member(obj, name: str) -> int | None: """``int(obj.)`` or ``None`` when the member is absent or not a number.""" try: return int(getattr(obj, name)) except Exception: return None def _owner_layout_block(doc, ent, handle): """The block table record that owns ``ent``, refused unless it is a layout. ``Explode()`` places the members in the owner space, so anything this method adds alongside them must go there too — never ``ActiveLayout``, which is whatever tab the user happens to have open. An owner that is a block definition means ``ent`` is a nested reference; exploding it in place would redefine the block under every other reference, so it is refused before any call is dispatched. An owner that cannot be read at all is refused too rather than guessed. """ try: owner = doc.ObjectIdToObject(ent.OwnerID) is_layout = bool(owner.IsLayout) except Exception as exc: raise RuntimeError( f"Entity {handle} has no resolvable owner layout ({exc}); refusing to " "explode into a guessed space" ) from exc if not is_layout: try: name = str(owner.Name) except Exception: name = "?" raise RuntimeError( f"Entity {handle} is nested inside block definition {name!r}; explode the " "outer reference instead (exploding here would redefine the block under " "every other reference)" ) return owner def _require_block_defined(doc, name): """``doc.Blocks.Item(name)``, or a ``ValueError`` before ``InsertBlock`` runs. ``InsertBlock`` with an unknown name raises a bare COM error after the call has already been dispatched; refusing here names the block, costs one ``Blocks.Item`` probe and matches the headless engine's message. A layout block (``IsLayout``) cannot be inserted into itself — same refusal by name. Shared by ``block_insert`` and ``entity_create_block_ref``. """ if not isinstance(name, str) or not name.strip(): raise TypeError("block name must be a non-empty string") try: block = doc.Blocks.Item(name) except Exception as exc: raise ValueError(f"block {name!r} is not defined") from exc try: is_layout = bool(block.IsLayout) except Exception: is_layout = False if is_layout: raise ValueError(f"block {name!r} is a layout block and cannot be inserted") return block _BUILTIN_LINETYPES = {"continuous", "bylayer", "byblock"} #: `drawing_properties_*` field → `IAcadSummaryInfo` property. _SUMMARY_ATTRS = {field: field.capitalize() for field in SUMMARY_FIELDS} #: `IAcadSummaryInfo` failure codes, as the `scode` in a `com_error`'s #: excepinfo (measured on AutoCAD 2026; the description next to them is #: English while the HRESULT text is localised, so the code is matched first). _SUMMARYINFO_DUPLICATE_KEY = -2145386475 # AddCustomInfo: 'Duplicate key' _SUMMARYINFO_KEY_NOT_FOUND = -2145386476 # Set/RemoveCustomByKey: 'Key not found' def _summaryinfo_error_is(exc: BaseException, scode: int, description: str) -> bool: """True when `exc` is the `IAcadSummaryInfo` failure `scode` / `description`. `com_error.args[2]` is the excepinfo tuple ``(wCode, source, description, helpfile, helpcontext, scode)``; a missing or foreign excepinfo is never a match, so an unrelated COM failure keeps propagating. """ info = exc.args[2] if len(exc.args) > 2 else None if not isinstance(info, tuple): return False code = info[5] if len(info) > 5 else None desc = str(info[2] or "") if len(info) > 2 else "" return code == scode or desc.strip().casefold() == description.casefold() def _com_error_description(exc: BaseException) -> str: """The English excepinfo description of a `com_error`, else ``str(exc)``. The HRESULT text is localised (``'Özel durum oluştu.'`` on a Turkish seat) while the excepinfo description is AutoCAD's own English message (``'Undefined linetype'``, ``'File system error'``), so the latter is what a caller can act on. """ info = exc.args[2] if len(exc.args) > 2 else None if isinstance(info, tuple) and len(info) > 2 and info[2]: return str(info[2]).strip() return str(exc) def _default_lin_file(doc) -> str: """``acadiso.lin`` on a metric seat, ``acad.lin`` on an imperial one. MEASUREMENT is a document system variable: ``GetVariable`` is a member of AcadDocument and the Application has none (measured on AutoCAD 2026, ``hasattr(app, "GetVariable") is False``). Read through the Application inside a try/except this always failed and every seat was "metric"; a read the document itself refuses is an error now, not a silent default. """ return "acadiso.lin" if int(doc.GetVariable("MEASUREMENT")) == 1 else "acad.lin" def _load_linetype(doc, name: str, lin_file: str) -> None: """``Linetypes.Load(name, file)`` — the ActiveX member for the job. The previous route, ``SendCommand("_-LINETYPE _LOAD \\n\\n")`` behind FILEDIA=0, was measured on AutoCAD 2026 (2026-09-22, scratch document): with FILEDIA=0 the macro loads nothing and leaves ``-LINETYPE`` at ``Enter an option [?/Create/Load/Set]:`` — the whole seat then rejects every COM call (RPC_E_CALL_REJECTED) until someone presses ESC — and with FILEDIA=1 it opens the modal "Select Linetype File" dialog. ``Load`` returns at once, touches no system variable, and raises a `com_error` whose excepinfo names the cause: ``'Undefined linetype'`` for a name the file does not carry, ``'File system error'`` for a file that is not on the support path, ``'Duplicate record name'`` for one already loaded (the callers check the table first, so that one is not reached). """ try: doc.Linetypes.Load(name, lin_file) except Exception as exc: raise RuntimeError( f"Failed to load linetype '{name}' from '{lin_file}': " f"{_com_error_description(exc)}. Check the linetype name spelling " "and that the .lin file is on AutoCAD's support path." ) from exc def _ensure_linetype_loaded(name: str) -> None: """If `name` is not already loaded, load it with ``Linetypes.Load``. Called by attribute setters so users can write `linetype="CENTER"` without having to remember to load it first. Must run on the COM thread. """ if not name or name.lower() in _BUILTIN_LINETYPES: return # Before any COM call: a linetype is a symbol-table name, and the DXF # rule refuses what AutoCAD could never load (and what a SendCommand # macro would have executed as a second command). name = sanitize_symbol_name(name, kind="linetype") doc = _acad_doc() existing = {doc.Linetypes.Item(i).Name.lower() for i in range(doc.Linetypes.Count)} if name.lower() in existing: return _load_linetype(doc, name, _default_lin_file(doc)) def _apply_entity_attrs(entity, layer: str | None, color: int | None, linetype: str | None): """Apply common entity attributes after creation.""" if layer is not None: entity.Layer = layer if color is not None: entity.color = int(color) if linetype is not None: _ensure_linetype_loaded(linetype) entity.Linetype = linetype def _regen(): """Regen the active viewport so new entities appear immediately on screen.""" if _COM_STATE.get("batch_mode"): return try: _acad_doc().Regen(0) # 0 = acActiveViewport except Exception as exc: log.debug("Regen failed: %s", exc) # ── styles (track E): module helpers ───────────────────────────────────────── # # The ActiveX object model has no per-style getters or setters for dimension # variables: `AcadDimStyle` exposes `Name` and `CopyFrom` and nothing else. A # style's values are therefore written the way AutoCAD's own DIMSTYLE command # writes them — set the DIM* system variables on the document (which makes # them overrides of the *current* style) and `CopyFrom(document)` to save # those settings into the target style. Assigning `ActiveDimStyle` restores # that style's saved settings and discards unsaved overrides (AutoCAD's # `-DIMSTYLE Restore` rule; the fake in tests/test_styles.py models exactly # this, and scripts/smoke_settings_com.py confirms it live). Reading is # one-sided: `GetVariable` reads the *current* style only, so `dimstyle_list` # reports values for the current style and `values_available: False` for the # rest rather than cycling the active style behind the operator's back. def _com_named(collection, name: str): """A collection member by name, case-insensitive (AutoCAD's rule), or None.""" wanted = name.strip().lower() for index in range(collection.Count): item = collection.Item(index) if str(item.Name).lower() == wanted: return item return None def _com_dimvar_for_write(var: str, value: Any) -> Any: """ActiveX takes DIMDSEP and the arrowhead/text-style names as strings.""" if var in ("DIMDSEP", "DIMBLK", "DIMBLK1", "DIMBLK2", "DIMTXSTY"): return str(value) if isinstance(value, float): return float(value) return int(value) def _com_dimvar_for_report(var: str, raw: Any) -> Any: if var == "DIMDSEP": if isinstance(raw, int): return "," if raw == 0 else chr(raw) text = str(raw) return text[:1] if text else "," if var in ("DIMBLK", "DIMBLK1", "DIMBLK2"): # AutoCAD reports the block name (``_OBLIQUE``, ``""`` for closed # filled) — already canonical; the same read-side rule as ezdxf keeps # the two engines' rows spelled identically. from engineering.standards.dimstyles import reported_arrowhead return reported_arrowhead(raw) if var == "DIMTXSTY": return str(raw) return raw def _same_dimvar(old: Any, new: Any) -> bool: if isinstance(old, (int, float)) and isinstance(new, (int, float)): return abs(float(old) - float(new)) < 1e-9 return str(old) == str(new) #: ``SelectionSet.Select`` mode ``acSelectionSetAll`` (AcSelect enum): every #: entity in the database, in every layout — ``ssget "X"``. _AC_SELECTION_SET_ALL = 5 #: Characters an ``ssget`` filter string treats as wildcards (AutoCAD's #: ``wcmatch`` grammar). Symbol names may carry most of them: a dimension #: style called ``A.B`` also matched ``A_B``, ``N#1`` matched ``N21``, #: ``P[1]`` matched ``P1`` and ``S@X`` matched ``SAX`` until the name was #: escaped (measured live, AutoCAD 2026). _WCMATCH_SPECIALS = frozenset("*?#@.~[]-,`") def _wcmatch_literal(text: str) -> str: """``text`` spelled so an ``ssget`` filter matches it literally. Every wildcard-significant character is escaped with a backtick, the grammar's own escape. A symbol name can never contain a backtick itself (AutoCAD refuses one), so the escaped form is unambiguous; the match stays case-insensitive, as any symbol-table name comparison is. """ return "".join("`" + char if char in _WCMATCH_SPECIALS else char for char in text) def _com_dimensions_using(doc, style_name: str) -> list[str]: """Handles of every DIMENSION, in every layout, whose style is ``style_name``. One filtered ``SelectionSet.Select(acSelectionSetAll)`` — the ``ssget "X"`` route with ``(0 . "DIMENSION") (3 .