feat(arch): the plan model and its foundation - layers, materials, EN/TR, two generalisations

Track F starts here; the three group branches are cut from this commit. - engineering/arch/model.py: Wall, Opening, Stair and Room as frozen dataclasses, and every refusal of spec 4 before anything is drawn, by path (walls[2].thickness): finite positive numbers, an axis of at least two distinct points with no zero-length segment, an arc segment refused by name rather than approximated, an opening on an unknown wall, past its wall, across an axis vertex or overlapping another (both ids named), an unknown material, and a typed room area - areas are measured, never typed. The plan travels as ACADMCP_ARCH XDATA; an opening's door/window kind and a stair's straight/l/u kind ride beside the payload kind as opening_kind / stair_kind, because one key cannot hold both. - engineering/arch/layers.py: the arch set - thirteen ISO 13567-style layers with ISO 128 lineweights only (0.05 mm on the construction scratch layer, as every set in this repository) - and ARCH_ROLE_LAYER, the one place a role becomes a layer. Registered in engineering/layers.py last, which is what keeps the engineering.layers <-> engineering.arch import cycle closed. - engineering/arch/materials.py: eight wall materials, one pattern each, every row citing its source: ANSI31 for brick (ISO 128-50 general hatching, acadiso.pat's own "Iron, Brick, Stone masonry"), AAC as ANSI31 at twice the spacing (a repository convention, and it says so), AR-CONC / AR-SAND as AutoCAD patterns, and the JIS A 0150 RC / stone / wood symbols. Reinforced concrete cannot be ANSI31 and AR-CONC in one HATCH, so it takes the one symbol that names it. Periods on a 1:50 sheet are recomputed from ezdxf's own pattern definitions by the tests. normalise_material is the one reading of a material name, shared by the model and wall_hatch, so the model never refuses "Reinforced Concrete" that the hatch table resolves. - engineering/arch/lang.py: EN / TR vocabulary, D/W and K/P tag prefixes, and the decimal comma for lang="tr" (24,50 m2). - draw_prims(role_layer=) and the XDATA codec's app_id= / kinds=: both generalised with the mechanical defaults, and the mechanical suite passes unchanged against them - that is the regression gate. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
2026-09-23 22:27:00 +03:00
parent b012fdf1be
commit d5c5e90406
12 changed files with 1657 additions and 32 deletions
@@ -17,6 +17,7 @@ from __future__ import annotations

 import logging
 import math
+from collections.abc import Mapping
 from typing import Any

 import anyio
@@ -98,6 +99,49 @@ def _number(value, where: str, *, positive: bool = True) -> float:
 # -- primitives to entities --------------------------------------------------


+#: The keys a role map may carry for a HatchArea, which has no role of its own:
+#: ``hatch`` (a general map) or ``wall_hatch`` (the architectural map, whose
+#: only hatch is wall poche). The first one the map carries wins.
+_HATCH_ROLES: tuple[str, ...] = ("hatch", "wall_hatch")
+
+
+def _layer_of(prim, role_layer: Mapping[str, str] | None) -> str:
+    """The layer a primitive lands on: the mechanical ROLE_LAYER, or ``role_layer``."""
+    if role_layer is None:
+        return layer_for(prim)
+    if isinstance(prim, HatchArea):
+        for key in _HATCH_ROLES:
+            if key in role_layer:
+                return role_layer[key]
+        raise ValueError(
+            "draw_prims: this role map has no hatch layer (a 'hatch' or 'wall_hatch' key); "
+            f"its roles are {', '.join(role_layer)}."
+        )
+    role = getattr(prim, "role", "visible")
+    try:
+        return role_layer[role]
+    except KeyError:
+        raise ValueError(
+            f"draw_prims: unknown role {role!r}; this role map has {', '.join(role_layer)}."
+        ) from None
+
+
+def _hatch_style(material: str, role_layer: Mapping[str, str] | None) -> dict:
+    """The pattern a HatchArea is filled with.
+
+    The architectural map (the one that carries ``wall_hatch``) resolves wall
+    materials through ``engineering/arch/materials.py::wall_hatch`` - brick,
+    aac, reinforced concrete ... at plan scale. Every other map, and no map,
+    keeps the mechanical ISO 128-50 section map, so a mechanical caller is
+    unchanged.
+    """
+    if role_layer is not None and "wall_hatch" in role_layer:
+        from engineering.arch.materials import wall_hatch
+
+        return wall_hatch(material)
+    return hatch_for(material)
+
+
 async def draw_prims(
    backend,
    prims,
@@ -105,11 +149,22 @@ async def draw_prims(
    at: tuple[float, float] = (0.0, 0.0),
    rotation: float = 0.0,
    layer_prefix: str = "",
+    role_layer: Mapping[str, str] | None = None,
 ) -> dict:
-    """Draw primitive descriptions, turned by ``rotation`` about their origin, then placed at ``at``."""
+    """Draw primitive descriptions, turned by ``rotation`` about their origin, then placed at ``at``.
+
+    ``role_layer`` maps ``Prim.role`` to a layer; ``None`` keeps the mechanical
+    ``ROLE_LAYER``. Every role and every hatch material is resolved - and every
+    target layer ensured - before the first entity, so an unknown role or
+    material refuses with nothing drawn.
+    """
    placed = tuple(prims)
    _refuse_islands_without_edge_paths(backend, placed)
-    await _ensure_layers(backend, sorted({f"{layer_prefix}{layer_for(p)}" for p in placed}))
+    targets = [f"{layer_prefix}{_layer_of(p, role_layer)}" for p in placed]
+    for prim in placed:
+        if isinstance(prim, HatchArea):
+            _hatch_style(prim.material, role_layer)
+    await _ensure_layers(backend, sorted(set(targets)))
    if abs(float(rotation)) > 1e-12:
        placed = rotate(placed, float(rotation), (0.0, 0.0))
    if abs(float(at[0])) > 1e-12 or abs(float(at[1])) > 1e-12:
@@ -118,8 +173,7 @@ async def draw_prims(
    handles: list[str] = []
    counts: dict[str, int] = {}

-    for prim in placed:
-        layer = f"{layer_prefix}{layer_for(prim)}"
+    for prim, layer in zip(placed, targets, strict=True):
        if isinstance(prim, Line):
            info = await backend.entity_create_line(
                prim.p1[0], prim.p1[1], prim.p2[0], prim.p2[1], layer=layer
@@ -151,7 +205,7 @@ async def draw_prims(
                layer=layer,
            )
        elif isinstance(prim, HatchArea):
-            style = hatch_for(prim.material)
+            style = _hatch_style(prim.material, role_layer)
            info = await backend.entity_create_hatch(
                style["pattern"],
                [[x, y] for x, y in prim.loops[0]],
@@ -207,10 +261,32 @@ async def _ensure_layers(backend, names) -> tuple[str, ...]:
    Group A measured the same refusal for annotation layers; this is the same
    cure (``ensure_annotation_layers``) at the places every mechanical drawing
    path passes through.
+
+    A layer of the ``arch`` set is created with its own ``ARCH_LAYERS`` row -
+    a grid on CENTER at 0.18 mm, a wall outline at 0.50 mm - rather than as a
+    bare white Continuous layer; every other name goes through
+    ``ensure_annotation_layers`` exactly as before.
    """
+    from engineering.arch.layers import ARCH_LAYER_DEFS
    from engineering.mech.annotate import ensure_annotation_layers

-    return await ensure_annotation_layers(backend, [str(name) for name in names if name])
+    wanted = [str(name) for name in names if name]
+    architectural = [name for name in wanted if name in ARCH_LAYER_DEFS]
+    created: list[str] = []
+    if architectural:
+        existing = {lyr.name.lower() for lyr in await backend.layer_list()}
+        for name in architectural:
+            if name.lower() in existing:
+                continue
+            color, linetype, lineweight = ARCH_LAYER_DEFS[name]
+            await backend.layer_create(
+                name=name, color=color, linetype=linetype, lineweight=lineweight
+            )
+            existing.add(name.lower())
+            created.append(name)
+    rest = [name for name in wanted if name not in ARCH_LAYER_DEFS]
+    created.extend(await ensure_annotation_layers(backend, rest))
+    return tuple(created)


 def _refuse_islands_without_edge_paths(backend, prims) -> None:
@@ -15,6 +15,12 @@ written, and both the writer and every reader test them literally:
    {"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", ...]}
+
+The codec is shared: ``app_id=`` / ``kinds=`` let another application write
+its own payloads through the same chunking and the same refusals (track F
+writes ``ACADMCP_ARCH`` with the kinds wall / opening / stair / room through
+`engineering/arch/model.py`). The defaults are the mechanical ones, so every
+mechanical caller is unchanged.
 """

 from __future__ import annotations
@@ -30,46 +36,48 @@ CHUNK = 255
 KINDS: tuple[str, ...] = ("part", "std_part", "balloon")


-def _validate(payload: dict) -> dict:
+def _validate(payload: dict, app_id: str, kinds: Sequence[str]) -> dict:
    if not isinstance(payload, dict):
-        raise TypeError(f"{APP_ID}: a payload must be a dict, got {type(payload).__name__}")
+        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"{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)}")
+    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)
+def encode(
+    payload: dict, *, app_id: str = APP_ID, kinds: Sequence[str] = KINDS
+) -> list[tuple[int, str]]:
+    """``[(1001, app_id), (1000, chunk), ...]`` — refused before anything is written."""
+    checked = _validate(payload, app_id, kinds)
    text = json.dumps(checked, separators=(",", ":"), sort_keys=True, ensure_ascii=True)
-    rows: list[tuple[int, str]] = [(1001, APP_ID)]
+    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])
+    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 "
+            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 to_values(payload: dict, *, app_id: str = APP_ID, kinds: Sequence[str] = KINDS) -> list[str]:
+    """Just the chunk strings — what ``entity_set_xdata(handle, app_id, values)`` takes."""
+    return [value for code, value in encode(payload, app_id=app_id, kinds=kinds) if code == 1000]


-def decode(rows: Sequence) -> dict:
+def decode(rows: Sequence, *, app_id: str = APP_ID, kinds: Sequence[str] = KINDS) -> 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")
+        raise ValueError(f"{app_id}: empty XDATA")
    parts: list[str] = []
    for row in rows:
        if isinstance(row, str):
@@ -78,20 +86,20 @@ def decode(rows: Sequence) -> dict:
        try:
            code, value = row
        except (TypeError, ValueError):
-            raise ValueError(f"{APP_ID}: {row!r} is not an XDATA (code, value) row") from None
+            raise ValueError(f"{app_id}: {row!r} is not an XDATA (code, value) row") from None
        if code == 1001:
-            if str(value) != APP_ID:
+            if str(value) != app_id:
                raise ValueError(
-                    f"{APP_ID}: this XDATA belongs to application {value!r}, not to us"
+                    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")
+            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
+        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)
+        raise ValueError(f"{app_id}: the payload decoded to {type(payload).__name__}, not a dict")
+    return _validate(payload, app_id, kinds)