"""``ACADMCP_MECH`` XDATA: the part model written onto the drawing. The Track A codec (`engineering/pid/xdata.py`) with its limits made explicit. XDATA strings cap at 255 characters and an entity carries at most 16 KB across every application, so a payload is JSON-serialised ASCII-only and split into 1000 chunks under one 1001 application marker. `json.dumps(ensure_ascii=True)` escapes every newline, so no chunk can carry a line separator into the DXF. A payload that is not ours is *refused*, never decoded into something plausible: a foreign application marker, a chunk stream that is not JSON, a version this build does not read, or a kind outside `KINDS`. Three shapes are written, and both the writer and every reader test them literally: {"v": 1, "kind": "part", "id": "", "view": "front", "part": {...}} {"v": 1, "kind": "std_part", "designation": "ISO 4014 - M12x60", "standard": "ISO 4014", "size": "M12x60", "qty": 1, "material": "8.8"} {"v": 1, "kind": "balloon", "item": 3, "targets": ["1A2B", ...]} """ from __future__ import annotations import json from collections.abc import Sequence from backends.xdata_specs import MAX_BYTES, app_size APP_ID = "ACADMCP_MECH" PAYLOAD_VERSION = 1 CHUNK = 255 KINDS: tuple[str, ...] = ("part", "std_part", "balloon") def _validate(payload: dict) -> dict: if not isinstance(payload, dict): raise TypeError(f"{APP_ID}: a payload must be a dict, got {type(payload).__name__}") version = payload.get("v", PAYLOAD_VERSION) if version != PAYLOAD_VERSION: raise ValueError( f"{APP_ID}: payload version {version!r}; this build writes and reads " f"version {PAYLOAD_VERSION}" ) kind = payload.get("kind") if kind not in KINDS: raise ValueError(f"{APP_ID}: unknown payload kind {kind!r}; kinds are {', '.join(KINDS)}") return {**payload, "v": PAYLOAD_VERSION} def encode(payload: dict) -> list[tuple[int, str]]: """``[(1001, APP_ID), (1000, chunk), ...]`` — refused before anything is written.""" checked = _validate(payload) text = json.dumps(checked, separators=(",", ":"), sort_keys=True, ensure_ascii=True) rows: list[tuple[int, str]] = [(1001, APP_ID)] rows.extend((1000, text[i : i + CHUNK]) for i in range(0, len(text), CHUNK)) size = app_size(APP_ID, [(code, value) for code, value in rows if code != 1001]) if size > MAX_BYTES: raise ValueError( f"{APP_ID}: payload encodes to {size} bytes; the limit is 16 KB per entity " "across every application" ) return rows def to_values(payload: dict) -> list[str]: """Just the chunk strings — what ``entity_set_xdata(handle, APP_ID, values)`` takes.""" return [value for code, value in encode(payload) if code == 1000] def decode(rows: Sequence) -> dict: """``rows`` may be ``(code, value)`` pairs or the bare value list a backend returns from ``entity_get_xdata``. Raises ValueError on anything foreign.""" if not rows: raise ValueError(f"{APP_ID}: empty XDATA") parts: list[str] = [] for row in rows: if isinstance(row, str): parts.append(row) continue try: code, value = row except (TypeError, ValueError): raise ValueError(f"{APP_ID}: {row!r} is not an XDATA (code, value) row") from None if code == 1001: if str(value) != APP_ID: raise ValueError( f"{APP_ID}: this XDATA belongs to application {value!r}, not to us" ) continue if code != 1000: raise ValueError(f"{APP_ID}: group code {code} is not a payload chunk") parts.append(str(value)) try: payload = json.loads("".join(parts)) except json.JSONDecodeError as exc: raise ValueError(f"{APP_ID}: the chunk stream is not JSON ({exc.msg})") from exc if not isinstance(payload, dict): raise ValueError(f"{APP_ID}: the payload decoded to {type(payload).__name__}, not a dict") return _validate(payload)