feat(environment): portable layer states, named views and UCS on both engines
Layer states are ACADMCP_LAYERSTATES XRECORD chunks (<= 255 ASCII chars each), in the file and not in AutoCAD's Layer States Manager. Headless view restore writes the *Active VPORT (ezdxf's header has no $VIEWCTR; measured). UCS is stored and made current; tool coordinates stay WCS. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
co-authored by
Claude Opus 5
Claude Opus 5
parent
2805697d16
commit
dcc8f07bf4
@@ -132,6 +132,14 @@ def _ai(values: list[int]):
|
||||
)
|
||||
|
||||
|
||||
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:
|
||||
@@ -815,6 +823,17 @@ class ComBackend(AutoCADBackend):
|
||||
"solid_3d": _solid_3d_capability(),
|
||||
"lisp": FeatureCapability(True, "sanitized"),
|
||||
"documents": FeatureCapability(True, "native"),
|
||||
# Spelling pinned by F Task 24's test_layer_states_are_declared_as_ours_not_autocads
|
||||
# and quoted by the README and CLAUDE.md: mode "xrecord", this reason verbatim.
|
||||
"layer_states": FeatureCapability(
|
||||
True,
|
||||
"xrecord",
|
||||
reason="portable_acadmcp_xrecord;not_listed_in_autocad_layer_states_manager",
|
||||
),
|
||||
"named_views": FeatureCapability(True, "native"),
|
||||
"ucs": FeatureCapability(
|
||||
True, "native", reason="world_restore_via_ucs_command;tool_coordinates_stay_wcs"
|
||||
),
|
||||
"live_application": FeatureCapability(True, "native"),
|
||||
"preferences": FeatureCapability(True, "native", reason="whitelisted_keys_only"),
|
||||
"interactive_prompt": FeatureCapability(
|
||||
@@ -1403,6 +1422,414 @@ class ComBackend(AutoCADBackend):
|
||||
await self._ensure_document_state()
|
||||
return result
|
||||
|
||||
# ── layer states ──────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _com_layer_states(doc, *, create: bool):
|
||||
from engineering.environment.layer_states import DICT_NAME
|
||||
|
||||
try:
|
||||
return doc.Dictionaries.Item(DICT_NAME)
|
||||
except Exception as exc: # absent
|
||||
log.debug("Dictionaries.Item(%s): %s", DICT_NAME, exc)
|
||||
if not create:
|
||||
return None
|
||||
return doc.Dictionaries.Add(DICT_NAME)
|
||||
|
||||
@staticmethod
|
||||
def _com_dict_entries(states) -> list[tuple[str, Any]]:
|
||||
if states is None:
|
||||
return []
|
||||
out = []
|
||||
for index in range(int(states.Count)):
|
||||
obj = states.Item(index)
|
||||
out.append((str(states.GetName(obj)), obj))
|
||||
return out
|
||||
|
||||
def _com_layer_state_entry(self, states, name: str):
|
||||
for key, obj in self._com_dict_entries(states):
|
||||
if key.lower() == name.lower():
|
||||
return key, obj
|
||||
return None, None
|
||||
|
||||
@staticmethod
|
||||
def _com_xrecord_chunks(xrecord) -> list[str]:
|
||||
# pywin32 returns the two [out] parameters as a tuple, exactly as
|
||||
# GetBoundingBox() / GetXData() do elsewhere in this file.
|
||||
codes, values = xrecord.GetXRecordData()
|
||||
return [
|
||||
str(value)
|
||||
for code, value in zip(list(codes or []), list(values or []), strict=False)
|
||||
if int(code) == 1000
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _com_layer_snapshot(doc, description):
|
||||
from engineering.environment.layer_states import snapshot_from_layers
|
||||
|
||||
current = str(doc.ActiveLayer.Name)
|
||||
layers, plot = [], {}
|
||||
for index in range(int(doc.Layers.Count)):
|
||||
lyr = doc.Layers.Item(index)
|
||||
layers.append(_layer_info(lyr, current))
|
||||
plot[str(lyr.Name)] = bool(lyr.Plottable)
|
||||
return snapshot_from_layers(layers, current, plot=plot, description=description)
|
||||
|
||||
async def layer_state_save(self, name, description=None) -> dict:
|
||||
from engineering.environment.layer_states import encode_state, validate_state_name
|
||||
|
||||
clean = validate_state_name(name)
|
||||
if description is not None and not isinstance(description, str):
|
||||
raise TypeError("layer_state_save: description must be a string")
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
state = self._com_layer_snapshot(doc, description)
|
||||
chunks = encode_state(state)
|
||||
states = self._com_layer_states(doc, create=True)
|
||||
key, xrecord = self._com_layer_state_entry(states, clean)
|
||||
replaced = xrecord is not None
|
||||
if xrecord is None:
|
||||
xrecord = states.AddXRecord(clean)
|
||||
key = clean
|
||||
xrecord.SetXRecordData(_ai([1000] * len(chunks)), _avar(chunks))
|
||||
return {
|
||||
"ok": True,
|
||||
"name": key,
|
||||
"layer_count": len(state["layers"]),
|
||||
"replaced": replaced,
|
||||
"chunks": len(chunks),
|
||||
"backend": "com",
|
||||
}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def layer_state_restore(self, name, properties=None) -> dict:
|
||||
from engineering.environment.layer_states import (
|
||||
decode_state,
|
||||
diff_snapshot,
|
||||
validate_properties,
|
||||
validate_state_name,
|
||||
)
|
||||
|
||||
clean = validate_state_name(name)
|
||||
props = validate_properties(properties)
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
states = self._com_layer_states(doc, create=False)
|
||||
key, xrecord = self._com_layer_state_entry(states, clean)
|
||||
if xrecord is None:
|
||||
raise ValueError(
|
||||
f"layer_state_restore: no layer state named {clean!r}; "
|
||||
f"saved states: {[k for k, _ in self._com_dict_entries(states)]}"
|
||||
)
|
||||
state = decode_state(self._com_xrecord_chunks(xrecord))
|
||||
current_name = str(doc.ActiveLayer.Name)
|
||||
layers = [
|
||||
_layer_info(doc.Layers.Item(i), current_name) for i in range(int(doc.Layers.Count))
|
||||
]
|
||||
missing, new = diff_snapshot(state, layers)
|
||||
current = None
|
||||
warnings: list[str] = []
|
||||
# Current first: AutoCAD refuses to freeze the active layer, so the
|
||||
# active one must already be the state's before the loop freezes.
|
||||
if "current" in props and state["current_layer"] not in missing:
|
||||
doc.ActiveLayer = doc.Layers.Item(state["current_layer"])
|
||||
current = state["current_layer"]
|
||||
applied = 0
|
||||
for layer_name, snap in state["layers"].items():
|
||||
if layer_name in missing:
|
||||
continue
|
||||
lyr = doc.Layers.Item(layer_name)
|
||||
try:
|
||||
if "color" in props:
|
||||
lyr.Color = int(snap["color"])
|
||||
if "linetype" in props:
|
||||
_ensure_linetype_loaded(snap["linetype"])
|
||||
lyr.Linetype = snap["linetype"]
|
||||
if "lineweight" in props:
|
||||
lyr.LineWeight = int(snap["lineweight"])
|
||||
if "plot" in props:
|
||||
lyr.Plottable = bool(snap["plot"])
|
||||
if "on" in props:
|
||||
lyr.LayerOn = bool(snap["on"])
|
||||
if "frozen" in props:
|
||||
lyr.Freeze = bool(snap["frozen"])
|
||||
if "locked" in props:
|
||||
lyr.Lock = bool(snap["locked"])
|
||||
except Exception as exc: # e.g. freezing the active layer
|
||||
warnings.append(f"{layer_name}: {exc}")
|
||||
continue
|
||||
applied += 1
|
||||
_regen()
|
||||
result = {
|
||||
"ok": True,
|
||||
"name": key,
|
||||
"applied": {
|
||||
"layers": applied,
|
||||
"properties": list(props),
|
||||
"current_layer": current,
|
||||
},
|
||||
"missing_layers": missing,
|
||||
"new_layers": new,
|
||||
"backend": "com",
|
||||
}
|
||||
if warnings:
|
||||
result["warnings"] = warnings
|
||||
return result
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def layer_state_list(self) -> list[dict]:
|
||||
from engineering.environment.layer_states import decode_state
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
states = self._com_layer_states(doc, create=False)
|
||||
rows = []
|
||||
for key, xrecord in self._com_dict_entries(states):
|
||||
state = decode_state(self._com_xrecord_chunks(xrecord))
|
||||
rows.append(
|
||||
{
|
||||
"name": key,
|
||||
"description": state.get("description"),
|
||||
"layer_count": len(state["layers"]),
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def layer_state_delete(self, name) -> dict:
|
||||
from engineering.environment.layer_states import validate_state_name
|
||||
|
||||
clean = validate_state_name(name)
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
states = self._com_layer_states(doc, create=False)
|
||||
key, xrecord = self._com_layer_state_entry(states, clean)
|
||||
if xrecord is None:
|
||||
raise ValueError(
|
||||
f"layer_state_delete: no layer state named {clean!r}; "
|
||||
f"saved states: {[k for k, _ in self._com_dict_entries(states)]}"
|
||||
)
|
||||
removed = states.Remove(key)
|
||||
try:
|
||||
removed.Delete()
|
||||
except Exception as exc: # Remove already detached it
|
||||
log.debug("XRecord.Delete after Remove: %s", exc)
|
||||
return {"ok": True, "deleted": key, "backend": "com"}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
# ── named views ───────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _com_view(doc, name: str):
|
||||
try:
|
||||
return doc.Views.Item(name)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _com_view_row(view) -> dict:
|
||||
center = view.Center
|
||||
return {
|
||||
"name": str(view.Name),
|
||||
"center": [float(center[0]), float(center[1])],
|
||||
"height": float(view.Height),
|
||||
"width": float(view.Width),
|
||||
}
|
||||
|
||||
async def view_named_save(self, name, center=None, height=None, width=None) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.views import resolve_view_args
|
||||
|
||||
clean = validate_name(name, what="view name")
|
||||
args = resolve_view_args(center, height, width)
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
vport = doc.ActiveViewport
|
||||
vp_center = vport.Center
|
||||
vp_height = float(vport.Height) or 1.0
|
||||
aspect = float(vport.Width) / vp_height if vp_height else 1.5
|
||||
cx, cy = args["center"] or (float(vp_center[0]), float(vp_center[1]))
|
||||
h = args["height"] or vp_height
|
||||
w = args["width"] or h * aspect
|
||||
view = self._com_view(doc, clean)
|
||||
replaced = view is not None
|
||||
if view is None:
|
||||
view = doc.Views.Add(clean)
|
||||
view.Center = _av([cx, cy])
|
||||
view.Height = float(h)
|
||||
view.Width = float(w)
|
||||
return {
|
||||
"ok": True,
|
||||
"name": str(view.Name),
|
||||
"center": [float(cx), float(cy)],
|
||||
"height": float(h),
|
||||
"width": float(w),
|
||||
"replaced": replaced,
|
||||
"backend": "com",
|
||||
}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def view_named_restore(self, name) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
|
||||
clean = validate_name(name, what="view name")
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
view = self._com_view(doc, clean)
|
||||
if view is None:
|
||||
names = [str(doc.Views.Item(i).Name) for i in range(int(doc.Views.Count))]
|
||||
raise ValueError(
|
||||
f"view_named_restore: no named view {clean!r}; saved views: {names}"
|
||||
)
|
||||
row = self._com_view_row(view)
|
||||
vport = doc.ActiveViewport
|
||||
vport.Center = _av(row["center"])
|
||||
vport.Height = row["height"]
|
||||
vport.Width = row["width"]
|
||||
# ActiveX applies viewport changes only when the object is assigned back.
|
||||
doc.ActiveViewport = vport
|
||||
_regen()
|
||||
return {"ok": True, **row, "applied": "active_viewport", "backend": "com"}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def view_named_list(self) -> list[dict]:
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
return [self._com_view_row(doc.Views.Item(i)) for i in range(int(doc.Views.Count))]
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
# ── UCS ───────────────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _com_ucs(doc, name: str):
|
||||
try:
|
||||
return doc.UserCoordinateSystems.Item(name)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _com_ucs_row(ucs, current: str) -> dict:
|
||||
return {
|
||||
"name": str(ucs.Name),
|
||||
"origin": [float(c) for c in ucs.Origin],
|
||||
"x_axis": [float(c) for c in ucs.XVector],
|
||||
"y_axis": [float(c) for c in ucs.YVector],
|
||||
"current": current != "" and str(ucs.Name).lower() == current.lower(),
|
||||
}
|
||||
|
||||
async def ucs_list(self) -> list[dict]:
|
||||
from engineering.environment.ucs import WORLD
|
||||
|
||||
def _sync():
|
||||
app = _acad_app()
|
||||
doc = _acad_doc()
|
||||
current = str(app.GetVariable("UCSNAME") or "")
|
||||
rows = [
|
||||
{
|
||||
"name": WORLD,
|
||||
"origin": [0.0, 0.0, 0.0],
|
||||
"x_axis": [1.0, 0.0, 0.0],
|
||||
"y_axis": [0.0, 1.0, 0.0],
|
||||
"current": current == "",
|
||||
}
|
||||
]
|
||||
collection = doc.UserCoordinateSystems
|
||||
for index in range(int(collection.Count)):
|
||||
rows.append(self._com_ucs_row(collection.Item(index), current))
|
||||
return rows
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def ucs_set(self, name, origin, x_axis, y_axis) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.ucs import WORLD, resolve_ucs_axes
|
||||
|
||||
clean = validate_name(name, what="UCS name")
|
||||
if clean.lower() == WORLD:
|
||||
raise ValueError("ucs_set: 'world' is reserved; ucs_restore('world') resets to WCS")
|
||||
axes = resolve_ucs_axes(origin, x_axis, y_axis) # refuses before any ActiveX call
|
||||
|
||||
def _sync():
|
||||
doc = _acad_doc()
|
||||
o, x, y = axes["origin"], axes["x_axis"], axes["y_axis"]
|
||||
ucs = self._com_ucs(doc, clean)
|
||||
replaced = ucs is not None
|
||||
if ucs is None:
|
||||
# ActiveX takes POINTS on the axes, not direction vectors.
|
||||
ucs = doc.UserCoordinateSystems.Add(
|
||||
_apoint(*o),
|
||||
_apoint(o[0] + x[0], o[1] + x[1], o[2] + x[2]),
|
||||
_apoint(o[0] + y[0], o[1] + y[1], o[2] + y[2]),
|
||||
clean,
|
||||
)
|
||||
else:
|
||||
ucs.Origin = _apoint(*o)
|
||||
ucs.XVector = _apoint(*x)
|
||||
ucs.YVector = _apoint(*y)
|
||||
doc.ActiveUCS = ucs
|
||||
return {
|
||||
"ok": True,
|
||||
"name": str(ucs.Name),
|
||||
"origin": list(o),
|
||||
"x_axis": list(x),
|
||||
"y_axis": list(y),
|
||||
"replaced": replaced,
|
||||
"current": True,
|
||||
"backend": "com",
|
||||
}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
async def ucs_restore(self, name) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.ucs import WORLD
|
||||
|
||||
clean = validate_name(name, what="UCS name")
|
||||
|
||||
def _sync():
|
||||
app = _acad_app()
|
||||
doc = _acad_doc()
|
||||
if clean.lower() == WORLD:
|
||||
# No ActiveX member selects WCS; the command does. Same guard as
|
||||
# system_run_command: never send into an active prompt.
|
||||
try:
|
||||
cmd_active = int(app.GetVariable("CMDACTIVE"))
|
||||
except Exception as exc:
|
||||
log.debug("CMDACTIVE read failed, proceeding: %s", exc)
|
||||
cmd_active = 0
|
||||
if cmd_active:
|
||||
raise RuntimeError(
|
||||
"AutoCAD has an active command or prompt (CMDACTIVE="
|
||||
f"{cmd_active}). Press ESC in AutoCAD to cancel, then retry."
|
||||
)
|
||||
doc.SendCommand("_.UCS _W\n")
|
||||
return {"ok": True, "name": WORLD, "current": True, "backend": "com"}
|
||||
ucs = self._com_ucs(doc, clean)
|
||||
if ucs is None:
|
||||
names = [
|
||||
str(doc.UserCoordinateSystems.Item(i).Name)
|
||||
for i in range(int(doc.UserCoordinateSystems.Count))
|
||||
]
|
||||
raise ValueError(
|
||||
f"ucs_restore: no UCS named {clean!r}; saved: {names} (or 'world')"
|
||||
)
|
||||
doc.ActiveUCS = ucs
|
||||
return {"ok": True, "name": str(ucs.Name), "current": True, "backend": "com"}
|
||||
|
||||
return await self._run(_sync)
|
||||
|
||||
# ── selection filters (M8 / F1) ──────────────────────────────────────────
|
||||
#
|
||||
# VERIFIED against a live AutoCAD 2026 (2026-08-05).
|
||||
|
||||
@@ -44,7 +44,60 @@ class EnvironmentContract(ABC):
|
||||
"""
|
||||
...
|
||||
|
||||
# ── layer states, named views, UCS: appended by Task 22 ──────────────────
|
||||
# ── layer states (portable ACADMCP_LAYERSTATES XRECORDs, both engines) ──
|
||||
|
||||
@abstractmethod
|
||||
async def layer_state_save(self, name: str, description: str | None = None) -> dict:
|
||||
"""→ ``{"ok", "name", "layer_count", "replaced"}``."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def layer_state_restore(self, name: str, properties: list[str] | None = None) -> dict:
|
||||
"""``properties`` ⊆ {"on","frozen","locked","color","linetype","lineweight","plot","current"};
|
||||
→ ``{"ok", "name", "applied", "missing_layers", "new_layers"}``."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def layer_state_list(self) -> list[dict]:
|
||||
"""→ ``[{"name", "description", "layer_count"}]``."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def layer_state_delete(self, name: str) -> dict: ...
|
||||
|
||||
# ── named views ─────────────────────────────────────────────────────────
|
||||
|
||||
@abstractmethod
|
||||
async def view_named_save(
|
||||
self, name: str, center=None, height: float | None = None, width: float | None = None
|
||||
) -> dict:
|
||||
"""→ ``{"ok", "name", "center": [x, y], "height", "width", "replaced"}``."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def view_named_restore(self, name: str) -> dict:
|
||||
"""ezdxf: sets the ``*Active`` VPORT, ``"applied": "header_only"``."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def view_named_list(self) -> list[dict]: ...
|
||||
|
||||
# ── UCS (stored and made current; tool coordinates stay WCS) ────────────
|
||||
|
||||
@abstractmethod
|
||||
async def ucs_list(self) -> list[dict]:
|
||||
"""→ ``[{"name", "origin", "x_axis", "y_axis", "current": bool}]`` + the implicit "world"."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def ucs_set(self, name: str, origin, x_axis, y_axis) -> dict:
|
||||
"""non-orthogonal → ``ValueError`` with the measured angle."""
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
async def ucs_restore(self, name: str) -> dict:
|
||||
"""``"world"`` resets."""
|
||||
...
|
||||
|
||||
# ── application, preferences, operator prompts (COM only) ───────────────
|
||||
|
||||
|
||||
@@ -1246,6 +1246,21 @@ class EzdxfBackend(AutoCADBackend):
|
||||
),
|
||||
"lisp": FeatureCapability(False, reason="live_com_only"),
|
||||
"documents": FeatureCapability(True, "registry"),
|
||||
# Spelling pinned by F Task 24's test_layer_states_are_declared_as_ours_not_autocads
|
||||
# and quoted by the README and CLAUDE.md: mode "xrecord", this reason verbatim.
|
||||
"layer_states": FeatureCapability(
|
||||
True,
|
||||
"xrecord",
|
||||
reason="portable_acadmcp_xrecord;not_listed_in_autocad_layer_states_manager",
|
||||
),
|
||||
"named_views": FeatureCapability(
|
||||
True,
|
||||
"vport_active",
|
||||
reason="restore_sets_the_view_the_file_opens_on;no_live_display",
|
||||
),
|
||||
"ucs": FeatureCapability(
|
||||
True, "stored_not_interpreted", reason="tool_coordinates_stay_wcs"
|
||||
),
|
||||
# Track E keys. `@capability` registers its key at import, and
|
||||
# tests/test_capability_contract.py checks both maps from that
|
||||
# moment on — so every key the contract module uses is declared
|
||||
@@ -2312,6 +2327,431 @@ class EzdxfBackend(AutoCADBackend):
|
||||
|
||||
return await self._async(_sync, quarantine_exit=closing_active)
|
||||
|
||||
# ── layer states ──────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _layer_state_entry(states, name: str):
|
||||
"""Case-insensitive lookup → ``(stored_key, xrecord)`` or ``(None, None)``."""
|
||||
if states is None:
|
||||
return None, None
|
||||
if name in states:
|
||||
return name, states.get(name)
|
||||
for key in states.keys():
|
||||
if key.lower() == name.lower():
|
||||
return key, states.get(key)
|
||||
return None, None
|
||||
|
||||
@staticmethod
|
||||
def _layer_state_chunks(xrecord) -> list[str]:
|
||||
return [str(tag.value) for tag in xrecord.tags if tag.code == 1000]
|
||||
|
||||
@staticmethod
|
||||
def _layer_states(doc):
|
||||
from engineering.environment.layer_states import DICT_NAME
|
||||
|
||||
return doc.rootdict.get(DICT_NAME) if DICT_NAME in doc.rootdict else None
|
||||
|
||||
@staticmethod
|
||||
def _layer_state_names(states) -> list[str]:
|
||||
return list(states.keys()) if states is not None else []
|
||||
|
||||
def _layer_state_snapshot(self, doc, description):
|
||||
from engineering.environment.layer_states import snapshot_from_layers
|
||||
|
||||
layers = [_layer_info_dxf(lyr, self._current_layer) for lyr in doc.layers]
|
||||
plot = {lyr.dxf.name: bool(lyr.dxf.get("plot", 1)) for lyr in doc.layers}
|
||||
return snapshot_from_layers(layers, self._current_layer, plot=plot, description=description)
|
||||
|
||||
async def layer_state_save(self, name, description=None) -> dict:
|
||||
"""Snapshot the layer table into ``ACADMCP_LAYERSTATES/<name>``.
|
||||
|
||||
Portable and file-resident; not an AutoCAD Layer States Manager entry.
|
||||
"""
|
||||
from engineering.environment.layer_states import (
|
||||
DICT_NAME,
|
||||
encode_state,
|
||||
validate_state_name,
|
||||
)
|
||||
|
||||
clean = validate_state_name(name)
|
||||
if description is not None and not isinstance(description, str):
|
||||
raise TypeError("layer_state_save: description must be a string")
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
state = self._layer_state_snapshot(doc, description)
|
||||
chunks = encode_state(state)
|
||||
states = doc.rootdict.get_required_dict(DICT_NAME)
|
||||
existing_key, _ = self._layer_state_entry(states, clean)
|
||||
replaced = existing_key is not None
|
||||
if replaced:
|
||||
states.remove(existing_key)
|
||||
xrecord = states.add_xrecord(clean)
|
||||
xrecord.extend([(1000, chunk) for chunk in chunks])
|
||||
self._mark_dirty()
|
||||
return {
|
||||
"ok": True,
|
||||
"name": clean,
|
||||
"layer_count": len(state["layers"]),
|
||||
"replaced": replaced,
|
||||
"chunks": len(chunks),
|
||||
"backend": "ezdxf",
|
||||
}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def layer_state_restore(self, name, properties=None) -> dict:
|
||||
from engineering.environment.layer_states import (
|
||||
decode_state,
|
||||
diff_snapshot,
|
||||
validate_properties,
|
||||
validate_state_name,
|
||||
)
|
||||
|
||||
clean = validate_state_name(name)
|
||||
props = validate_properties(properties)
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
states = self._layer_states(doc)
|
||||
key, xrecord = self._layer_state_entry(states, clean)
|
||||
if xrecord is None:
|
||||
raise ValueError(
|
||||
f"layer_state_restore: no layer state named {clean!r}; "
|
||||
f"saved states: {self._layer_state_names(states)}"
|
||||
)
|
||||
state = decode_state(self._layer_state_chunks(xrecord))
|
||||
layers = [_layer_info_dxf(lyr, self._current_layer) for lyr in doc.layers]
|
||||
missing, new = diff_snapshot(state, layers)
|
||||
applied = 0
|
||||
for layer_name, snap in state["layers"].items():
|
||||
if layer_name not in doc.layers:
|
||||
continue # missing: reported, never created
|
||||
lyr = doc.layers.get(layer_name)
|
||||
# Colour first: `Layer.color` keeps the on/off sign, and the
|
||||
# on/off call below must see the final colour.
|
||||
if "color" in props:
|
||||
lyr.color = int(snap["color"])
|
||||
if "linetype" in props:
|
||||
_ensure_linetype_loaded(doc, snap["linetype"])
|
||||
lyr.dxf.linetype = snap["linetype"]
|
||||
if "lineweight" in props:
|
||||
lyr.dxf.lineweight = int(snap["lineweight"])
|
||||
if "plot" in props:
|
||||
lyr.dxf.plot = 1 if snap["plot"] else 0
|
||||
if "on" in props:
|
||||
if snap["on"]:
|
||||
lyr.on()
|
||||
else:
|
||||
lyr.off()
|
||||
if "frozen" in props:
|
||||
if snap["frozen"]:
|
||||
lyr.freeze()
|
||||
else:
|
||||
lyr.thaw()
|
||||
if "locked" in props:
|
||||
if snap["locked"]:
|
||||
lyr.lock()
|
||||
else:
|
||||
lyr.unlock()
|
||||
applied += 1
|
||||
current = None
|
||||
if "current" in props and state["current_layer"] in doc.layers:
|
||||
current = state["current_layer"]
|
||||
self._current_layer = current
|
||||
doc.header["$CLAYER"] = current
|
||||
self._mark_dirty()
|
||||
return {
|
||||
"ok": True,
|
||||
"name": key,
|
||||
"applied": {
|
||||
"layers": applied,
|
||||
"properties": list(props),
|
||||
"current_layer": current,
|
||||
},
|
||||
"missing_layers": missing,
|
||||
"new_layers": new,
|
||||
"backend": "ezdxf",
|
||||
}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def layer_state_list(self) -> list[dict]:
|
||||
from engineering.environment.layer_states import decode_state
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
states = self._layer_states(doc)
|
||||
rows = []
|
||||
for key in self._layer_state_names(states):
|
||||
state = decode_state(self._layer_state_chunks(states.get(key)))
|
||||
rows.append(
|
||||
{
|
||||
"name": key,
|
||||
"description": state.get("description"),
|
||||
"layer_count": len(state["layers"]),
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def layer_state_delete(self, name) -> dict:
|
||||
from engineering.environment.layer_states import validate_state_name
|
||||
|
||||
clean = validate_state_name(name)
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
states = self._layer_states(doc)
|
||||
key, xrecord = self._layer_state_entry(states, clean)
|
||||
if xrecord is None:
|
||||
raise ValueError(
|
||||
f"layer_state_delete: no layer state named {clean!r}; "
|
||||
f"saved states: {self._layer_state_names(states)}"
|
||||
)
|
||||
states.remove(key)
|
||||
self._mark_dirty()
|
||||
return {"ok": True, "deleted": key, "backend": "ezdxf"}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
# ── named views ───────────────────────────────────────────────────────────
|
||||
|
||||
# TEMPORARY (track-e/environment worktree only): group C's Task 14 defines
|
||||
# the same `_active_vport` in its `# ── settings (track E, group C) ──`
|
||||
# block, and C's is the one that ships. This copy exists so this worktree
|
||||
# runs without C's code; the merge agent deletes THIS definition (two in
|
||||
# one class are ruff F811, and the later one would silently win).
|
||||
@staticmethod
|
||||
def _active_vport(doc):
|
||||
"""The ``*Active`` VPORT entry — the view the file opens on. Created if absent."""
|
||||
entries = doc.viewports.get("*Active")
|
||||
if entries:
|
||||
return entries[0]
|
||||
return doc.viewports.new("*Active")
|
||||
|
||||
@staticmethod
|
||||
def _view_entry(doc, name: str):
|
||||
if name in doc.views: # case-insensitive in ezdxf's table
|
||||
return doc.views.get(name)
|
||||
return None
|
||||
|
||||
async def view_named_save(self, name, center=None, height=None, width=None) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.views import resolve_view_args
|
||||
|
||||
clean = validate_name(name, what="view name")
|
||||
args = resolve_view_args(center, height, width)
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
vport = self._active_vport(doc)
|
||||
aspect = float(vport.dxf.get("aspect_ratio", 1.34)) or 1.34
|
||||
if args["center"] is None or args["height"] is None:
|
||||
# Headless there is no display to read a "current view" from;
|
||||
# the drawing's extents are the honest default.
|
||||
bb = ezdxf_bbox.extents(doc.modelspace()) if _BBOX_OK else None
|
||||
if bb is None or not bb.has_data:
|
||||
raise ValueError(
|
||||
f"view_named_save: {clean!r} — an empty drawing has no extents; "
|
||||
"pass center and height"
|
||||
)
|
||||
ext_center = (
|
||||
(bb.extmin.x + bb.extmax.x) / 2.0,
|
||||
(bb.extmin.y + bb.extmax.y) / 2.0,
|
||||
)
|
||||
# The window must hold the extents at the viewport's aspect:
|
||||
# whichever side binds decides the height.
|
||||
ext_height = max(bb.extmax.y - bb.extmin.y, (bb.extmax.x - bb.extmin.x) / aspect)
|
||||
if ext_height <= 0:
|
||||
ext_height = 1.0
|
||||
else:
|
||||
ext_center, ext_height = args["center"], args["height"]
|
||||
cx, cy = args["center"] or ext_center
|
||||
h = args["height"] or ext_height
|
||||
w = args["width"] or h * aspect
|
||||
existing = self._view_entry(doc, clean)
|
||||
replaced = existing is not None
|
||||
stored = existing.dxf.name if replaced else clean
|
||||
if replaced:
|
||||
doc.views.remove(stored)
|
||||
doc.views.add(stored, dxfattribs={"center": (cx, cy, 0.0), "height": h, "width": w})
|
||||
self._mark_dirty()
|
||||
return {
|
||||
"ok": True,
|
||||
"name": stored,
|
||||
"center": [float(cx), float(cy)],
|
||||
"height": float(h),
|
||||
"width": float(w),
|
||||
"replaced": replaced,
|
||||
"backend": "ezdxf",
|
||||
}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def view_named_restore(self, name) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
|
||||
clean = validate_name(name, what="view name")
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
view = self._view_entry(doc, clean)
|
||||
if view is None:
|
||||
raise ValueError(
|
||||
f"view_named_restore: no named view {clean!r}; "
|
||||
f"saved views: {[v.dxf.name for v in doc.views]}"
|
||||
)
|
||||
cx, cy = float(view.dxf.center.x), float(view.dxf.center.y)
|
||||
h, w = float(view.dxf.height), float(view.dxf.width)
|
||||
vport = self._active_vport(doc)
|
||||
vport.dxf.center = (cx, cy)
|
||||
vport.dxf.height = h
|
||||
if h > 0:
|
||||
vport.dxf.aspect_ratio = w / h
|
||||
self._mark_dirty()
|
||||
return {
|
||||
"ok": True,
|
||||
"name": view.dxf.name,
|
||||
"center": [cx, cy],
|
||||
"height": h,
|
||||
"width": w,
|
||||
# The plan's literal. Measured: ezdxf's header refuses $VIEWCTR,
|
||||
# so the store is the *Active VPORT — what AutoCAD reads as the
|
||||
# initial view when it opens the file. No live display moves.
|
||||
"applied": "header_only",
|
||||
"store": "vport_active",
|
||||
"backend": "ezdxf",
|
||||
}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def view_named_list(self) -> list[dict]:
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
return [
|
||||
{
|
||||
"name": view.dxf.name,
|
||||
"center": [float(view.dxf.center.x), float(view.dxf.center.y)],
|
||||
"height": float(view.dxf.height),
|
||||
"width": float(view.dxf.width),
|
||||
}
|
||||
for view in doc.views
|
||||
]
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
# ── UCS ───────────────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _ucs_entry(doc, name: str):
|
||||
for ucs in doc.ucs:
|
||||
if ucs.dxf.name.lower() == name.lower():
|
||||
return ucs
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _set_ucs_header(doc, name: str, origin, x_axis, y_axis) -> None:
|
||||
doc.header["$UCSNAME"] = name
|
||||
doc.header["$UCSORG"] = tuple(origin)
|
||||
doc.header["$UCSXDIR"] = tuple(x_axis)
|
||||
doc.header["$UCSYDIR"] = tuple(y_axis)
|
||||
|
||||
@staticmethod
|
||||
def _ucs_row(name, origin, x_axis, y_axis, current: bool) -> dict:
|
||||
return {
|
||||
"name": name,
|
||||
"origin": [float(c) for c in origin],
|
||||
"x_axis": [float(c) for c in x_axis],
|
||||
"y_axis": [float(c) for c in y_axis],
|
||||
"current": current,
|
||||
}
|
||||
|
||||
async def ucs_list(self) -> list[dict]:
|
||||
from engineering.environment.ucs import WORLD
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
current = str(doc.header.get("$UCSNAME", "") or "")
|
||||
rows = [self._ucs_row(WORLD, (0, 0, 0), (1, 0, 0), (0, 1, 0), current == "")]
|
||||
for ucs in doc.ucs:
|
||||
rows.append(
|
||||
self._ucs_row(
|
||||
ucs.dxf.name,
|
||||
ucs.dxf.origin,
|
||||
ucs.dxf.xaxis,
|
||||
ucs.dxf.yaxis,
|
||||
current != "" and ucs.dxf.name.lower() == current.lower(),
|
||||
)
|
||||
)
|
||||
return rows
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def ucs_set(self, name, origin, x_axis, y_axis) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.ucs import WORLD, resolve_ucs_axes
|
||||
|
||||
clean = validate_name(name, what="UCS name")
|
||||
if clean.lower() == WORLD:
|
||||
raise ValueError("ucs_set: 'world' is reserved; ucs_restore('world') resets to WCS")
|
||||
axes = resolve_ucs_axes(origin, x_axis, y_axis) # refuses before any write
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
existing = self._ucs_entry(doc, clean)
|
||||
replaced = existing is not None
|
||||
stored = existing.dxf.name if replaced else clean
|
||||
if replaced:
|
||||
doc.ucs.remove(stored)
|
||||
doc.ucs.add(
|
||||
stored,
|
||||
dxfattribs={
|
||||
"origin": axes["origin"],
|
||||
"xaxis": axes["x_axis"],
|
||||
"yaxis": axes["y_axis"],
|
||||
},
|
||||
)
|
||||
self._set_ucs_header(doc, stored, axes["origin"], axes["x_axis"], axes["y_axis"])
|
||||
self._mark_dirty()
|
||||
return {
|
||||
"ok": True,
|
||||
"name": stored,
|
||||
"origin": list(axes["origin"]),
|
||||
"x_axis": list(axes["x_axis"]),
|
||||
"y_axis": list(axes["y_axis"]),
|
||||
"replaced": replaced,
|
||||
"current": True,
|
||||
"backend": "ezdxf",
|
||||
}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
async def ucs_restore(self, name) -> dict:
|
||||
from engineering.environment.names import validate_name
|
||||
from engineering.environment.ucs import WORLD
|
||||
|
||||
clean = validate_name(name, what="UCS name")
|
||||
|
||||
def _sync():
|
||||
doc = self._require_doc()
|
||||
if clean.lower() == WORLD:
|
||||
self._set_ucs_header(doc, "", (0, 0, 0), (1, 0, 0), (0, 1, 0))
|
||||
self._mark_dirty()
|
||||
return {"ok": True, "name": WORLD, "current": True, "backend": "ezdxf"}
|
||||
ucs = self._ucs_entry(doc, clean)
|
||||
if ucs is None:
|
||||
raise ValueError(
|
||||
f"ucs_restore: no UCS named {clean!r}; "
|
||||
f"saved: {[u.dxf.name for u in doc.ucs]} (or 'world')"
|
||||
)
|
||||
self._set_ucs_header(doc, ucs.dxf.name, ucs.dxf.origin, ucs.dxf.xaxis, ucs.dxf.yaxis)
|
||||
self._mark_dirty()
|
||||
return {"ok": True, "name": ucs.dxf.name, "current": True, "backend": "ezdxf"}
|
||||
|
||||
return await self._async(_sync)
|
||||
|
||||
# ── viewports ────────────────────────────────────────────────────────────
|
||||
|
||||
#: DXF viewport flag bit for "display locked" (VSF_LOCK_ZOOM).
|
||||
|
||||
Reference in New Issue
Block a user