feat(arch): arch_plan_from_spec, three arch_* critique focuses, the arch pack, lean 64 and the counts
The three track-F branches are merged; this is the integration that makes the
twelve SECTION 25 tools one surface.
- engineering/arch/spec.py: a whole plan from one document - walls with their
openings, stairs, room labels measured off the walls just drawn, the
exterior chains and the schedules - inside one transaction. Everything the
model, the wall engine or the stair symbol can refuse is refused before the
transaction opens, by path; a room point in no closed face rolls it back,
and so does a cancelled request, under the anyio shield of pid/spec.py.
EXAMPLE_SPEC's rooms measure 4825 x 5750 and 3825 x 5750 - the net floor,
computed by hand in the tests.
- engineering/arch/critique.py: arch_room_unlabelled, arch_wall_gap (50 mm,
one issue per pair of walls, measured to the wall body so an L corner and a
T stem both read as joined) and arch_opening_clash. One index per critique
run; a drawing without an ACADMCP_ARCH record yields nothing. Each focus
has a positive and a negative fixture.
- TOOL_PACK_NAMES gains "arch" (the twelve SECTION 25 tools); lean gains
arch_wall, arch_opening, arch_room and arch_plan_from_spec and reaches 64;
the arch tag ranks after mech, so SECTION 25 files as one `architecture`
group - 241 tools in 26 groups.
- Six golden tuning cases, two of them Turkish, the arch_plan_from_spec alias
record, the shared-command allowlist reconciled with what the merged records
claim, the inventory regenerated, the registry snapshot retaken, and the
README figures taken from check_doc_numbers rather than typed.
- Two more Turkish golden cases close the reconciliation's alias gaps
("vaziyet planı çiz" -> arch_plan_from_spec, "oda alanlarını hesapla" ->
arch_rooms_detect); the arch records carry vaziyet / alan / oda / kat planı
in diacritic and ASCII-folded spellings, because the search tokenizer splits
on a non-ASCII letter. BOUNDARY and AREA stay with boundary_trace and
analysis_measure_entity on purpose, recorded in the arch alias comment.
Tuning top-1 73/84, top-3 83/84; holdout top-3 15/15.
- tests/test_token_suite.py: TOOL_PACKS=core,mech now drops the arch pack
too (measured 193 = 236 - 9 - 22 - 12).
Co-Authored-By: Claude Opus 5.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.5
Claude Opus 5.5
parent
ad5a32a34a
commit
6e004cc658
@@ -0,0 +1,372 @@
|
||||
"""The three architectural critique focuses (spec §9).
|
||||
|
||||
Every check is pure over one index built once per critique run and shared
|
||||
through ``run_critique``'s context, the way the P&ID and mechanical focuses
|
||||
are. A drawing with no ``ACADMCP_ARCH`` record on it yields no issue at all: a
|
||||
mechanical sheet, a P&ID or a foreign plan of plain lines must not be scored
|
||||
against architectural conventions, and ``focus=None`` has to stay cheap for
|
||||
them.
|
||||
|
||||
What the index reads, and why only that: every plan record this server writes
|
||||
(a wall on its outline, an opening on its leaf or frame, a room on its label)
|
||||
lands on a layer of ``ARCH_ROLE_LAYER``, so XDATA is read on those layers only,
|
||||
every entity of them (paged, not the first 200). A record carried by several
|
||||
entities is kept once, by kind and id, with every handle that carries it. The
|
||||
room faces come from ``rooms_detect`` on the wall layer - the same reader
|
||||
``arch_rooms_detect`` runs, which closes door and window gaps from the opening
|
||||
records - so the focus and the tool cannot disagree about what a room is.
|
||||
|
||||
* ``arch_room_unlabelled`` (warning): a face of at least ``rooms_detect``'s
|
||||
minimum area (1 m²) with no room record whose label point lies inside it.
|
||||
* ``arch_wall_gap`` (error): a free wall end that comes within ``GAP_MM``
|
||||
(50 mm) of another wall - its end or its body - without joining it
|
||||
(``JOIN_TOL_MM``, 1 mm, the wall engine's ``join_tol``). The room leaks, and
|
||||
its area is measured through the gap. One issue per pair of walls.
|
||||
* ``arch_opening_clash`` (error): two openings overlapping on one wall, an
|
||||
opening running past the end of its wall's axis or across an axis vertex,
|
||||
and an opening whose host wall has no record on the drawing.
|
||||
|
||||
The two thresholds are declared by the design spec, not taken from a
|
||||
standard.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from collections.abc import Sequence
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from engineering.plan_spec import Issue
|
||||
|
||||
from .layers import ARCH_ROLE_LAYER
|
||||
from .model import APP_ID, Opening, Room, Stair, Wall, axis_segments, decode, from_payload
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from backends.base import AutoCADBackend
|
||||
|
||||
ARCH_FOCUSES = ("arch_room_unlabelled", "arch_wall_gap", "arch_opening_clash")
|
||||
_KEY = "arch_index"
|
||||
|
||||
#: Two wall ends closer than this without joining are a gap (spec §9).
|
||||
GAP_MM = 50.0
|
||||
#: Within this a wall end joins the wall it touches - the wall engine's join_tol.
|
||||
JOIN_TOL_MM = 1.0
|
||||
#: Below this, two lengths along a wall are the same length.
|
||||
EPS = 1e-9
|
||||
|
||||
_BUCKETS = {Wall: "walls", Opening: "openings", Stair: "stairs", Room: "rooms"}
|
||||
|
||||
|
||||
# ── the index ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def build_index(backend: AutoCADBackend) -> dict:
|
||||
"""One read of the drawing: every ACADMCP_ARCH record, and the room faces.
|
||||
|
||||
``{"walls" | "openings" | "stairs" | "rooms": {id: {"obj", "handles"}},
|
||||
"faces": [{"loop", "area", ...}]}``. The faces are read only when there is
|
||||
a wall record to bound them; a ``rooms_detect`` refusal (no wall layer)
|
||||
reads as no faces.
|
||||
"""
|
||||
from engineering.sheet.bom import all_entities
|
||||
|
||||
index: dict = {"walls": {}, "openings": {}, "stairs": {}, "rooms": {}, "faces": []}
|
||||
for layer in sorted(set(ARCH_ROLE_LAYER.values())):
|
||||
try:
|
||||
entities = await all_entities(backend, layer=layer)
|
||||
except Exception:
|
||||
continue
|
||||
for entity in entities:
|
||||
try:
|
||||
raw = await backend.entity_get_xdata(entity.handle, APP_ID)
|
||||
except Exception:
|
||||
continue
|
||||
values = (raw.get("xdata") or {}).get(APP_ID) or []
|
||||
if not values:
|
||||
continue
|
||||
try:
|
||||
obj = from_payload(decode(values))
|
||||
except (ValueError, KeyError, TypeError):
|
||||
# Not ours, or corrupt: the codec refused it, and guessing at
|
||||
# it here would undo that.
|
||||
continue
|
||||
bucket = index[_BUCKETS[type(obj)]]
|
||||
entry = bucket.setdefault(obj.id, {"obj": obj, "handles": []})
|
||||
entry["handles"].append(entity.handle)
|
||||
if index["walls"]:
|
||||
from engineering.arch.rooms import rooms_detect
|
||||
|
||||
try:
|
||||
detected = await rooms_detect(backend, layers=[ARCH_ROLE_LAYER["wall"]])
|
||||
except ValueError:
|
||||
detected = {"rooms": []}
|
||||
index["faces"] = list(detected.get("rooms") or [])
|
||||
return index
|
||||
|
||||
|
||||
def has_arch_content(index: dict) -> bool:
|
||||
return any(index[bucket] for bucket in ("walls", "openings", "stairs", "rooms"))
|
||||
|
||||
|
||||
# ── geometry ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _band(wall: Wall) -> tuple[float, float]:
|
||||
"""Where the two faces lie along the segment's left normal, low to high."""
|
||||
t = float(wall.thickness)
|
||||
if wall.justification == "left":
|
||||
return (0.0, t)
|
||||
if wall.justification == "right":
|
||||
return (-t, 0.0)
|
||||
return (-t / 2.0, t / 2.0)
|
||||
|
||||
|
||||
def body_distance(point, a, b, band: tuple[float, float]) -> float:
|
||||
"""Distance from ``point`` to the body of one straight wall segment.
|
||||
|
||||
The body is the rectangle between the segment's two faces, from ``a`` to
|
||||
``b``: 0 inside it or on its edge.
|
||||
"""
|
||||
length = math.dist(a, b)
|
||||
ux, uy = (b[0] - a[0]) / length, (b[1] - a[1]) / length
|
||||
dx, dy = point[0] - a[0], point[1] - a[1]
|
||||
along = dx * ux + dy * uy
|
||||
across = -dx * uy + dy * ux
|
||||
out_along = max(0.0 - along, 0.0, along - length)
|
||||
out_across = max(band[0] - across, 0.0, across - band[1])
|
||||
return math.hypot(out_along, out_across)
|
||||
|
||||
|
||||
def _ends(wall: Wall) -> list[tuple[str, tuple[float, float], int]]:
|
||||
"""The free ends of a wall: (which end, the point, the segment it belongs to)."""
|
||||
if wall.closed:
|
||||
return []
|
||||
last = len(wall.axis) - 2
|
||||
return [("start", wall.axis[0], 0), ("end", wall.axis[-1], last)]
|
||||
|
||||
|
||||
def wall_gaps(
|
||||
walls: Sequence[Wall], *, gap: float = GAP_MM, tol: float = JOIN_TOL_MM
|
||||
) -> list[dict]:
|
||||
"""Free wall ends within ``gap`` of another wall that do not join it.
|
||||
|
||||
An end joins a wall when it lies within ``tol`` of that wall's body (an L
|
||||
corner's shared end, a T stem ending in the crossing wall). The walls an
|
||||
end is measured against are every other wall, and its own wall's segments
|
||||
other than the one it ends and that segment's neighbours - a wall that
|
||||
nearly closes on itself leaks too. One row per pair of walls, the smallest
|
||||
gap kept: ``{"walls": (a, b), "end": "W2.start", "other": "W1", "gap_mm", "at"}``.
|
||||
"""
|
||||
found: dict[frozenset, dict] = {}
|
||||
for wall in walls:
|
||||
for end, point, own in _ends(wall):
|
||||
best: tuple[float, str] | None = None
|
||||
joined = False
|
||||
for other in walls:
|
||||
band = _band(other)
|
||||
segments = axis_segments(other)
|
||||
for index, (a, b) in enumerate(segments):
|
||||
if other.id == wall.id and abs(index - own) <= 1:
|
||||
continue
|
||||
distance = body_distance(point, a, b, band)
|
||||
if distance <= tol:
|
||||
joined = True
|
||||
break
|
||||
if best is None or distance < best[0]:
|
||||
best = (distance, other.id)
|
||||
if joined:
|
||||
break
|
||||
if joined or best is None or best[0] > gap:
|
||||
continue
|
||||
pair = frozenset((wall.id, best[1]))
|
||||
row = {
|
||||
"walls": tuple(sorted(pair)) if len(pair) == 2 else (wall.id, wall.id),
|
||||
"end": f"{wall.id}.{end}",
|
||||
"other": best[1],
|
||||
"gap_mm": round(best[0], 3),
|
||||
"at": (float(point[0]), float(point[1])),
|
||||
}
|
||||
if pair not in found or row["gap_mm"] < found[pair]["gap_mm"]:
|
||||
found[pair] = row
|
||||
return sorted(found.values(), key=lambda row: row["walls"])
|
||||
|
||||
|
||||
def opening_clashes(walls: Sequence[Wall], openings: Sequence[Opening]) -> list[dict]:
|
||||
"""Every opening its wall cannot host: ``{"openings": (ids), "wall", "reason"}``.
|
||||
|
||||
The same rules ``validate_openings`` refuses before a draw, collected
|
||||
rather than raised - the drawing already exists, and every clash on it is
|
||||
reported, not only the first.
|
||||
"""
|
||||
by_id = {wall.id: wall for wall in walls}
|
||||
rows: list[dict] = []
|
||||
spans: dict[str, list[tuple[float, float, str]]] = {}
|
||||
for opening in openings:
|
||||
wall = by_id.get(opening.wall)
|
||||
if wall is None:
|
||||
rows.append(
|
||||
{
|
||||
"openings": (opening.id,),
|
||||
"wall": opening.wall,
|
||||
"reason": f"its host wall {opening.wall!r} has no record on this drawing",
|
||||
}
|
||||
)
|
||||
continue
|
||||
start = float(opening.offset)
|
||||
end = start + float(opening.width)
|
||||
runs = []
|
||||
run = 0.0
|
||||
for a, b in axis_segments(wall):
|
||||
run += math.dist(a, b)
|
||||
runs.append(run)
|
||||
total = runs[-1]
|
||||
if end > total + EPS:
|
||||
rows.append(
|
||||
{
|
||||
"openings": (opening.id,),
|
||||
"wall": wall.id,
|
||||
"reason": f"it runs {end - total:g} mm past the end of the wall "
|
||||
f"({start:g}-{end:g} on an axis {total:g} long)",
|
||||
}
|
||||
)
|
||||
for vertex in runs[:-1]:
|
||||
if start + EPS < vertex < end - EPS:
|
||||
rows.append(
|
||||
{
|
||||
"openings": (opening.id,),
|
||||
"wall": wall.id,
|
||||
"reason": f"it runs across the axis vertex at {vertex:g} "
|
||||
f"({start:g}-{end:g})",
|
||||
}
|
||||
)
|
||||
for other_start, other_end, other_id in spans.get(wall.id, ()):
|
||||
if start < other_end - EPS and other_start < end - EPS:
|
||||
rows.append(
|
||||
{
|
||||
"openings": (other_id, opening.id),
|
||||
"wall": wall.id,
|
||||
"reason": f"they overlap ({other_start:g}-{other_end:g} and "
|
||||
f"{start:g}-{end:g})",
|
||||
}
|
||||
)
|
||||
spans.setdefault(wall.id, []).append((start, end, opening.id))
|
||||
return rows
|
||||
|
||||
|
||||
def point_in_loop(point, loop) -> bool:
|
||||
"""Even-odd ray cast; a point on an edge counts as inside."""
|
||||
x, y = float(point[0]), float(point[1])
|
||||
pts = [(float(p[0]), float(p[1])) for p in loop]
|
||||
inside = False
|
||||
for i, (x1, y1) in enumerate(pts):
|
||||
x2, y2 = pts[(i + 1) % len(pts)]
|
||||
cross = (x2 - x1) * (y - y1) - (y2 - y1) * (x - x1)
|
||||
if (
|
||||
abs(cross) <= 1e-9 * max(1.0, math.dist((x1, y1), (x2, y2)))
|
||||
and min(x1, x2) - 1e-9 <= x <= max(x1, x2) + 1e-9
|
||||
and min(y1, y2) - 1e-9 <= y <= max(y1, y2) + 1e-9
|
||||
):
|
||||
return True
|
||||
if (y1 > y) != (y2 > y):
|
||||
if x < x1 + (y - y1) * (x2 - x1) / (y2 - y1):
|
||||
inside = not inside
|
||||
return inside
|
||||
|
||||
|
||||
def unlabelled_faces(faces: Sequence[dict], rooms: Sequence[Room]) -> list[dict]:
|
||||
"""The faces no room record's label point lies in."""
|
||||
return [
|
||||
face for face in faces if not any(point_in_loop(room.at, face["loop"]) for room in rooms)
|
||||
]
|
||||
|
||||
|
||||
# ── the three checks ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _objs(index: dict, bucket: str) -> list:
|
||||
return [entry["obj"] for entry in index[bucket].values()]
|
||||
|
||||
|
||||
def _handles(index: dict, bucket: str, ids) -> list[str]:
|
||||
return [h for i in ids if i in index[bucket] for h in index[bucket][i]["handles"]]
|
||||
|
||||
|
||||
def _room_unlabelled(index: dict) -> list[Issue]:
|
||||
out: list[Issue] = []
|
||||
for face in unlabelled_faces(index["faces"], _objs(index, "rooms")):
|
||||
centroid = face.get("centroid") or face["loop"][0]
|
||||
out.append(
|
||||
Issue(
|
||||
"warning",
|
||||
"arch_room_unlabelled",
|
||||
f"A closed face of {float(face['area']) / 1.0e6:.2f} m² around "
|
||||
f"({float(centroid[0]):.0f}, {float(centroid[1]):.0f}) carries no room label; "
|
||||
"label it with arch_room so its measured area reaches the room schedule.",
|
||||
[],
|
||||
{"area_mm2": float(face["area"]), "centroid": [float(c) for c in centroid]},
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _wall_gap(index: dict) -> list[Issue]:
|
||||
out: list[Issue] = []
|
||||
for row in wall_gaps(_objs(index, "walls")):
|
||||
a, b = row["walls"]
|
||||
out.append(
|
||||
Issue(
|
||||
"error",
|
||||
"arch_wall_gap",
|
||||
f"Wall end {row['end']} stops {row['gap_mm']:g} mm short of wall "
|
||||
f"{row['other']} without joining it: the room leaks and its area is measured "
|
||||
"through the gap. Extend the wall to the joint (arch_wall) or move it clear.",
|
||||
_handles(index, "walls", sorted({a, b})),
|
||||
{"walls": [a, b], "gap_mm": row["gap_mm"], "at": list(row["at"])},
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _opening_clash(index: dict) -> list[Issue]:
|
||||
out: list[Issue] = []
|
||||
for row in opening_clashes(_objs(index, "walls"), _objs(index, "openings")):
|
||||
names = " and ".join(row["openings"])
|
||||
noun = "Openings" if len(row["openings"]) > 1 else "Opening"
|
||||
out.append(
|
||||
Issue(
|
||||
"error",
|
||||
"arch_opening_clash",
|
||||
f"{noun} {names} on wall {row['wall']}: {row['reason']}.",
|
||||
_handles(index, "openings", row["openings"]),
|
||||
{"openings": list(row["openings"]), "wall": row["wall"]},
|
||||
)
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
_CHECKS = {
|
||||
"arch_room_unlabelled": _room_unlabelled,
|
||||
"arch_wall_gap": _wall_gap,
|
||||
"arch_opening_clash": _opening_clash,
|
||||
}
|
||||
|
||||
|
||||
def issues_for(focus: str, index: dict) -> list[Issue]:
|
||||
if not has_arch_content(index):
|
||||
return []
|
||||
return _CHECKS[focus](index)
|
||||
|
||||
|
||||
def _make(focus: str):
|
||||
async def check(backend: AutoCADBackend, shared: dict) -> list[Issue]:
|
||||
index = shared.get(_KEY)
|
||||
if index is None:
|
||||
index = shared[_KEY] = await build_index(backend)
|
||||
return issues_for(focus, index)
|
||||
|
||||
check.needs_shared = True # type: ignore[attr-defined]
|
||||
check.__name__ = f"check_{focus}"
|
||||
return check
|
||||
|
||||
|
||||
ARCH_DISPATCH = {focus: _make(focus) for focus in ARCH_FOCUSES}
|
||||
@@ -0,0 +1,380 @@
|
||||
"""``arch_plan_from_spec``: a whole plan from one declarative document, in one transaction.
|
||||
|
||||
Walls with their hosted openings, stairs, room labels (each area measured off
|
||||
the walls just drawn), the exterior dimension chains and the schedules — in
|
||||
that order, because each step reads what the one before it drew: a room label
|
||||
measures the wall faces, and a schedule reads the ``ACADMCP_ARCH`` records.
|
||||
|
||||
Everything that can be refused without a drawing is refused before the
|
||||
transaction opens, by path (``walls[2].thickness``, ``openings[1]: ...``): the
|
||||
model, the openings against their walls, the junctions (the wall engine is
|
||||
pure and is run once in memory for exactly this), every stair's plan symbol,
|
||||
the rooms (a typed ``area`` included), the chain sides and the schedule kinds.
|
||||
What only the drawing can refuse - a room point that lies in no closed face -
|
||||
refuses inside the transaction, and the transaction is rolled back.
|
||||
|
||||
The rollback is shielded for the reason ``engineering/pid/spec.py`` documents
|
||||
at length: the MCP SDK cancels a request by cancelling an anyio scope, anyio
|
||||
cancellation is level-triggered, and a bare ``await`` in the handler would
|
||||
itself be cancelled before the rollback body ran - leaving half a plan inside
|
||||
an open transaction.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import math
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import anyio
|
||||
|
||||
from engineering.arch import draw as _draw
|
||||
from engineering.arch import rooms as _rooms
|
||||
from engineering.arch import schedule as _schedule
|
||||
|
||||
from .lang import LANGS
|
||||
from .model import (
|
||||
opening_from_dict,
|
||||
room_from_dict,
|
||||
stair_from_dict,
|
||||
validate_openings,
|
||||
wall_from_dict,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from backends.base import AutoCADBackend
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SPEC_KEYS = (
|
||||
"sheet",
|
||||
"lang",
|
||||
"poche",
|
||||
"walls",
|
||||
"openings",
|
||||
"stairs",
|
||||
"rooms",
|
||||
"chains",
|
||||
"schedules",
|
||||
)
|
||||
SHEET_KEYS = ("intent", "sheet_size", "scale")
|
||||
CHAIN_SIDES = ("bottom", "top", "left", "right")
|
||||
SCHEDULE_KINDS = ("doors", "windows", "rooms")
|
||||
|
||||
#: A two-room plan the tool docstring and the tests use: an entrance door and
|
||||
#: an interior door, two windows, a straight stair, both rooms labelled, the
|
||||
#: two exterior chains and all three schedules. The room numbers are "01" /
|
||||
#: "02", not "R1": the finalize validator reads R-and-a-digit as a radius
|
||||
#: callout typed as text.
|
||||
EXAMPLE_SPEC: dict = {
|
||||
"sheet": {"intent": "Two-room ground floor plan", "sheet_size": "A3", "scale": 50},
|
||||
"lang": "en",
|
||||
"walls": [
|
||||
{
|
||||
"id": "EXT",
|
||||
"axis": [[0, 0], [9000, 0], [9000, 6000], [0, 6000]],
|
||||
"thickness": 250,
|
||||
"closed": True,
|
||||
"material": "brick",
|
||||
},
|
||||
{"id": "INT", "axis": [[5000, 0], [5000, 6000]], "thickness": 100, "material": "aac"},
|
||||
],
|
||||
"openings": [
|
||||
{"id": "D1", "wall": "EXT", "kind": "door", "offset": 1000, "width": 1000, "tag": "D1"},
|
||||
{
|
||||
"id": "D2",
|
||||
"wall": "INT",
|
||||
"kind": "door",
|
||||
"offset": 2000,
|
||||
"width": 900,
|
||||
"hand": "right",
|
||||
"tag": "D2",
|
||||
},
|
||||
{
|
||||
"id": "W1",
|
||||
"wall": "EXT",
|
||||
"kind": "window",
|
||||
"offset": 20500,
|
||||
"width": 1500,
|
||||
"sill": 900,
|
||||
"height": 1200,
|
||||
"tag": "W1",
|
||||
},
|
||||
{
|
||||
"id": "W2",
|
||||
"wall": "EXT",
|
||||
"kind": "window",
|
||||
"offset": 11000,
|
||||
"width": 1200,
|
||||
"sill": 900,
|
||||
"height": 1200,
|
||||
"tag": "W2",
|
||||
},
|
||||
],
|
||||
"stairs": [
|
||||
{
|
||||
"id": "S1",
|
||||
"start": [6000, 1000],
|
||||
"direction_deg": 90,
|
||||
"width": 900,
|
||||
"risers": 16,
|
||||
"riser_height": 175,
|
||||
"going": 280,
|
||||
}
|
||||
],
|
||||
"rooms": [
|
||||
{"id": "R1", "name": "LIVING", "number": "01", "at": [2500, 3000]},
|
||||
{"id": "R2", "name": "HALL", "number": "02", "at": [7900, 3000]},
|
||||
],
|
||||
"chains": {"sides": ["bottom", "left"]},
|
||||
"schedules": [
|
||||
{"kind": "doors", "at": [10500, 6000]},
|
||||
{"kind": "windows", "at": [10500, 4000]},
|
||||
{"kind": "rooms", "at": [10500, 2000]},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _number(value, where: str) -> float:
|
||||
if isinstance(value, bool):
|
||||
raise ValueError(f"{where}: expected a number, got {value!r}")
|
||||
try:
|
||||
number = float(value)
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError(f"{where}: expected a number, got {value!r}") from None
|
||||
if not math.isfinite(number) or number <= 0.0:
|
||||
raise ValueError(f"{where}: must be a finite number greater than zero, got {value!r}")
|
||||
return number
|
||||
|
||||
|
||||
def _point(value, where: str) -> tuple[float, float]:
|
||||
if isinstance(value, (str, bytes)) or not isinstance(value, (list, tuple)) or len(value) != 2:
|
||||
raise ValueError(f"{where}: expected [x, y], got {value!r}")
|
||||
try:
|
||||
x, y = float(value[0]), float(value[1])
|
||||
except (TypeError, ValueError):
|
||||
raise ValueError(f"{where}: expected [x, y], got {value!r}") from None
|
||||
if not (math.isfinite(x) and math.isfinite(y)):
|
||||
raise ValueError(f"{where}: must be two finite numbers, got {value!r}")
|
||||
return (x, y)
|
||||
|
||||
|
||||
def _list(spec: dict, key: str) -> list:
|
||||
value = spec.get(key)
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, (str, bytes, dict)) or not isinstance(value, (list, tuple)):
|
||||
raise ValueError(f"{key}: expected a list, got {type(value).__name__}")
|
||||
return list(value)
|
||||
|
||||
|
||||
def validate_spec(spec: dict) -> dict:
|
||||
"""The whole document checked and normalised - nothing is drawn here.
|
||||
|
||||
Returns ``{"sheet", "lang", "scale", "poche", "walls", "openings",
|
||||
"stairs", "rooms", "chains", "schedules"}`` with model objects in place of
|
||||
dicts (rooms stay dicts: ``label_room`` takes the request, not a Room).
|
||||
"""
|
||||
from engineering.arch.stairs import stair_prims
|
||||
from engineering.arch.walls import wall_geometry
|
||||
|
||||
if not isinstance(spec, dict):
|
||||
raise ValueError(f"spec: expected a dict, got {type(spec).__name__}")
|
||||
unknown = sorted(set(spec) - set(SPEC_KEYS))
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"spec: unknown key(s) {', '.join(unknown)}; keys are {', '.join(SPEC_KEYS)}"
|
||||
)
|
||||
sheet = spec.get("sheet") or {}
|
||||
if not isinstance(sheet, dict):
|
||||
raise ValueError(f"sheet: expected a dict, got {type(sheet).__name__}")
|
||||
extra = sorted(set(sheet) - set(SHEET_KEYS))
|
||||
if extra:
|
||||
raise ValueError(
|
||||
f"sheet: unknown key(s) {', '.join(extra)}; keys are {', '.join(SHEET_KEYS)}"
|
||||
)
|
||||
scale = _number(sheet.get("scale", 50), "sheet.scale")
|
||||
lang = spec.get("lang", "en")
|
||||
if lang not in LANGS:
|
||||
raise ValueError(f"lang: unknown language {lang!r}; languages are {', '.join(LANGS)}")
|
||||
poche = spec.get("poche", True)
|
||||
if not isinstance(poche, bool):
|
||||
raise ValueError(f"poche: expected true or false, got {poche!r}")
|
||||
|
||||
raw_walls = _list(spec, "walls")
|
||||
if not raw_walls:
|
||||
raise ValueError("walls: name at least one wall - a plan without walls has no rooms")
|
||||
walls = [wall_from_dict(item, f"walls[{i}]") for i, item in enumerate(raw_walls)]
|
||||
openings = [
|
||||
opening_from_dict(item, f"openings[{i}]") for i, item in enumerate(_list(spec, "openings"))
|
||||
]
|
||||
try:
|
||||
validate_openings(walls, openings)
|
||||
geometry = wall_geometry(walls, openings, poche=poche)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"walls/openings: {exc}") from exc
|
||||
|
||||
stairs = []
|
||||
for i, item in enumerate(_list(spec, "stairs")):
|
||||
stair = stair_from_dict(item, f"stairs[{i}]")
|
||||
try:
|
||||
stair_prims(stair, lang=lang, scale=scale)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"stairs[{i}]: {exc}") from exc
|
||||
stairs.append(stair)
|
||||
|
||||
rooms = []
|
||||
seen_rooms: set[str] = set()
|
||||
for i, item in enumerate(_list(spec, "rooms")):
|
||||
room = room_from_dict(item, f"rooms[{i}]")
|
||||
if room.id in seen_rooms:
|
||||
raise ValueError(f"rooms[{i}].id: {room.id!r} is used twice")
|
||||
seen_rooms.add(room.id)
|
||||
rooms.append(dict(item))
|
||||
|
||||
chains = spec.get("chains")
|
||||
sides: tuple[str, ...] | None = None
|
||||
if chains not in (None, False):
|
||||
if chains is True:
|
||||
chains = {}
|
||||
if not isinstance(chains, dict) or set(chains) - {"sides"}:
|
||||
raise ValueError(f"chains: expected {{sides: [...]}}, got {chains!r}")
|
||||
sides = tuple(chains.get("sides") or ("bottom", "left"))
|
||||
bad = [side for side in sides if side not in CHAIN_SIDES]
|
||||
if bad or not sides:
|
||||
raise ValueError(f"chains.sides: {bad or sides!r} - sides are {', '.join(CHAIN_SIDES)}")
|
||||
|
||||
schedules = []
|
||||
for i, item in enumerate(_list(spec, "schedules")):
|
||||
if not isinstance(item, dict) or set(item) - {"kind", "at"}:
|
||||
raise ValueError(f"schedules[{i}]: expected {{kind, at}}, got {item!r}")
|
||||
kind = item.get("kind")
|
||||
if kind not in SCHEDULE_KINDS:
|
||||
raise ValueError(
|
||||
f"schedules[{i}].kind: {kind!r} is not a schedule; "
|
||||
f"schedules are {', '.join(SCHEDULE_KINDS)}"
|
||||
)
|
||||
schedules.append({"kind": kind, "at": _point(item.get("at"), f"schedules[{i}].at")})
|
||||
|
||||
return {
|
||||
"sheet": {
|
||||
"intent": str(sheet.get("intent") or "architectural plan"),
|
||||
"sheet_size": str(sheet.get("sheet_size") or "A3"),
|
||||
},
|
||||
"scale": scale,
|
||||
"lang": lang,
|
||||
"poche": poche,
|
||||
"walls": walls,
|
||||
"openings": openings,
|
||||
"omitted": list(geometry.omitted),
|
||||
"stairs": stairs,
|
||||
"rooms": rooms,
|
||||
"chains": sides,
|
||||
"schedules": schedules,
|
||||
}
|
||||
|
||||
|
||||
async def draw_plan_from_spec(backend: AutoCADBackend, spec: dict) -> dict:
|
||||
"""Validate, then draw the whole plan inside one transaction.
|
||||
|
||||
Returns ``{"walls", "openings", "stairs", "rooms", "chains", "schedules",
|
||||
"omitted", "drawn", "critique", "backend"}``: the wall and opening ids that
|
||||
were drawn (an opening the wall engine could not cut is in ``omitted``, not
|
||||
in ``openings``), each drawing step's own result (``drawn`` is
|
||||
``draw_walls``'), and the three ``arch_*`` critique focuses run over the
|
||||
committed drawing.
|
||||
"""
|
||||
from engineering.arch.critique import ARCH_FOCUSES, build_index, issues_for
|
||||
|
||||
norm = validate_spec(spec)
|
||||
scale = norm["scale"]
|
||||
lang = norm["lang"]
|
||||
if backend.get_plan_spec() is None:
|
||||
await backend.drawing_plan(
|
||||
norm["sheet"]["intent"], norm["sheet"]["sheet_size"], 1.0 / scale, "arch"
|
||||
)
|
||||
|
||||
begun = await backend.transaction_begin()
|
||||
if not isinstance(begun, dict) or not begun.get("ok"):
|
||||
# a refusal, not a crash: nothing has been drawn yet
|
||||
detail = begun.get("error") if isinstance(begun, dict) else begun
|
||||
raise ValueError(
|
||||
f"arch_plan_from_spec: could not open a transaction ({detail}); nothing was drawn. "
|
||||
"If one is already open, close it with transaction_commit or "
|
||||
"transaction_rollback and retry."
|
||||
)
|
||||
|
||||
result: dict = {
|
||||
"walls": None,
|
||||
"stairs": [],
|
||||
"rooms": [],
|
||||
"chains": None,
|
||||
"schedules": [],
|
||||
}
|
||||
try:
|
||||
await backend.drawing_apply_iso_layers("arch")
|
||||
result["walls"] = await _draw.draw_walls(
|
||||
backend, norm["walls"], norm["openings"], poche=norm["poche"]
|
||||
)
|
||||
for i, stair in enumerate(norm["stairs"]):
|
||||
try:
|
||||
result["stairs"].append(
|
||||
await _draw.draw_stair(backend, stair, lang=lang, scale=scale)
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"stairs[{i}]: {exc}") from exc
|
||||
for i, room in enumerate(norm["rooms"]):
|
||||
try:
|
||||
result["rooms"].append(
|
||||
await _rooms.label_room(backend, room, lang=lang, scale=scale)
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"rooms[{i}]: {exc}") from exc
|
||||
if norm["chains"] is not None:
|
||||
result["chains"] = await _draw.draw_chains(
|
||||
backend, norm["walls"], norm["openings"], sides=norm["chains"], scale=scale
|
||||
)
|
||||
for i, item in enumerate(norm["schedules"]):
|
||||
try:
|
||||
result["schedules"].append(
|
||||
await _schedule.draw_schedule(
|
||||
backend, item["kind"], at=item["at"], lang=lang, scale=scale
|
||||
)
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"schedules[{i}]: {exc}") from exc
|
||||
except BaseException as exc:
|
||||
cancelled = isinstance(exc, anyio.get_cancelled_exc_class())
|
||||
with anyio.CancelScope(shield=True):
|
||||
try:
|
||||
await backend.transaction_rollback()
|
||||
except Exception as rollback_exc:
|
||||
if not cancelled:
|
||||
raise
|
||||
log.error(
|
||||
"arch_plan_from_spec: rollback after a cancelled request failed (%s: %s); "
|
||||
"the plan may be half-drawn - check system_status and "
|
||||
"transaction_rollback before retrying",
|
||||
type(rollback_exc).__name__,
|
||||
rollback_exc,
|
||||
)
|
||||
raise
|
||||
await backend.transaction_commit()
|
||||
|
||||
index = await build_index(backend)
|
||||
critique = [issue.to_dict() for focus in ARCH_FOCUSES for issue in issues_for(focus, index)]
|
||||
drawn = result["walls"] or {}
|
||||
omitted = list(drawn.get("omitted", norm["omitted"]))
|
||||
skipped = {str(item.get("element")) for item in omitted}
|
||||
return {
|
||||
"walls": [wall.id for wall in norm["walls"]],
|
||||
"openings": [o.id for o in norm["openings"] if o.id not in skipped],
|
||||
"stairs": result["stairs"],
|
||||
"rooms": result["rooms"],
|
||||
"chains": result["chains"],
|
||||
"schedules": result["schedules"],
|
||||
"omitted": omitted,
|
||||
"drawn": drawn,
|
||||
"critique": critique,
|
||||
"backend": backend.name,
|
||||
}
|
||||
Reference in New Issue
Block a user