From a0b16f399b7b9125b5c6f584769629c74997ec7d Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Tue, 15 Sep 2026 15:04:01 +0300 Subject: [PATCH] feat(pid): ACADMCP_PID payload encoding with 255-char chunking Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- engineering/pid/xdata.py | 93 ++++++++++++++++++++++++++++++++++++++++ tests/test_pid_xdata.py | 69 +++++++++++++++++++++++++++++ 2 files changed, 162 insertions(+) create mode 100644 engineering/pid/xdata.py create mode 100644 tests/test_pid_xdata.py diff --git a/engineering/pid/xdata.py b/engineering/pid/xdata.py new file mode 100644 index 0000000..3467a68 --- /dev/null +++ b/engineering/pid/xdata.py @@ -0,0 +1,93 @@ +"""``ACADMCP_PID`` XDATA payloads. + +XDATA strings are capped at 255 characters, so a JSON payload is stored as +``["json", chunk, chunk, ...]`` and re-joined on read. Every symbol INSERT and +every P&ID line carries one, so a reader without the catalogue (another tool, +a later catalogue version) still sees ports, classes and numbers. +""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from backends.base import AutoCADBackend + +APP_NAME = "ACADMCP_PID" +MARKER = "json" +CHUNK = 255 +PAYLOAD_VERSION = 1 + + +def encode_payload(payload: dict) -> list[str]: + text = json.dumps(payload, separators=(",", ":"), sort_keys=True, ensure_ascii=True) + return [MARKER] + [text[i : i + CHUNK] for i in range(0, len(text), CHUNK)] + + +def decode_payload(values: list) -> dict | None: + if not values or values[0] != MARKER: + return None + parts = [v for v in values[1:] if isinstance(v, str)] + try: + decoded = json.loads("".join(parts)) + except json.JSONDecodeError: + return None + return decoded if isinstance(decoded, dict) else None + + +async def write_payload(backend: AutoCADBackend, handle: str, payload: dict) -> dict: + return await backend.entity_set_xdata(handle, APP_NAME, encode_payload(payload)) + + +async def read_payload(backend: AutoCADBackend, handle: str) -> dict | None: + result = await backend.entity_get_xdata(handle, APP_NAME) + return decode_payload(result.get("xdata", {}).get(APP_NAME, [])) + + +def symbol_payload(spec: Any, params: dict | None = None) -> dict: + """``spec`` is an ``engineering.pid.symbols.SymbolSpec`` (typed loosely to + avoid an import cycle). Ports are symbol-local ``[x, y, direction, kind, radius]``.""" + from .symbols import CATALOG_VERSION + + return { + "v": PAYLOAD_VERSION, + "kind": "symbol", + "catalog": CATALOG_VERSION, + "name": spec.name, + "family": spec.family, + "symbol": spec.symbol, + "variant": spec.variant, + "params": dict(params or spec.params), + "ports": {p.name: [p.x, p.y, p.direction_deg, p.kind, p.radius] for p in spec.ports}, + } + + +def line_payload( + line_class: str, + number: str | None, + size: str | None, + service: str | None, + spec: str | None, + insulation: str | None, + seq: int | None, + from_ref: dict | None, + to_ref: dict | None, +) -> dict: + return { + "v": PAYLOAD_VERSION, + "kind": "line", + "class": line_class, + "number": number, + "size": size, + "service": service, + "spec": spec, + "insulation": insulation, + "seq": seq, + "from": from_ref, + "to": to_ref, + } + + +def marker_payload(line_handle: str) -> dict: + return {"v": PAYLOAD_VERSION, "kind": "marker", "line": line_handle} diff --git a/tests/test_pid_xdata.py b/tests/test_pid_xdata.py new file mode 100644 index 0000000..2ba3d92 --- /dev/null +++ b/tests/test_pid_xdata.py @@ -0,0 +1,69 @@ +"""ACADMCP_PID payloads survive the 255-character XDATA string limit.""" + +from __future__ import annotations + +import pytest + +from engineering.pid.xdata import ( + APP_NAME, + MARKER, + decode_payload, + encode_payload, + line_payload, + marker_payload, + read_payload, + write_payload, +) + +pytestmark = pytest.mark.asyncio + + +def test_short_payload_is_marker_plus_one_chunk(): + values = encode_payload({"v": 1, "kind": "marker", "line": "3A"}) + assert values[0] == MARKER and len(values) == 2 + assert decode_payload(values) == {"kind": "marker", "line": "3A", "v": 1} + + +def test_long_payload_is_chunked_at_255_and_rejoined(): + payload = {"v": 1, "kind": "symbol", "ports": {f"N{i}": [i, i, 90] for i in range(60)}} + values = encode_payload(payload) + assert all(len(v) <= 255 for v in values) + assert len(values) > 2 + assert decode_payload(values) == payload + + +def test_decode_rejects_foreign_or_broken_values(): + assert decode_payload([]) is None + assert decode_payload(["not-json", "{}"]) is None + assert decode_payload([MARKER, "{not json"]) is None + + +def test_builders_carry_the_documented_keys(): + line = line_payload( + "process_major", + "100-P-1001", + "100", + "P", + "CS1", + "IH", + 1001, + {"handle": "2F", "port": "discharge"}, + {"handle": "31", "port": "N1"}, + ) + assert line["kind"] == "line" and line["class"] == "process_major" + assert line["from"] == {"handle": "2F", "port": "discharge"} and line["seq"] == 1001 + assert marker_payload("3A") == {"v": 1, "kind": "marker", "line": "3A"} + + +async def test_roundtrip_through_a_backend(backend): + line = await backend.entity_create_line(0, 0, 10, 0) + payload = {"v": 1, "kind": "line", "class": "electric", "number": None} + await write_payload(backend, line.handle, payload) + assert await read_payload(backend, line.handle) == payload + got = await backend.entity_get_xdata(line.handle, APP_NAME) + assert got["xdata"][APP_NAME][0] == MARKER + + +async def test_read_payload_is_none_without_xdata(backend): + line = await backend.entity_create_line(0, 0, 10, 0) + assert await read_payload(backend, line.handle) is None