merge: track F group R

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
2026-09-23 23:03:13 +03:00
co-authored by Claude Opus 5.5
16 changed files with 2769 additions and 15 deletions
+3 -3
View File
@@ -27,8 +27,8 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
</div>
> **v1.5 release snapshot:** 233 tools · 8 resources · 5 prompt templates · 4005 collected tests.
> 233 is the **registered** count; a default install advertises 228 over `tools/list`,
> **v1.5 release snapshot:** 236 tools · 8 resources · 5 prompt templates · 4005 collected tests.
> 236 is the **registered** count; a default install advertises 231 over `tools/list`,
> because `ENABLE_3D` is unset. `system_about` is the runtime authority.
## Why this exists
@@ -120,7 +120,7 @@ Claude Desktop, Cursor, or any stdio MCP host. For HTTP: `autocad-mcp --transpor
| Quality loop | `drawing_preflight` → `drawing_plan` → `drawing_critique` → `drawing_refine` → `drawing_finalize` (0–100 score) |
| Delivery | `drawing_deliver`: DXF/PDF/PNG + SHA-256 manifest + reopen-parity checks |
<sub>233 tools in 26 groups; <code>TOOL_PACKS=core</code> hides the nine <code>pid_*</code> tools, the 22 environment tools and the 14 mechanical tools (<code>TOOL_PACKS=core,mech</code> keeps the last) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (<code>autocad://drawing/info</code>, <code>layers</code>, <code>blocks</code>, <code>entities/stats</code>, <code>entities/{layer_name}</code>, <code>system/status</code>, <code>pid/symbols</code>, <code>standards/isa51</code>) and 5 prompt templates.</sub>
<sub>236 tools in 26 groups; <code>TOOL_PACKS=core</code> hides the nine <code>pid_*</code> tools, the 22 environment tools and the 14 mechanical tools (<code>TOOL_PACKS=core,mech</code> keeps the last) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (<code>autocad://drawing/info</code>, <code>layers</code>, <code>blocks</code>, <code>entities/stats</code>, <code>entities/{layer_name}</code>, <code>system/status</code>, <code>pid/symbols</code>, <code>standards/isa51</code>) and 5 prompt templates.</sub>
**Two rules worth knowing.** Every coordinate in and out of a tool is WCS on both engines — the one exception is TEXT `rotation`, which stays in the entity frame because a mirrored TEXT is mirror-imaged and no scalar angle expresses that. And never read vertices back and shoelace them: that loses **28.2%** of the area on a semicircular edge, silently. `analysis_measure_entity(handle)` reads the real geometry and states its own accuracy.
+5 -4
View File
@@ -4792,14 +4792,15 @@ class ComBackend(AutoCADBackend):
)
for column, width in enumerate(layout.column_widths):
table.SetColumnWidth(column, width)
# MEASURED on AutoCAD 2026: IAcadTable has no TextHeight property
# (the old `table.TextHeight = h` raised and was swallowed, leaving
# the Standard style's 6.0 / 4.5 - illegible at 1:50). The member
# is SetCellTextHeight(row, col, h), set per cell.
for row_index, row in enumerate(layout.cells):
table.SetRowHeight(row_index, layout.row_height)
for column_index, value in enumerate(row):
table.SetText(row_index, column_index, value)
try:
table.TextHeight = layout.text_height
except Exception as exc:
log.debug("setting table TextHeight failed: %s", exc)
table.SetCellTextHeight(row_index, column_index, layout.text_height)
_apply_entity_attrs(table, layer, None, None)
_regen()
info = _entity_info(table)
+45 -2
View File
@@ -94,8 +94,9 @@ class ToolAliases:
# REVCLOUD -- one command draws a bare cloud (entity_create_revcloud) or a
# revision cloud that also files a revision row (revision_add); a
# drafter typing it could mean either.
# TABLE -- one command makes an empty table (entity_create_table) or the
# ISO 7573 parts list (bom_table); both are correct destinations.
# TABLE -- one command makes an empty table (entity_create_table), the
# ISO 7573 parts list (bom_table) or a door / window / room schedule
# (arch_schedule); all three are correct destinations.
# SECTIONPLANE -- the section a drafter asks for is either a whole section
# view of a drawn part (mech_view_add) or the cutting-plane line and its
# labels on an existing view (section_line); tracks B+G groups M and A
@@ -2304,6 +2305,48 @@ TOOL_ALIASES: dict[str, ToolAliases] = {
"aks olcusu",
),
),
# ── Architecture: rooms (SECTION 25, track F group R) ───────────────────
# AREA and BOUNDARY stay with analysis_measure_entity and boundary_trace:
# on a plain AutoCAD seat a drafter typing them means those tools, and
# sharing them would move the golden BPOLY / area cases. The room
# vocabulary below reaches these two without either command.
"arch_room": ToolAliases(
acad=(),
synonyms=(
"room label",
"label the room",
"room name and number",
"net floor area",
"measured room area",
"oda etiketi",
"oda alani",
"mahal adi",
),
),
"arch_rooms_detect": ToolAliases(
acad=(),
synonyms=(
"find the rooms",
"detect rooms in a plan",
"rooms of a floor plan",
"room areas of this plan",
"read rooms from lines",
"odalari bul",
"kat plani alanlari",
),
),
"arch_schedule": ToolAliases(
acad=("TABLE",),
synonyms=(
"door schedule",
"window schedule",
"room schedule",
"opening schedule",
"kapi cizelgesi",
"pencere cizelgesi",
"mahal listesi",
),
),
}
+44 -2
View File
@@ -3,14 +3,14 @@
"schema_version": 1,
"package_version": "1.5.1",
"totals": {
"tools": 233,
"tools": 236,
"resources": 8,
"prompts": 5,
"groups": 26
},
"groups": {
"analysis": 12,
"architecture": 4,
"architecture": 7,
"batch": 3,
"blocks": 9,
"corner_ops": 4,
@@ -209,6 +209,48 @@
"destructiveHint": false
}
},
{
"name": "arch_room",
"group": "architecture",
"cost": "mutate",
"summary": "Label a room with its name, number and measured area.",
"tags": [
"arch",
"create"
],
"annotations": {
"title": "Architecture: Label Room",
"destructiveHint": false
}
},
{
"name": "arch_rooms_detect",
"group": "architecture",
"cost": "read",
"summary": "Find the rooms of any plan and measure them; never modifies it.",
"tags": [
"arch",
"query"
],
"annotations": {
"title": "Architecture: Detect Rooms",
"readOnlyHint": true
}
},
{
"name": "arch_schedule",
"group": "architecture",
"cost": "mutate",
"summary": "Draw a door, window or room schedule from the plan records.",
"tags": [
"arch",
"create"
],
"annotations": {
"title": "Architecture: Schedule",
"destructiveHint": false
}
},
{
"name": "arch_stair",
"group": "architecture",
+24
View File
@@ -13,6 +13,7 @@ from .dimension import (
STEP_PAPER_MM,
exterior_chains,
)
from .faces import DEFAULT_TOL, Face, face_containing, planar_faces, tagged_faces
from .lang import LANGS, fmt_area_m2, fmt_number, tag_prefix, vocab
from .layers import ARCH_LAYER_DEFS, ARCH_LAYER_ROLES, ARCH_LAYERS, ARCH_ROLE_LAYER
from .materials import WALL_HATCH_TABLE, WALL_MATERIALS, wall_hatch
@@ -38,6 +39,15 @@ from .model import (
wall_length,
)
from .openings import opening_prims
from .rooms import (
detect_rooms,
label_room,
opening_closures,
read_records,
room_label_prims,
rooms_detect,
)
from .schedule import SCHEDULE_KINDS, draw_schedule, schedule_rows
from .stairs import (
ARROW_PAPER_MM,
BLONDEL_RANGE,
@@ -65,13 +75,16 @@ __all__ = [
"ARCH_ROLE_LAYER",
"ARROW_PAPER_MM",
"BLONDEL_RANGE",
"DEFAULT_TOL",
"FIRST_OFFSET_PAPER_MM",
"Face",
"KINDS",
"LANGS",
"MIN_JOIN_ANGLE_DEG",
"Opening",
"ROWS",
"Room",
"SCHEDULE_KINDS",
"SIDES",
"STAIR_KINDS",
"STEP_PAPER_MM",
@@ -85,20 +98,31 @@ __all__ = [
"axis_segments",
"blondel",
"decode",
"detect_rooms",
"draw_schedule",
"encode",
"exterior_chains",
"face_containing",
"face_offsets",
"fmt_area_m2",
"fmt_number",
"from_payload",
"label_room",
"opening_closures",
"opening_from_dict",
"opening_prims",
"planar_faces",
"read_records",
"room_from_dict",
"room_label_prims",
"room_segments",
"rooms_detect",
"schedule_rows",
"segment_of",
"stair_from_dict",
"stair_prims",
"tag_prefix",
"tagged_faces",
"to_payload",
"to_values",
"validate_openings",
+398
View File
@@ -0,0 +1,398 @@
"""The planar-face finder: straight segments in, the closed faces they bound out.
A room is the floor the walls enclose, so its area is a *face* of the planar
graph the wall outlines draw - never the axis rectangle and never a number the
caller types. This module finds those faces from plain segments, which is what
lets `arch_room` measure a room and `arch_rooms_detect` read a foreign plan made
of LINEs on both engines: the live engine has no `boundary_trace`, and it does
not need one.
The algorithm, in the order it runs:
1. **Merge.** Endpoints closer than ``tol`` become one vertex (the first one
seen keeps its coordinates), so a corner drawn 0.4 mm short still closes.
2. **Split.** Every segment is split at every point where another one crosses
it and at every vertex that lies on it within ``tol`` - the T-touch, where a
partition wall stops on the face of the wall it meets. Collinear overlaps
fall out of the same rule: each overlapping end splits the other segment and
the coincident pieces become one edge.
3. **Prune.** An edge that ends in a vertex of degree one bounds nothing - a
dangling line, an overshoot past a crossing - and is removed, repeatedly,
until no such edge is left.
4. **Walk.** Each undirected edge is two half-edges. From the half-edge
``u -> v`` the walk continues at ``v`` along the next edge *clockwise* from
``v -> u``, which keeps the face on the left. A bounded face comes out
counter-clockwise (positive signed area); the outer boundary of each
connected piece comes out clockwise.
5. **Nest.** A clockwise boundary that lies inside a bounded face of *another*
piece is a hole in the smallest such face: the island a column makes in a
room, or the whole inner outline of a ring of walls inside its outer outline.
A face's ``area`` is its loop's area minus its holes'.
Areas come from `engineering.measure.polygon_area_perimeter`, the same
arithmetic `analysis_measure_entity` reports; this module computes the *sign* of
the shoelace itself only to tell a face from a boundary, because the shared
function returns the magnitude by design.
Pairs of segments are tested against each other, which is quadratic: a floor
plan of a few thousand wall segments is well inside it, a city block is not.
"""
from __future__ import annotations
import math
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from engineering.measure import polygon_area_perimeter
Pt = tuple[float, float]
__all__ = ["DEFAULT_TOL", "Face", "face_containing", "planar_faces", "tagged_faces"]
#: Endpoints closer than this (drawing units, mm) are one vertex.
DEFAULT_TOL = 1.0
_EPS = 1e-12
@dataclass(frozen=True)
class Face:
"""One bounded face.
``loop`` is counter-clockwise and starts at its lowest, then leftmost,
vertex. ``holes`` are the clockwise outer boundaries of the pieces nested
inside it. ``area`` (mm^2) is the loop's area minus the holes'; ``centroid``
is the centroid of that net region.
"""
loop: tuple[Pt, ...]
area: float
centroid: Pt
holes: tuple[tuple[Pt, ...], ...] = ()
class _Vertices:
"""Snap points to vertices: a point within ``tol`` of a vertex *is* that vertex."""
def __init__(self, tol: float) -> None:
self.tol = tol
self.points: list[Pt] = []
self._grid: dict[tuple[int, int], list[int]] = {}
def _cell(self, p: Pt) -> tuple[int, int]:
return (math.floor(p[0] / self.tol), math.floor(p[1] / self.tol))
def index(self, p: Pt) -> int:
cx, cy = self._cell(p)
best: int | None = None
best_d = self.tol
for dx in (-1, 0, 1):
for dy in (-1, 0, 1):
for i in self._grid.get((cx + dx, cy + dy), ()):
d = math.dist(self.points[i], p)
if d <= best_d:
best, best_d = i, d
if best is not None:
return best
i = len(self.points)
self.points.append((float(p[0]), float(p[1])))
self._grid.setdefault((cx, cy), []).append(i)
return i
def _number(value, where: str) -> float:
try:
number = float(value)
except (TypeError, ValueError):
raise ValueError(f"{where}: expected a number, got {value!r}") from None
if not math.isfinite(number):
raise ValueError(f"{where}: must be finite, got {value!r}")
return number
def _point(value, where: str) -> Pt:
try:
x, y = value[0], value[1]
except (TypeError, IndexError, KeyError):
raise ValueError(f"{where}: expected [x, y], got {value!r}") from None
return (_number(x, f"{where}[0]"), _number(y, f"{where}[1]"))
def _segments(segments: Iterable) -> list[tuple[Pt, Pt, object]]:
"""``(p1, p2)`` or ``(p1, p2, tag)``; the tag rides along to the edges."""
out: list[tuple[Pt, Pt, object]] = []
for i, seg in enumerate(segments):
try:
count = len(seg)
except TypeError:
raise ValueError(f"segments[{i}]: expected (p1, p2) or (p1, p2, tag)") from None
if count not in (2, 3):
raise ValueError(
f"segments[{i}]: expected (p1, p2) or (p1, p2, tag), got {count} items"
)
tag = seg[2] if count == 3 else None
out.append((_point(seg[0], f"segments[{i}][0]"), _point(seg[1], f"segments[{i}][1]"), tag))
return out
def _cross(ax: float, ay: float, bx: float, by: float) -> float:
return ax * by - ay * bx
def _crossing(p1: Pt, p2: Pt, q1: Pt, q2: Pt) -> Pt | None:
"""Where two segments meet, ends included; None for parallel or apart."""
rx, ry = p2[0] - p1[0], p2[1] - p1[1]
sx, sy = q2[0] - q1[0], q2[1] - q1[1]
denom = _cross(rx, ry, sx, sy)
if abs(denom) <= _EPS * math.hypot(rx, ry) * math.hypot(sx, sy):
return None
qx, qy = q1[0] - p1[0], q1[1] - p1[1]
t = _cross(qx, qy, sx, sy) / denom
u = _cross(qx, qy, rx, ry) / denom
if -_EPS <= t <= 1.0 + _EPS and -_EPS <= u <= 1.0 + _EPS:
return (p1[0] + t * rx, p1[1] + t * ry)
return None
def _along(p: Pt, a: Pt, b: Pt) -> tuple[float, float]:
"""(distance along a->b of p's foot, distance of p from the line)."""
dx, dy = b[0] - a[0], b[1] - a[1]
length = math.hypot(dx, dy)
ux, uy = dx / length, dy / length
px, py = p[0] - a[0], p[1] - a[1]
return (px * ux + py * uy, abs(_cross(ux, uy, px, py)))
def _signed_area(points: Sequence[Pt]) -> float:
total = 0.0
for i in range(len(points)):
x1, y1 = points[i]
x2, y2 = points[(i + 1) % len(points)]
total += x1 * y2 - x2 * y1
return total / 2.0
def _area(points: Sequence[Pt]) -> float:
return polygon_area_perimeter([(x, y, 0.0) for x, y in points], True)[0]
def _centroid(points: Sequence[Pt]) -> tuple[Pt, float]:
"""(centroid, signed area) of a simple loop; a zero-area loop answers its vertex mean."""
signed = _signed_area(points)
if abs(signed) < _EPS:
n = len(points)
return (sum(p[0] for p in points) / n, sum(p[1] for p in points) / n), 0.0
cx = cy = 0.0
for i in range(len(points)):
x1, y1 = points[i]
x2, y2 = points[(i + 1) % len(points)]
w = x1 * y2 - x2 * y1
cx += (x1 + x2) * w
cy += (y1 + y2) * w
return (cx / (6.0 * signed), cy / (6.0 * signed)), signed
def _inside(point: Pt, loop: Sequence[Pt]) -> bool:
"""Even-odd ray cast; a point exactly on an edge may answer either way."""
x, y = point
inside = False
n = len(loop)
for i in range(n):
x1, y1 = loop[i]
x2, y2 = loop[(i + 1) % n]
if (y1 > y) != (y2 > y):
xs = x1 + (y - y1) * (x2 - x1) / (y2 - y1)
if xs > x:
inside = not inside
return inside
def _start_low_left(loop: list[int], points: list[Pt]) -> list[int]:
k = min(range(len(loop)), key=lambda i: (points[loop[i]][1], points[loop[i]][0]))
return loop[k:] + loop[:k]
def tagged_faces(
segments, *, tol: float = DEFAULT_TOL
) -> tuple[tuple[Face, tuple[frozenset, ...]], ...]:
"""`planar_faces`, plus, per face, the tag set of every edge it is bounded by.
An edge carries the tags of every input segment that covers it, so a reader
can ask whether a face is enclosed *only* by segments of a kind - which is
what `detect_rooms` turns into a confidence.
"""
tol = _number(tol, "tol")
if tol <= 0.0:
raise ValueError(f"tol: must be greater than zero, got {tol:g}")
raw = _segments(segments)
verts = _Vertices(tol)
# 1. merge endpoints
snapped: list[tuple[int, int, object]] = []
for p1, p2, tag in raw:
a, b = verts.index(p1), verts.index(p2)
if a != b:
snapped.append((a, b, tag))
# 2. split at crossings and at vertices lying on a segment
splits: list[set[int]] = [set() for _ in snapped]
boxes = []
for a, b, _tag in snapped:
(ax, ay), (bx, by) = verts.points[a], verts.points[b]
boxes.append((min(ax, bx) - tol, min(ay, by) - tol, max(ax, bx) + tol, max(ay, by) + tol))
for i in range(len(snapped)):
ai, bi, _ = snapped[i]
for j in range(i + 1, len(snapped)):
bx0, by0, bx1, by1 = boxes[j]
if boxes[i][0] > bx1 or bx0 > boxes[i][2] or boxes[i][1] > by1 or by0 > boxes[i][3]:
continue
aj, bj, _ = snapped[j]
hit = _crossing(verts.points[ai], verts.points[bi], verts.points[aj], verts.points[bj])
if hit is not None:
v = verts.index(hit)
splits[i].add(v)
splits[j].add(v)
for i, (a, b, _tag) in enumerate(snapped):
pa, pb = verts.points[a], verts.points[b]
length = math.dist(pa, pb)
x0, y0, x1, y1 = boxes[i]
for v, p in enumerate(verts.points):
if v in (a, b) or not (x0 <= p[0] <= x1 and y0 <= p[1] <= y1):
continue
s, off = _along(p, pa, pb)
if off <= tol and tol < s < length - tol:
splits[i].add(v)
edges: dict[tuple[int, int], set] = {}
for i, (a, b, tag) in enumerate(snapped):
pa, pb = verts.points[a], verts.points[b]
chain = (
[a]
+ sorted(
(v for v in splits[i] if v not in (a, b)),
key=lambda v: _along(verts.points[v], pa, pb)[0],
)
+ [b]
)
for u, v in zip(chain, chain[1:], strict=False):
if u != v:
edges.setdefault((min(u, v), max(u, v)), set()).add(tag)
# 3. prune dangling edges
adjacent: dict[int, set[int]] = {}
for u, v in edges:
adjacent.setdefault(u, set()).add(v)
adjacent.setdefault(v, set()).add(u)
loose = [v for v, ns in adjacent.items() if len(ns) == 1]
while loose:
v = loose.pop()
for w in list(adjacent.get(v, ())):
adjacent[w].discard(v)
adjacent[v].discard(w)
edges.pop((min(v, w), max(v, w)), None)
if len(adjacent[w]) == 1:
loose.append(w)
adjacent = {v: ns for v, ns in adjacent.items() if ns}
# 4. walk: the next half-edge is the next one clockwise at the far vertex
pts = verts.points
order: dict[int, list[int]] = {
v: sorted(ns, key=lambda w, v=v: math.atan2(pts[w][1] - pts[v][1], pts[w][0] - pts[v][0]))
for v, ns in adjacent.items()
}
seen: set[tuple[int, int]] = set()
cycles: list[list[int]] = []
for u, ns in adjacent.items():
for v in ns:
if (u, v) in seen:
continue
cycle: list[int] = []
a, b = u, v
while (a, b) not in seen:
seen.add((a, b))
cycle.append(a)
around = order[b]
a, b = b, around[around.index(a) - 1]
cycles.append(cycle)
component: dict[int, int] = {}
for start in adjacent:
if start in component:
continue
stack = [start]
component[start] = start
while stack:
v = stack.pop()
for w in adjacent[v]:
if w not in component:
component[w] = start
stack.append(w)
bounded: list[list[int]] = []
boundaries: list[list[int]] = []
for cycle in cycles:
signed = _signed_area([pts[v] for v in cycle])
if signed > _EPS:
bounded.append(_start_low_left(cycle, pts))
elif signed < -_EPS:
boundaries.append(cycle)
# 5. nest each piece's outer boundary in the smallest face of another piece
gross = [_area([pts[v] for v in loop]) for loop in bounded]
holes: list[list[list[int]]] = [[] for _ in bounded]
for boundary in boundaries:
probe = pts[boundary[0]]
home = None
for k, loop in enumerate(bounded):
if component[loop[0]] == component[boundary[0]]:
continue
if _inside(probe, [pts[v] for v in loop]) and (home is None or gross[k] < gross[home]):
home = k
if home is not None:
holes[home].append(_start_low_left(boundary, pts))
out: list[tuple[Face, tuple[frozenset, ...]]] = []
for k, loop in enumerate(bounded):
outer = [pts[v] for v in loop]
(gx, gy), g_signed = _centroid(outer)
net = gross[k]
mx, my, weight = gx * g_signed, gy * g_signed, g_signed
hole_loops = []
for hole in holes[k]:
ring = [pts[v] for v in hole]
(hx, hy), h_signed = _centroid(ring) # clockwise, so negative: it subtracts
net -= _area(ring)
mx, my, weight = mx + hx * h_signed, my + hy * h_signed, weight + h_signed
hole_loops.append(tuple(ring))
centroid = (mx / weight, my / weight) if abs(weight) > _EPS else (gx, gy)
tags = []
for ring in [loop, *holes[k]]:
for i in range(len(ring)):
u, v = ring[i], ring[(i + 1) % len(ring)]
tags.append(frozenset(edges[(min(u, v), max(u, v))]))
face = Face(loop=tuple(outer), area=net, centroid=centroid, holes=tuple(hole_loops))
out.append((face, tuple(tags)))
out.sort(key=lambda item: (item[0].loop[0][1], item[0].loop[0][0], item[0].area))
return tuple(out)
def planar_faces(segments, *, tol: float = DEFAULT_TOL) -> tuple[Face, ...]:
"""Every bounded face the segments enclose, CCW, net of nested pieces, in mm^2.
Refused by path, before anything is computed: a segment that is not
``(p1, p2)`` or ``(p1, p2, tag)``, a coordinate that is not a finite
number, a ``tol`` that is not positive.
"""
return tuple(face for face, _tags in tagged_faces(segments, tol=tol))
def face_containing(faces: Iterable[Face], point) -> Face | None:
"""The smallest face whose loop holds ``point`` and none of whose holes do."""
p = _point(point, "point")
found = [
face
for face in faces
if _inside(p, face.loop) and not any(_inside(p, hole) for hole in face.holes)
]
return min(found, key=lambda face: face.area) if found else None
+598
View File
@@ -0,0 +1,598 @@
"""Rooms: labels that carry a measured area, and a reader for plans that are not ours.
A room's area is the floor its walls enclose. `label_room` never takes it from
the caller: it reads the wall faces off the drawing, finds the planar face the
label point lies in (`engineering.arch.faces`) and writes *that* area into the
label and into the ``room`` record of ``ACADMCP_ARCH`` - so a room schedule is
a read of the drawing, and the number on the sheet is the number the geometry
gives.
A door or a window cuts its wall, so the two rooms either side of a door would
read as one face through the gap. The opening records on the drawing close it:
for each opening, the wall's two face lines are continued across the gap - in
memory only, never drawn - which is the convention a net floor area is measured
to (the wall face, not the reveal). A plan without our records has no such
lines, and `rooms_detect` says so rather than guessing where a door was.
`rooms_detect` runs the same finder on any lines and lightweight polylines -
a foreign plan - and reports each face with a confidence, the way `pid_graph`
does: 1.0 when every edge of the face lies on our wall layer, 0.6 when any
edge is a plain line from somewhere else. It never modifies the drawing. What
it cannot read - an arc, an old-style POLYLINE, walls inside a block
reference - is listed in ``skipped`` with its handle, never dropped and never
approximated, so a room it could not close is a room the answer says it
missed, not a wrong area.
DECLARED, not standard:
* The label text heights (3.5 mm name, 2.5 mm number and area, on paper) are
two values of the ISO 3098-1 series; which line gets which is this module's
choice. The line pitch of 1.6 x the text height is this module's choice too.
* ``MIN_ROOM_WIDTH``: a face whose mean width (2 x area / perimeter) is under
600 mm is taken for a wall body or an opening reveal, not a room. That is
this reader's heuristic, not a building rule - a 200 mm wall ring measures
200, a 900 mm corridor 5 m long measures 763.
"""
from __future__ import annotations
import math
from dataclasses import replace
from typing import TYPE_CHECKING
from engineering.arch.faces import DEFAULT_TOL, Face, face_containing, planar_faces, tagged_faces
from engineering.arch.lang import LANGS, fmt_area_m2
from engineering.arch.layers import ARCH_ROLE_LAYER
from engineering.arch.model import (
APP_ID,
Opening,
Room,
Stair,
Wall,
decode,
from_payload,
room_from_dict,
segment_of,
to_payload,
to_values,
)
from engineering.measure import polygon_area_perimeter
from engineering.mech.primitives import Pt, Text
if TYPE_CHECKING:
from backends.base import AutoCADBackend
__all__ = [
"CONFIDENCE_FOREIGN",
"CONFIDENCE_OURS",
"DEFAULT_MIN_AREA",
"LINE_PITCH",
"MIN_ROOM_WIDTH",
"NAME_HEIGHT",
"TEXT_HEIGHT",
"WALL_LAYER_KEYWORDS",
"detect_rooms",
"label_room",
"mean_width",
"opening_closures",
"read_records",
"read_segments",
"room_label_prims",
"rooms_detect",
]
NAME_HEIGHT = 3.5 # mm on paper (ISO 3098-1 series value)
TEXT_HEIGHT = 2.5 # mm on paper (ISO 3098-1 series value)
LINE_PITCH = 1.6 # baseline step, x the height of the line being placed
MIN_ROOM_WIDTH = 600.0 # mm; the wall-body heuristic above
DEFAULT_MIN_AREA = 1.0e6 # mm^2 = 1 m^2
CONFIDENCE_OURS = 1.0
CONFIDENCE_FOREIGN = 0.6
#: A layer whose name contains one of these (any case) is read by default.
WALL_LAYER_KEYWORDS = ("WALL", "DUVAR")
#: Entity types that bound a region with a curve; the finder reads straight
#: segments only, so these are reported as skipped, never approximated.
CURVE_TYPES = frozenset({"ARC", "CIRCLE", "ELLIPSE", "SPLINE"})
#: Polylines whose ``points`` are the vertices on both engines: the headless
#: LWPOLYLINE, and the live AcDbPolyline / AcDbLWPolyline (reported POLYLINE /
#: LWPOLYLINE). A headless POLYLINE is the old-style one - it arrives with no
#: ``points`` at all, and is skipped by that test.
LIGHT_POLYLINE_TYPES = frozenset({"LWPOLYLINE", "POLYLINE"})
#: Old-style polylines whose ``points`` are NOT to be trusted: the live
#: AcDb2dPolyline hands its Coordinates over as x, y, z triples, which the
#: backend reads two at a time (measured on AutoCAD 2026: a 4000 x 5000 outline
#: came back as [[200, 200], [0, 4200], [200, 0], ...]), so a room read off them
#: is a wrong number, not a missing one. Skipped, by name.
HEAVY_POLYLINE_TYPES = frozenset({"2DPOLYLINE", "3DPOLYLINE", "POLYFACEMESH", "POLYGONMESH"})
#: Block references on both engines (headless INSERT, live AcDbBlockReference).
BLOCK_REFERENCE_TYPES = frozenset({"INSERT", "BLOCKREFERENCE", "MINSERTBLOCK"})
#: Annotation and fills: they sit on wall layers (a poche hatch, a wall tag)
#: without bounding anything, so they are passed over without a skipped entry.
#: Every DIMENSION kind (the live engine names them ROTATEDDIMENSION, ...) is
#: matched by name below.
NON_BOUNDARY_TYPES = frozenset(
{
"TEXT",
"MTEXT",
"ATTDEF",
"ATTRIBUTEDEFINITION",
"ATTRIB",
"POINT",
"HATCH",
"LEADER",
"MLEADER",
"MULTILEADER",
"TOLERANCE",
"FCF",
}
)
_ARC_REASON = "the face finder reads straight segments only and does not approximate an arc"
def _check_lang(lang: str) -> str:
if lang not in LANGS:
raise ValueError(
f"lang: {lang!r} is not a label language; languages are {', '.join(LANGS)}"
)
return lang
def _positive(value, where: str) -> float:
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 room_label_prims(room: Room, *, lang: str = "en", scale=50) -> tuple[Text, ...]:
"""Name, number (when there is one) and measured area, stacked down from ``room.at``.
Heights are paper millimetres times the plot-scale denominator: at 1:50 the
3.5 mm name is 175 drawing units high. The first baseline is ``room.at``;
each next one is 1.6 x its own height lower.
"""
_check_lang(lang)
factor = _positive(scale, "scale")
x, y = float(room.at[0]), float(room.at[1])
lines = [(room.name, NAME_HEIGHT * factor)]
if room.number:
lines.append((str(room.number), TEXT_HEIGHT * factor))
lines.append((fmt_area_m2(room.area, lang), TEXT_HEIGHT * factor))
prims: list[Text] = []
baseline = y
for index, (text, height) in enumerate(lines):
if index:
baseline -= LINE_PITCH * height
prims.append(Text(at=(x, baseline), text=text, height=height, rotation=0.0, role="room"))
return tuple(prims)
def _face_offsets(wall: Wall) -> tuple[float, float]:
"""Signed distances of the wall's two faces from its axis, + to the left."""
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 opening_closures(walls, openings) -> tuple[tuple[tuple[Pt, Pt], ...], tuple[dict, ...]]:
"""The wall-face lines continued across every opening, and what could not be closed.
For an opening at ``offset`` of ``width`` on its wall, the segment of the
axis it sits on gives the direction ``u`` and the left normal ``n``; each
face line runs from ``start + n*d`` to ``start + u*width + n*d`` for the
face's signed offset ``d``. An opening whose wall has no record is returned
in ``omitted`` - its gap stays open - never guessed.
"""
by_id = {wall.id: wall for wall in walls}
closures: list[tuple[Pt, Pt]] = []
omitted: list[dict] = []
for opening in openings:
wall = by_id.get(opening.wall)
if wall is None:
omitted.append(
{
"element": opening.id,
"reason": f"host wall {opening.wall!r} has no record on this drawing; "
"its gap is left open",
}
)
continue
try:
index, along = segment_of(wall, float(opening.offset))
except ValueError as exc:
# a stale record (its wall was shortened since): the reader is
# read-only and answers for every other room, so it says so here
omitted.append({"element": opening.id, "reason": f"{exc}; its gap is left open"})
continue
axis = [(float(p[0]), float(p[1])) for p in wall.axis]
p0, p1 = axis[index], axis[(index + 1) % len(axis)]
length = math.dist(p0, p1)
ux, uy = (p1[0] - p0[0]) / length, (p1[1] - p0[1]) / length
nx, ny = -uy, ux
sx, sy = p0[0] + ux * along, p0[1] + uy * along
ex, ey = sx + ux * float(opening.width), sy + uy * float(opening.width)
for d in _face_offsets(wall):
closures.append(((sx + nx * d, sy + ny * d), (ex + nx * d, ey + ny * d)))
return tuple(closures), tuple(omitted)
def mean_width(face: Face) -> float:
"""2 x area / perimeter: the thickness of a long thin face, holes included."""
perimeter = sum(
polygon_area_perimeter([(x, y, 0.0) for x, y in ring], True)[1]
for ring in (face.loop, *face.holes)
)
return 2.0 * face.area / perimeter if perimeter > 0.0 else 0.0
def detect_rooms(
segments,
*,
min_area: float = DEFAULT_MIN_AREA,
tol: float = DEFAULT_TOL,
min_width: float = MIN_ROOM_WIDTH,
) -> tuple[dict, ...]:
"""Faces of at least ``min_area`` mm^2 and ``min_width`` mean width, with a confidence.
``segments`` are ``(p1, p2)`` or ``(p1, p2, layer)``. A face scores 1.0
when every edge of it (holes included) is covered by a segment on
``ARCH_ROLE_LAYER["wall"]``, and 0.6 otherwise - an untagged segment is a
plain line from somewhere else. The minimum of the inputs, never raised by
agreement: one foreign edge is enough to make the face 0.6.
"""
min_area = _positive(min_area, "min_area")
min_width = _positive(min_width, "min_width")
wall = ARCH_ROLE_LAYER["wall"].casefold()
rooms: list[dict] = []
for face, tags in tagged_faces(segments, tol=tol):
if face.area < min_area or mean_width(face) < min_width:
continue
ours = all(any(isinstance(t, str) and t.casefold() == wall for t in edge) for edge in tags)
rooms.append(
{
"loop": face.loop,
"holes": face.holes,
"area": face.area,
"centroid": face.centroid,
"confidence": CONFIDENCE_OURS if ours else CONFIDENCE_FOREIGN,
}
)
return tuple(rooms)
async def read_records(backend: AutoCADBackend) -> dict:
"""Every ``ACADMCP_ARCH`` record on the drawing's architectural layers.
The shape `read_plan` (Task 4) returns - ``walls``, ``openings``, ``stairs``,
``rooms`` as model dataclasses - plus ``room_handles`` (room id -> the
handle of the label that carries it). The current space is paged through
once, not sampled and not once per layer: a record past the 200th entity is
still a record, and a live seat, which pays a cross-process call per entity
per scan, scans once rather than once for each of the twelve arch layers.
Entities on the ``ARCH_ROLE_LAYER`` layers (any case) are read in layer-name
order, drawing order within a layer. XDATA that is not ours or does not
decode is skipped - this reader never guesses a record - and counted in
``unreadable``. A record that appears on several entities (a wall on each
of its outlines) is kept once, by kind and id.
"""
from engineering.sheet.bom import all_entities # lazy: sheet imports stay out of arch
found: dict[str, list] = {"walls": [], "openings": [], "stairs": [], "rooms": []}
room_handles: dict[str, str] = {}
seen: set[tuple[str, str]] = set()
unreadable = 0
buckets = {Wall: "walls", Opening: "openings", Stair: "stairs", Room: "rooms"}
arch_layers = {name.casefold(): name for name in ARCH_ROLE_LAYER.values()}
ours = [
entity
for entity in await all_entities(backend)
if str(entity.layer or "").casefold() in arch_layers
]
ours.sort(key=lambda entity: arch_layers[str(entity.layer).casefold()]) # stable
for entity in ours:
raw = await backend.entity_get_xdata(entity.handle, APP_ID)
values = (raw.get("xdata") or {}).get(APP_ID) or []
if not values:
continue
try:
obj = from_payload(decode(values))
except (ValueError, KeyError, TypeError):
obj = None
bucket = buckets.get(type(obj))
if bucket is None:
unreadable += 1
continue
key = (bucket, obj.id)
if key in seen:
continue
seen.add(key)
found[bucket].append(obj)
if bucket == "rooms":
room_handles[obj.id] = entity.handle
return {**found, "room_handles": room_handles, "unreadable": unreadable}
async def read_segments(backend: AutoCADBackend, layers) -> tuple[list, list[dict]]:
"""Straight segments on ``layers`` as ``(p1, p2, layer)``, and what was skipped.
Read: LINE, and the lightweight polyline by its ``points`` (the live engine
names it POLYLINE, the headless one LWPOLYLINE). Everything else that can
bound a room is listed in the second return value with its handle, type and
reason - never dropped, never guessed:
* an ARC/CIRCLE/ELLIPSE/SPLINE or a bulged polyline edge is an arc, and the
finder reads straight segments only - it is never flattened into chords;
* an old-style POLYLINE (``HEAVY_POLYLINE_TYPES`` live; a headless POLYLINE
reports no vertices) is not read, because its vertices do not reach this
reader correctly on either engine;
* a block reference is not exploded: walls inside a block are not read;
* any other type is listed as not a line.
Text, dimensions, leaders, points and hatches (``NON_BOUNDARY_TYPES``)
bound nothing and are passed over without an entry.
"""
from engineering.sheet.bom import all_entities # lazy: sheet imports stay out of arch
segments: list = []
skipped: list[dict] = []
def skip(entity, kind: str, reason: str) -> None:
skipped.append({"handle": entity.handle, "type": kind, "reason": reason})
for layer in layers:
for entity in await all_entities(backend, layer=layer):
props = entity.properties or {}
kind = str(entity.type).upper()
if kind in NON_BOUNDARY_TYPES or "DIMENSION" in kind:
continue
if kind == "LINE" and "start" in props and "end" in props:
segments.append((tuple(props["start"][:2]), tuple(props["end"][:2]), entity.layer))
elif kind in LIGHT_POLYLINE_TYPES and props.get("points"):
points = [tuple(p[:2]) for p in props["points"]]
bulges = list(props.get("bulges") or [])
count = len(points)
arcs = 0
for i in range(count if props.get("closed") else count - 1):
if i < len(bulges) and abs(float(bulges[i])) > 1e-12:
arcs += 1
continue
segments.append((points[i], points[(i + 1) % count], entity.layer))
if arcs:
skip(entity, kind, f"{arcs} arc segment(s): {_ARC_REASON}")
elif kind in CURVE_TYPES:
skip(entity, kind, f"a curve: {_ARC_REASON}")
elif kind in HEAVY_POLYLINE_TYPES or kind == "POLYLINE":
skip(
entity,
kind,
"an old-style polyline: its vertices are not read reliably by this "
"reader on either engine, so its edges are not counted; convert it to a "
"lightweight polyline (CONVERTPOLY) or explode it to lines",
)
elif kind in BLOCK_REFERENCE_TYPES:
skip(
entity,
kind,
"a block reference: walls inside a block are not read; explode it "
"(block_explode) to count them",
)
else:
skip(
entity,
kind,
"not a LINE or a lightweight polyline: the face finder does not read it",
)
return segments, skipped
def _wcs_loop(loop) -> list[list[float]]:
return [[float(x), float(y)] for x, y in loop]
async def label_room(
backend: AutoCADBackend, room_dict: dict, *, lang: str = "en", scale=50
) -> dict:
"""Label the room around ``room_dict["at"]`` with its name, number and measured area.
Refused before anything is drawn: an ``area`` in the request (areas are
measured, never typed), an unknown ``lang``, a non-positive ``scale``, an
``id`` already labelled on the drawing, a point that lies in no closed face
of the wall layer, a point inside a wall body, and a point in a face that
already holds a room record (one face, one room: a second record would be
counted twice by every schedule total). ``id`` defaults to the first free
``R<n>``.
"""
if not isinstance(room_dict, dict):
raise ValueError(
f"room: expected an object with name and at, got {type(room_dict).__name__}"
)
if "area" in room_dict:
raise ValueError(
"room.area: a room's area is measured from its walls, never typed; remove it"
)
_check_lang(lang)
_positive(scale, "scale")
records = await read_records(backend)
taken = {room.id for room in records["rooms"]}
data = dict(room_dict)
data.setdefault("number", None)
if not data.get("id"):
number = len(taken) + 1
while f"R{number}" in taken:
number += 1
data["id"] = f"R{number}"
room = room_from_dict(data, "room")
if room.id in taken:
raise ValueError(
f"room.id: {room.id!r} is already labelled on this drawing "
f"(label {records['room_handles'][room.id]}); choose another id"
)
wall_layer = ARCH_ROLE_LAYER["wall"]
segments, skipped = await read_segments(backend, [wall_layer])
closures, omitted = opening_closures(records["walls"], records["openings"])
faces = planar_faces(
[*segments, *((p1, p2, wall_layer) for p1, p2 in closures)], tol=DEFAULT_TOL
)
x, y = float(room.at[0]), float(room.at[1])
face = face_containing(faces, (x, y))
if face is None:
raise ValueError(
f"room.at: ({x:g}, {y:g}) lies in no closed face of the {len(segments)} wall "
f"segment(s) on {wall_layer}; close the walls around it first (arch_wall), or read "
"a plan drawn by other means with arch_rooms_detect"
)
width = mean_width(face)
if width < MIN_ROOM_WIDTH:
raise ValueError(
f"room.at: ({x:g}, {y:g}) lies inside a wall body (a face {width:.0f} mm wide on "
f"average, under the {MIN_ROOM_WIDTH:.0f} mm a room is taken to need), not in a room"
)
# One face is one room: a second record on it would be a second area in
# every schedule total (a retried call after a timeout is the usual way in).
already = [r for r in records["rooms"] if face_containing(faces, r.at) is face]
if already:
held = ", ".join(
f"{r.id} {r.name!r} (label {records['room_handles'][r.id]})" for r in already
)
raise ValueError(
f"room.at: ({x:g}, {y:g}) lies in a room that is already labelled - {held}; "
"one face carries one room record, so its area is counted once. Delete that "
"label to relabel the room, or place this one in another room"
)
measured = replace(room, area=face.area)
from engineering.mech.draw import draw_prims # lazy: mech.draw imports the arch layer set
drawn = await draw_prims(
backend, room_label_prims(measured, lang=lang, scale=scale), role_layer=ARCH_ROLE_LAYER
)
handle = drawn["handles"][0]
await backend.entity_set_xdata(handle, APP_ID, to_values(to_payload(measured)))
return {
"ok": True,
"room": {
"id": measured.id,
"name": measured.name,
"number": measured.number,
"at": [x, y],
"area": measured.area,
"area_m2": measured.area / 1.0e6,
"area_text": fmt_area_m2(measured.area, lang),
},
"handle": handle,
"handles": list(drawn["handles"]),
"layer": ARCH_ROLE_LAYER["room"],
"loop": _wcs_loop(face.loop),
"holes": [_wcs_loop(hole) for hole in face.holes],
"wall_segments": len(segments),
"closures": len(closures),
"skipped": skipped,
"omitted": list(omitted),
}
async def rooms_detect(
backend: AutoCADBackend,
*,
layers=None,
min_area: float = DEFAULT_MIN_AREA,
tol: float = DEFAULT_TOL,
) -> dict:
"""Read the rooms of any plan: faces of the lines on ``layers``, measured, never drawn.
``layers=None`` reads every layer whose name contains WALL or DUVAR (any
case). A named layer the drawing does not have is refused with the list of
the ones it does. Our own opening records, when there are any, close door
and window gaps exactly as `label_room` does; a foreign plan's gaps stay
open, and two rooms joined by an open doorway read as one face.
Each room reports ``labels`` - every room record whose point lies in its
face - and ``label``, the first of them or None. A face carrying more than
one record is listed in ``label_conflicts``, because a schedule reading the
records would count its area once per record. What `read_segments` could
not read is in ``skipped``, with its handle and why.
"""
min_area = _positive(min_area, "min_area")
tol = _positive(tol, "tol")
available = [layer.name for layer in await backend.layer_list()]
if layers is None:
chosen = [
name for name in available if any(key in name.upper() for key in WALL_LAYER_KEYWORDS)
]
if not chosen:
raise ValueError(
"layers: no layer name contains "
f"{' or '.join(WALL_LAYER_KEYWORDS)}; name the layers the walls are on "
f"(layers are {', '.join(sorted(available))})"
)
else:
lookup = {name.casefold(): name for name in available}
missing = [str(name) for name in layers if str(name).casefold() not in lookup]
if missing:
raise ValueError(
f"layers: {', '.join(missing)} not in this drawing; "
f"layers are {', '.join(sorted(available))}"
)
chosen = [lookup[str(name).casefold()] for name in layers]
segments, skipped = await read_segments(backend, chosen)
records = await read_records(backend)
closures, omitted = opening_closures(records["walls"], records["openings"])
wall_layer = ARCH_ROLE_LAYER["wall"]
extra = [(p1, p2, wall_layer) for p1, p2 in closures]
rooms = detect_rooms([*segments, *extra], min_area=min_area, tol=tol)
out: list[dict] = []
conflicts: list[dict] = []
for room in rooms:
face = Face(
loop=room["loop"], area=room["area"], centroid=room["centroid"], holes=room["holes"]
)
labels = [
{"id": r.id, "name": r.name, "number": r.number, "area": r.area}
for r in records["rooms"]
if face_containing((face,), r.at) is not None
]
if len(labels) > 1:
conflicts.append(
{
"labels": [label["id"] for label in labels],
"centroid": [room["centroid"][0], room["centroid"][1]],
"reason": f"one face carries {len(labels)} room records; a schedule "
f"would count its {room['area'] / 1.0e6:.2f} m2 {len(labels)} times",
}
)
out.append(
{
"loop": _wcs_loop(room["loop"]),
"holes": [_wcs_loop(hole) for hole in room["holes"]],
"area": room["area"],
"area_m2": room["area"] / 1.0e6,
"centroid": [room["centroid"][0], room["centroid"][1]],
"confidence": room["confidence"],
"label": labels[0] if labels else None,
"labels": labels,
}
)
return {
"ok": True,
"modified": False,
"layers": chosen,
"segments": len(segments),
"closures": len(extra),
"count": len(out),
"rooms": out,
"confidence_min": min((room["confidence"] for room in out), default=None),
"label_conflicts": conflicts,
"skipped": skipped,
"omitted": list(omitted),
"min_area": min_area,
"tol": tol,
}
+284
View File
@@ -0,0 +1,284 @@
"""Door, window and room schedules - a read of the drawing, drawn as a real TABLE.
`schedule_rows` turns plan data (the shape `read_plan` returns: ``walls``,
``openings``, ``rooms`` lists of model dataclasses) into rows; `draw_schedule`
reads that data off the drawing's ``ACADMCP_ARCH`` records and draws it through
the same `entity_create_table` path the ISO 7573 parts list takes
(`engineering/sheet/bom.py::draw_bom_table`): insertion at the top-left corner,
the table growing downwards, and the engine's representation reported rather
than assumed - a real ACAD_TABLE live, a composite of rules and MTEXT headless.
Rows:
* doors - tag, width, height, swing, hand, wall; windows - tag, width,
height, sill, wall. Sorted by tag, numbers compared as numbers (D2 before
D10). An opening without a tag gets the next free ``<prefix><n>`` in wall
order (the order of ``walls``, then the offset along the wall); an explicit
tag is never renumbered and a generated one never repeats it.
* rooms - number, name, area (the measured one the label carries), sorted by
number with the unnumbered last, plus a total row.
A value the model does not carry (a door without a height) is an empty cell,
never a default. Headers come from `vocab(lang)`; ``lang="tr"`` also writes
decimal commas.
DECLARED, not standard: the paper column widths below and the reuse of the
parts list's 7 mm row and 2.5 mm text (`engineering.sheet.bom.ROW_HEIGHT`,
`TEXT_HEIGHT`) - no standard fixes a schedule's layout.
"""
from __future__ import annotations
import math
import re
from typing import TYPE_CHECKING
from engineering.arch.lang import LANGS, fmt_area_m2, fmt_number, tag_prefix, vocab
from engineering.arch.layers import ARCH_LAYERS, ARCH_ROLE_LAYER
if TYPE_CHECKING:
from backends.base import AutoCADBackend
__all__ = [
"COLUMN_WIDTHS",
"MAX_ROWS",
"SCHEDULE_COLUMNS",
"SCHEDULE_KINDS",
"SCHEDULE_TITLES",
"draw_schedule",
"schedule_rows",
]
SCHEDULE_KINDS: tuple[str, ...] = ("doors", "windows", "rooms")
SCHEDULE_COLUMNS: dict[str, tuple[str, ...]] = {
"doors": ("tag", "width", "height", "swing", "hand", "wall"),
"windows": ("tag", "width", "height", "sill", "wall"),
"rooms": ("number", "name", "area"),
}
#: The `vocab` key of each schedule's title row.
SCHEDULE_TITLES: dict[str, str] = {
"doors": "door_schedule",
"windows": "window_schedule",
"rooms": "room_schedule",
}
#: Column widths in paper millimetres (x the scale denominator in the drawing).
COLUMN_WIDTHS: dict[str, float] = {
"tag": 15.0,
"width": 20.0,
"height": 20.0,
"swing": 20.0,
"hand": 20.0,
"sill": 20.0,
"wall": 20.0,
"number": 15.0,
"name": 45.0,
"area": 25.0,
}
#: `entity_create_table` takes at most 200 rows, the title and header included.
MAX_ROWS = 198
_OPENING_KIND = {"doors": "door", "windows": "window"}
def _check(kind: str, lang: str) -> None:
if kind not in SCHEDULE_KINDS:
raise ValueError(
f"kind: {kind!r} is not a schedule; schedules are {', '.join(SCHEDULE_KINDS)}"
)
if lang not in LANGS:
raise ValueError(
f"lang: {lang!r} is not a schedule language; languages are {', '.join(LANGS)}"
)
def _natural(text: str) -> tuple:
"""``D10`` after ``D2``: digit runs compare as numbers."""
return tuple(
(0, int(part), "") if part.isdigit() else (1, 0, part)
for part in re.findall(r"\d+|\D+", str(text))
)
def _mm(value, lang: str) -> str:
"""A length in mm: whole numbers without decimals, others to 0.1 mm; None is blank."""
if value is None:
return ""
number = float(value)
decimals = 0 if abs(number - round(number)) < 1e-9 else 1
return fmt_number(number, lang, decimals)
def _opening_rows(kind: str, plan: dict, lang: str) -> tuple[dict, ...]:
words = vocab(lang)
wanted = _OPENING_KIND[kind]
wall_order = {wall.id: index for index, wall in enumerate(plan.get("walls") or ())}
openings = [o for o in plan.get("openings") or () if o.kind == wanted]
openings.sort(
key=lambda o: (wall_order.get(o.wall, len(wall_order)), o.wall, float(o.offset), o.id)
)
prefix = tag_prefix(wanted, lang)
used = {str(o.tag) for o in openings if o.tag}
counter = 0
rows: list[dict] = []
for opening in openings:
tag = str(opening.tag) if opening.tag else None
generated = tag is None
if generated:
counter += 1
while f"{prefix}{counter}" in used:
counter += 1
tag = f"{prefix}{counter}"
used.add(tag)
row = {
"id": opening.id,
"tag": tag,
"generated": generated,
"width": _mm(opening.width, lang),
"height": _mm(opening.height, lang),
"wall": opening.wall,
}
if kind == "doors":
row["swing"] = words[opening.swing]
row["hand"] = words[opening.hand]
else:
row["sill"] = _mm(opening.sill, lang)
rows.append(row)
rows.sort(key=lambda row: _natural(row["tag"]))
return tuple(rows)
def _room_rows(plan: dict, lang: str) -> tuple[dict, ...]:
rooms = sorted(
plan.get("rooms") or (),
key=lambda r: (
r.number is None or str(r.number) == "",
_natural(r.number or ""),
r.name,
r.id,
),
)
if not rooms:
return ()
rows = [
{
"id": room.id,
"number": "" if room.number is None else str(room.number),
"name": room.name,
"area": fmt_area_m2(room.area, lang),
"area_mm2": float(room.area),
"total": False,
}
for room in rooms
]
total = math.fsum(float(room.area) for room in rooms)
rows.append(
{
"id": "",
"number": "",
"name": vocab(lang)["total"],
"area": fmt_area_m2(total, lang),
"area_mm2": total,
"total": True,
}
)
return tuple(rows)
def schedule_rows(kind: str, plan: dict, *, lang: str = "en") -> tuple[dict, ...]:
"""The rows of one schedule, each a dict holding every column of ``SCHEDULE_COLUMNS[kind]``
as display text, plus ``id`` (and ``generated`` for openings, ``area_mm2`` and
``total`` for rooms). Refused: an unknown kind or language, named with the list."""
_check(kind, lang)
if kind == "rooms":
return _room_rows(plan, lang)
return _opening_rows(kind, plan, lang)
async def _ensure_layer(backend: AutoCADBackend, name: str) -> None:
"""Create the schedule layer from its `ARCH_LAYERS` row when the drawing lacks it.
Live ActiveX will not create a layer on assignment (the repository rule),
so the layer exists before the table does.
"""
if name.casefold() in {layer.name.casefold() for layer in await backend.layer_list()}:
return
row = next(r for r in ARCH_LAYERS if r[0] == name)
await backend.layer_create(name=row[0], color=row[1], linetype=row[2], lineweight=row[3])
async def draw_schedule(
backend: AutoCADBackend, kind: str, *, at, lang: str = "en", scale=50
) -> dict:
"""Read the drawing's records and draw one schedule as a TABLE whose top-left is ``at``.
Refused before anything is drawn: an unknown kind or language, a scale that
is not a finite positive number, an ``at`` that is not two finite numbers,
a drawing with nothing to schedule, and more rows than one TABLE takes.
"""
from engineering.arch.rooms import read_records # the one ACADMCP_ARCH reader
from engineering.sheet.bom import ROW_HEIGHT, TEXT_HEIGHT
_check(kind, lang)
try:
factor = float(scale)
except (TypeError, ValueError):
raise ValueError(f"scale: expected a number, got {scale!r}") from None
if not math.isfinite(factor) or factor <= 0.0:
raise ValueError(f"scale: must be a finite number greater than zero, got {scale!r}")
try:
x, y = float(at[0]), float(at[1])
except (TypeError, ValueError, IndexError):
raise ValueError(f"at: expected [x, y], got {at!r}") from None
if not (math.isfinite(x) and math.isfinite(y)):
raise ValueError(f"at: must be two finite numbers, got {at!r}")
plan = await read_records(backend)
rows = schedule_rows(kind, plan, lang=lang)
if not rows:
raise ValueError(
f"kind: this drawing carries no {kind[:-1]} records to schedule "
"(ACADMCP_ARCH on the architectural layers); draw the plan first"
)
if len(rows) > MAX_ROWS:
raise ValueError(
f"kind: {len(rows)} {kind} rows do not fit one TABLE ({MAX_ROWS} data rows at most)"
)
words = vocab(lang)
columns = SCHEDULE_COLUMNS[kind]
layer = ARCH_ROLE_LAYER["symbol"]
await _ensure_layer(backend, layer)
entity = await backend.entity_create_table(
x,
y,
[[row[c] for c in columns] for row in rows],
headers=[words[c] for c in columns],
column_widths=[COLUMN_WIDTHS[c] * factor for c in columns],
row_height=ROW_HEIGHT * factor,
text_height=TEXT_HEIGHT * factor,
title=words[SCHEDULE_TITLES[kind]],
layer=layer,
)
width = sum(COLUMN_WIDTHS[c] for c in columns) * factor
height = (len(rows) + 2) * ROW_HEIGHT * factor
return {
"ok": True,
"kind": kind,
"lang": lang,
"handle": entity.handle,
"entity_type": entity.type,
"representation": entity.properties.get("representation", "native"),
"child_handles": list(entity.properties.get("child_handles") or []),
"layer": layer,
"columns": list(columns),
"headers": [words[c] for c in columns],
"title": words[SCHEDULE_TITLES[kind]],
"rows": [dict(row) for row in rows],
"row_count": len(rows),
"table_rows": entity.properties.get("rows"),
"insert": [x, y],
"bbox": {"min": [x, y - height], "max": [x + width, y]},
}
+156 -1
View File
@@ -9783,7 +9783,7 @@ async def drawing_export_dwg(
# ---------------------------------------------------------------------------
# ── SECTION 25: Architecture (4 tools) ──────────────────────────────────────
# ── SECTION 25: Architecture (7 tools) ──────────────────────────────────────
# ---------------------------------------------------------------------------
@@ -10068,6 +10068,161 @@ async def arch_dimension_chains(
# ── arch: rooms (group R) ──
@cad_tool(summary="Label a room with its name, number and measured area.", cost="mutate")
@mcp.tool(
annotations={"title": "Architecture: Label Room", "destructiveHint": False},
tags={"arch", "create"},
)
async def arch_room(
name: Annotated[str, Field(description="Room name, e.g. LIVING / SALON.")],
x: Annotated[
float, Field(description="WCS X of a point inside the room; the label starts here.")
],
y: Annotated[float, Field(description="WCS Y of a point inside the room.")],
number: Annotated[
str | None, Field(default=None, description="Room number, e.g. 01; omit for none.")
] = None,
room_id: Annotated[
str | None,
Field(default=None, description="Record id; default is the first free R<n>."),
] = None,
lang: Annotated[
str, Field(default="en", description="Label language: en | tr (tr writes 24,50 m²).")
] = "en",
scale: Annotated[
float,
Field(default=50.0, gt=0, description="Plot-scale denominator: 50 is 1:50 (text x 50)."),
] = 50.0,
ctx: Context = None,
) -> dict:
"""Label the room around (x, y) with its name, number and MEASURED area.
The area is never typed: the tool reads the wall faces on the `wall` layer
of the arch set (LINE and lightweight polyline, both engines; anything else
there is listed in `skipped`), finds the closed face
the point lies in and writes that face's area - net of the wall thickness
and of any column inside it - into the label and into an `ACADMCP_ARCH`
room record on the name text, which `arch_schedule` reads. A door or window
recorded on the drawing closes its gap along the wall faces, so two rooms
either side of a door stay two rooms.
Refused before anything is drawn: an unknown `lang` (named with en, tr), a
`number`/`name` the room model refuses, a `room_id` already labelled on the
drawing, a point in no closed face of the walls (draw or close them first,
or read a plan drawn by other means with `arch_rooms_detect`), a point
inside a wall body, and a point in a room that already carries a label (one
face, one room record - a retried call does not add a second area to the
schedule).
"""
from engineering.arch.rooms import label_room
room = {"name": name, "number": number, "at": [x, y]}
if room_id:
room["id"] = room_id
await ctx.info(f"arch_room: {name!r} at ({x}, {y})")
try:
return await label_room(_backend(ctx), room, lang=lang, scale=scale)
except ValueError as exc:
raise ToolError(str(exc)) from exc
@cad_tool(summary="Find the rooms of any plan and measure them; never modifies it.", cost="read")
@mcp.tool(
annotations={"title": "Architecture: Detect Rooms", "readOnlyHint": True},
tags={"arch", "query"},
)
async def arch_rooms_detect(
layers: Annotated[
list[str] | None,
Field(
default=None,
description="Layers holding the walls; default: every layer whose name contains "
"WALL or DUVAR.",
),
] = None,
min_area: Annotated[
float,
Field(default=1.0e6, gt=0, description="Smallest room reported, mm² (1e6 = 1 m²)."),
] = 1.0e6,
tol: Annotated[
float,
Field(default=1.0, gt=0, description="Endpoints closer than this (mm) are one corner."),
] = 1.0,
ctx: Context = None,
) -> dict:
"""Read the rooms of a plan - ours or a foreign one made of plain lines.
Every LINE and lightweight polyline on the layers is split where lines
cross or touch, and the closed faces they bound are measured (mm², net of
nested pieces).
Faces under `min_area`, and thin faces (mean width under 600 mm - wall
bodies and reveals, this reader's heuristic) are left out. Each room
reports `confidence`: 1.0 when every edge lies on the arch wall layer, 0.6
when any edge is a plain line from elsewhere - read `confidence_min` before
trusting a foreign plan. Room labels already on the drawing are reported
in `labels` with the face they sit in; a face with more than one is listed
in `label_conflicts`.
Never modifies the drawing. Arcs and bulged polyline edges, old-style
POLYLINEs and block references are listed in `skipped` with their handle
and reason - never flattened into chords, never dropped silently. Read
`skipped` before trusting `count`. A foreign plan's open doorways are
not closed - two rooms joined by one read as one face. Refused: a named
layer the drawing does not have (with the list of the ones it has), and no
layer matching WALL or DUVAR when `layers` is omitted.
"""
from engineering.arch.rooms import rooms_detect
try:
return await rooms_detect(_backend(ctx), layers=layers, min_area=min_area, tol=tol)
except ValueError as exc:
raise ToolError(str(exc)) from exc
@cad_tool(summary="Draw a door, window or room schedule from the plan records.", cost="mutate")
@mcp.tool(
annotations={"title": "Architecture: Schedule", "destructiveHint": False},
tags={"arch", "create"},
)
async def arch_schedule(
kind: Annotated[str, Field(description="doors | windows | rooms")],
x: Annotated[float, Field(description="WCS X of the table's top-left corner.")],
y: Annotated[float, Field(description="WCS Y of the table's top-left corner.")],
lang: Annotated[
str, Field(default="en", description="Headers and numbers: en | tr (decimal comma).")
] = "en",
scale: Annotated[
float,
Field(default=50.0, gt=0, description="Plot-scale denominator: 50 is 1:50."),
] = 50.0,
ctx: Context = None,
) -> dict:
"""Draw a door, window or room schedule as a real TABLE - a read of the drawing.
The rows come from the `ACADMCP_ARCH` records on the drawing (what
`arch_wall` / `arch_opening` / `arch_room` write), never from the caller:
doors by tag with width, height, swing, hand and wall; windows by tag with
width, height, sill and wall; rooms by number with name and the measured
area, plus a total. An opening drawn without a tag gets the next free D1 /
W1 (K1 / P1 in Turkish) in wall order; an explicit tag is kept. A value the
model does not carry is an empty cell, never a default. `representation`
reports what was drawn: `native` (ACAD_TABLE, live) or `composite` (rules
and MTEXT, headless).
Refused before anything is drawn: an unknown `kind` or `lang` (named with
the list), nothing of that kind on the drawing, and more rows than one
TABLE takes (198).
"""
from engineering.arch.schedule import draw_schedule
await ctx.info(f"arch_schedule: {kind} at ({x}, {y})")
try:
return await draw_schedule(_backend(ctx), kind, at=(x, y), lang=lang, scale=scale)
except ValueError as exc:
raise ToolError(str(exc)) from exc
# ── arch: catalogue (group C) ──
+240
View File
@@ -0,0 +1,240 @@
"""The planar-face finder: segments in, closed faces out, areas measured.
Every expected coordinate and area below is computed by hand in the comment
beside it; nothing is read back from the implementation.
"""
from __future__ import annotations
import re
import pytest
from engineering.arch.faces import Face, face_containing, planar_faces, tagged_faces
EPS = 1e-9
def rect(x0, y0, x1, y1):
"""The four sides of an axis-aligned rectangle as segments."""
return [
((x0, y0), (x1, y0)),
((x1, y0), (x1, y1)),
((x1, y1), (x0, y1)),
((x0, y1), (x0, y0)),
]
def two_rooms():
"""Two 4000 x 5000 rooms between 200 mm walls, as clean outlines.
Outer outline (0,0)-(8600,5400). Inner face of the exterior walls
(200,200)-(8400,5200), with the shared wall's two faces x=4200 and x=4400
stopping on it at T-junctions - exactly what a clean wall engine draws.
Room A is (200,200)-(4200,5200), room B (4400,200)-(8400,5200).
"""
return [
*rect(0.0, 0.0, 8600.0, 5400.0),
*rect(200.0, 200.0, 8400.0, 5200.0),
((4200.0, 200.0), (4200.0, 5200.0)),
((4400.0, 200.0), (4400.0, 5200.0)),
]
def test_one_rectangle_is_one_ccw_face_starting_low_left():
(face,) = planar_faces(rect(0.0, 0.0, 4000.0, 3000.0))
assert face.loop == ((0.0, 0.0), (4000.0, 0.0), (4000.0, 3000.0), (0.0, 3000.0))
assert abs(face.area - 12_000_000.0) < EPS # 4000 x 3000
assert face.centroid == pytest.approx((2000.0, 1500.0), abs=EPS)
assert face.holes == ()
def test_the_same_rectangle_drawn_clockwise_is_still_a_ccw_face():
clockwise = [
((0.0, 0.0), (0.0, 3000.0)),
((0.0, 3000.0), (4000.0, 3000.0)),
((4000.0, 3000.0), (4000.0, 0.0)),
((4000.0, 0.0), (0.0, 0.0)),
]
(face,) = planar_faces(clockwise)
assert face.loop == ((0.0, 0.0), (4000.0, 0.0), (4000.0, 3000.0), (0.0, 3000.0))
def test_two_adjacent_rooms_are_two_room_faces_plus_the_wall_bodies():
faces = planar_faces(two_rooms())
# room A, room B, the shared wall body, and the exterior ring
assert len(faces) == 4
room_a = face_containing(faces, (2200.0, 2700.0))
room_b = face_containing(faces, (6400.0, 2700.0))
assert abs(room_a.area - 20_000_000.0) < EPS # 4000 x 5000
assert abs(room_b.area - 20_000_000.0) < EPS
assert room_a.loop == ((200.0, 200.0), (4200.0, 200.0), (4200.0, 5200.0), (200.0, 5200.0))
assert room_b.loop == ((4400.0, 200.0), (8400.0, 200.0), (8400.0, 5200.0), (4400.0, 5200.0))
assert room_a.centroid == pytest.approx((2200.0, 2700.0), abs=EPS)
shared = face_containing(faces, (4300.0, 2700.0))
assert abs(shared.area - 1_000_000.0) < EPS # 200 x 5000
ring = face_containing(faces, (100.0, 2700.0))
# 8600 x 5400 = 46 440 000 minus the inner outline 8200 x 5000 = 41 000 000
assert abs(ring.area - 5_440_000.0) < EPS
# the hole is the inner outline, clockwise, split where the shared wall meets it
assert ring.holes == (
(
(200.0, 200.0),
(200.0, 5200.0),
(4200.0, 5200.0),
(4400.0, 5200.0),
(8400.0, 5200.0),
(8400.0, 200.0),
(4400.0, 200.0),
(4200.0, 200.0),
),
)
# the ring's net region is symmetric about the outline's centre
assert ring.centroid == pytest.approx((4300.0, 2700.0), abs=1e-6)
def test_a_4000_by_5000_room_between_200_mm_walls_measures_20_m2_not_the_axis_area():
# walls on axes (-100,-100)-(4100,5100); the axis rectangle is 4200 x 5200
faces = planar_faces([*rect(-200.0, -200.0, 4200.0, 5200.0), *rect(0.0, 0.0, 4000.0, 5000.0)])
room = face_containing(faces, (2000.0, 2500.0))
assert abs(room.area - 20_000_000.0) < EPS # 20.00 m2, not 21.84 m2
wall = face_containing(faces, (-100.0, 2500.0))
# 4400 x 5400 = 23 760 000 minus 20 000 000
assert abs(wall.area - 3_760_000.0) < EPS
def test_a_t_junction_splits_the_line_it_touches():
# a partition from (2000,0) to (2000,3000) stops on both long sides
faces = planar_faces([*rect(0.0, 0.0, 4000.0, 3000.0), ((2000.0, 0.0), (2000.0, 3000.0))])
assert len(faces) == 2
left, right = faces
assert left.loop == ((0.0, 0.0), (2000.0, 0.0), (2000.0, 3000.0), (0.0, 3000.0))
assert right.loop == ((2000.0, 0.0), (4000.0, 0.0), (4000.0, 3000.0), (2000.0, 3000.0))
assert abs(left.area - 6_000_000.0) < EPS # 2000 x 3000
assert abs(right.area - 6_000_000.0) < EPS
def test_a_t_touch_within_tol_still_splits():
# the partition stops 0.5 mm short of the bottom side: within tol=1.0
faces = planar_faces(
[*rect(0.0, 0.0, 4000.0, 3000.0), ((2000.0, 0.5), (2000.0, 3000.0))], tol=1.0
)
assert len(faces) == 2
# the split vertex is the partition's own end (2000, 0.5), which bends the
# bottom side by 0.5 mm; each half is 2000 x 3000 minus a 0.5 mm-high sliver
# triangle of base 2000: 6 000 000 - 0.5 * 2000 * 0.5 = 5 999 500
assert [round(f.area, 6) for f in faces] == [5_999_500.0, 5_999_500.0]
def test_crossing_lines_split_each_other_and_their_overshoots_are_pruned():
segments = [
*rect(0.0, 0.0, 4000.0, 3000.0),
((-500.0, 1500.0), (4500.0, 1500.0)), # overshoots 500 at both ends
((2000.0, -500.0), (2000.0, 3500.0)),
]
faces = planar_faces(segments)
assert len(faces) == 4
assert all(abs(f.area - 3_000_000.0) < EPS for f in faces) # 2000 x 1500 each
assert faces[0].loop == ((0.0, 0.0), (2000.0, 0.0), (2000.0, 1500.0), (0.0, 1500.0))
def test_a_dangling_segment_is_ignored():
segments = [
*rect(0.0, 0.0, 4000.0, 3000.0),
((4000.0, 3000.0), (5000.0, 4000.0)), # a spur off a corner
((1000.0, 1000.0), (1500.0, 1200.0)), # a stray line inside the room
]
(face,) = planar_faces(segments)
assert abs(face.area - 12_000_000.0) < EPS
def test_endpoints_within_tol_merge_and_a_wider_gap_leaves_the_loop_open():
closing = ((0.0, 3000.4), (0.0, 0.0)) # starts 0.4 mm off the corner (0, 3000)
sides = [
((0.0, 0.0), (4000.0, 0.0)),
((4000.0, 0.0), (4000.0, 3000.0)),
((4000.0, 3000.0), (0.0, 3000.0)),
]
(face,) = planar_faces([*sides, closing], tol=1.0)
# (0, 3000) was seen first, so it is the vertex: the area is exact
assert abs(face.area - 12_000_000.0) < EPS
open_gap = ((0.0, 2995.0), (0.0, 0.0)) # stops 5 mm short of (0, 3000)
assert planar_faces([*sides, open_gap], tol=1.0) == ()
def test_an_overshoot_through_a_corner_still_closes_the_loop():
sides = [
((0.0, 0.0), (4000.0, 0.0)),
((4000.0, 0.0), (4000.0, 3000.0)),
((4000.0, 3000.0), (0.0, 3000.0)),
]
# the closing side runs 5 mm past the corner (0, 3000): the corner splits it
# and the 5 mm stub is a dangling edge
(face,) = planar_faces([*sides, ((0.0, 3005.0), (0.0, 0.0))])
assert face.loop == ((0.0, 0.0), (4000.0, 0.0), (4000.0, 3000.0), (0.0, 3000.0))
assert abs(face.area - 12_000_000.0) < EPS
def test_collinear_overlaps_become_one_edge():
segments = [
((0.0, 0.0), (3000.0, 0.0)),
((1000.0, 0.0), (4000.0, 0.0)), # overlaps the first from 1000 to 3000
((4000.0, 0.0), (4000.0, 3000.0)),
((4000.0, 3000.0), (0.0, 3000.0)),
((0.0, 3000.0), (0.0, 0.0)),
]
(face,) = planar_faces(segments)
assert face.loop == (
(0.0, 0.0),
(1000.0, 0.0),
(3000.0, 0.0),
(4000.0, 0.0),
(4000.0, 3000.0),
(0.0, 3000.0),
)
assert abs(face.area - 12_000_000.0) < EPS
def test_a_column_inside_a_room_is_a_hole_in_the_room():
segments = [*rect(0.0, 0.0, 4000.0, 3000.0), *rect(1000.0, 1000.0, 1400.0, 1400.0)]
faces = planar_faces(segments)
room = face_containing(faces, (3000.0, 2000.0))
column = face_containing(faces, (1200.0, 1200.0))
assert abs(room.area - 11_840_000.0) < EPS # 12 000 000 - 400 x 400
assert abs(column.area - 160_000.0) < EPS
assert len(room.holes) == 1
# net centroid: (12e6 * (2000, 1500) - 160 000 * (1200, 1200)) / 11 840 000
assert room.centroid == pytest.approx(
(
(12e6 * 2000.0 - 160_000.0 * 1200.0) / 11_840_000.0,
(12e6 * 1500.0 - 160_000.0 * 1200.0) / 11_840_000.0,
),
abs=1e-6,
)
def test_a_point_in_no_face_answers_none():
faces = planar_faces(rect(0.0, 0.0, 4000.0, 3000.0))
assert face_containing(faces, (5000.0, 5000.0)) is None
def test_edge_tags_ride_along_to_every_face_edge():
tagged = [(p1, p2, "WALL") for p1, p2 in rect(0.0, 0.0, 4000.0, 3000.0)]
tagged[0] = (tagged[0][0], tagged[0][1], "0")
((face, tags),) = tagged_faces(tagged)
assert tags == (frozenset({"0"}), frozenset({"WALL"}), frozenset({"WALL"}), frozenset({"WALL"}))
assert isinstance(face, Face)
@pytest.mark.parametrize(
("segments", "tol", "message"),
[
([((0.0, 0.0),)], 1.0, "segments[0]: expected (p1, p2) or (p1, p2, tag), got 1 items"),
([((0.0, "x"), (1.0, 1.0))], 1.0, "segments[0][0][1]: expected a number, got 'x'"),
([((0.0, float("nan")), (1.0, 1.0))], 1.0, "segments[0][0][1]: must be finite"),
(rect(0.0, 0.0, 1.0, 1.0), 0.0, "tol: must be greater than zero, got 0"),
],
)
def test_bad_input_is_refused_by_path(segments, tol, message):
with pytest.raises(ValueError, match=re.escape(message)):
planar_faces(segments, tol=tol)
+546
View File
@@ -0,0 +1,546 @@
"""Room labels with a measured area, and the foreign-plan reader.
The pure half asserts emitted primitives and hand-computed coordinates; the
async half runs against the headless backend (the `backend` fixture in
tests/conftest.py). The wall and opening records a drawn plan carries are
written with `model.to_values` directly - the wall engine that writes them in
production is Task 4's, on another branch, and the record shape is pinned by
the skeleton.
"""
from __future__ import annotations
import re
import pytest
from engineering.arch.layers import ARCH_ROLE_LAYER
from engineering.arch.model import APP_ID, Opening, Room, Wall, decode, to_payload, to_values
from engineering.arch.rooms import (
CONFIDENCE_FOREIGN,
CONFIDENCE_OURS,
detect_rooms,
label_room,
opening_closures,
read_records,
room_label_prims,
rooms_detect,
)
from engineering.mech.primitives import Text
WALL = ARCH_ROLE_LAYER["wall"]
ROOM = ARCH_ROLE_LAYER["room"]
EPS = 1e-9
def rect(x0, y0, x1, y1):
return [((x0, y0), (x1, y0)), ((x1, y0), (x1, y1)), ((x1, y1), (x0, y1)), ((x0, y1), (x0, y0))]
def two_rooms(tag=None):
"""Two 4000 x 5000 rooms between 200 mm walls (see tests/test_arch_faces.py)."""
segments = [
*rect(0.0, 0.0, 8600.0, 5400.0),
*rect(200.0, 200.0, 8400.0, 5200.0),
((4200.0, 200.0), (4200.0, 5200.0)),
((4400.0, 200.0), (4400.0, 5200.0)),
]
return segments if tag is None else [(p1, p2, tag) for p1, p2 in segments]
async def draw_lines(backend, segments, layer):
for (x1, y1), (x2, y2) in segments:
await backend.entity_create_line(x1, y1, x2, y2, layer=layer)
async def write_record(backend, obj, layer):
"""A record the way a drawn plan carries it: XDATA on an entity of its layer."""
carrier = await backend.entity_create_line(0.0, -1000.0, 1.0, -1000.0, layer=layer)
await backend.entity_set_xdata(carrier.handle, APP_ID, to_values(to_payload(obj)))
return carrier.handle
async def texts(backend, layer):
return {
e.properties.get("text"): e
for e in await backend.entity_list(type_filter="TEXT", layer_filter=layer, limit=1000)
}
# ── pure: the label ─────────────────────────────────────────────────────────
def test_label_prims_stack_name_number_area_down_from_the_point_at_plot_scale():
room = Room(id="R1", name="LIVING", number="01", at=(2000.0, 2500.0), area=20_000_000.0)
prims = room_label_prims(room, lang="en", scale=50)
# 1:50 -> name 3.5 x 50 = 175, number and area 2.5 x 50 = 125;
# baselines 2500, 2500 - 1.6 x 125 = 2300, 2300 - 1.6 x 125 = 2100
assert prims == (
Text(at=(2000.0, 2500.0), text="LIVING", height=175.0, rotation=0.0, role="room"),
Text(at=(2000.0, 2300.0), text="01", height=125.0, rotation=0.0, role="room"),
Text(at=(2000.0, 2100.0), text="20.00 m²", height=125.0, rotation=0.0, role="room"),
)
def test_label_without_a_number_puts_the_area_on_the_second_line_and_tr_uses_a_comma():
room = Room(id="R2", name="SALON", number=None, at=(0.0, 0.0), area=24_500_000.0)
prims = room_label_prims(room, lang="tr", scale=100)
# 1:100 -> name 350, area 250 at 0 - 1.6 x 250 = -400
assert [(p.at, p.text, p.height) for p in prims] == [
((0.0, 0.0), "SALON", 350.0),
((0.0, -400.0), "24,50 m²", 250.0),
]
@pytest.mark.parametrize(
("kwargs", "message"),
[
({"lang": "de"}, "lang: 'de' is not a label language; languages are en, tr"),
({"scale": 0}, "scale: must be a finite number greater than zero, got 0"),
],
)
def test_label_prims_refuse_an_unknown_language_or_scale(kwargs, message):
room = Room(id="R1", name="A", number=None, at=(0.0, 0.0), area=1.0)
with pytest.raises(ValueError, match=re.escape(message)):
room_label_prims(room, **kwargs)
# ── pure: openings close their gap ──────────────────────────────────────────
def test_an_opening_is_closed_along_both_wall_faces():
# the shared wall's axis runs up x=4300 from y=100 to y=5300, 200 thick, centred;
# a 900 door 1100 from the axis start spans y=1200..2100.
wall = Wall(id="w5", axis=((4300.0, 100.0), (4300.0, 5300.0)), thickness=200.0)
door = Opening(id="d1", wall="w5", kind="door", offset=1100.0, width=900.0)
closures, omitted = opening_closures([wall], [door])
# u = (0, 1), left normal n = (-1, 0); face offsets -100 and +100
assert closures == (
((4400.0, 1200.0), (4400.0, 2100.0)),
((4200.0, 1200.0), (4200.0, 2100.0)),
)
assert omitted == ()
def test_a_left_justified_wall_closes_its_faces_on_the_axis_and_one_thickness_left():
wall = Wall(id="w1", axis=((0.0, 0.0), (5000.0, 0.0)), thickness=200.0, justification="left")
window = Opening(id="p1", wall="w1", kind="window", offset=1000.0, width=1200.0)
closures, _ = opening_closures([wall], [window])
# u = (1, 0), n = (0, 1): faces at 0 and +200
assert closures == (((1000.0, 0.0), (2200.0, 0.0)), ((1000.0, 200.0), (2200.0, 200.0)))
def test_an_opening_whose_wall_has_no_record_is_omitted_not_guessed():
door = Opening(id="d9", wall="nowhere", kind="door", offset=0.0, width=900.0)
closures, omitted = opening_closures([], [door])
assert closures == ()
assert omitted == (
{
"element": "d9",
"reason": "host wall 'nowhere' has no record on this drawing; its gap is left open",
},
)
def test_a_stale_opening_past_the_end_of_its_wall_is_omitted_not_raised():
# a record left behind after its wall was shortened: 9000 on a 5000 axis.
# The reader is read-only and must still answer for every other room.
wall = Wall(id="w1", axis=((0.0, 0.0), (5000.0, 0.0)), thickness=200.0)
stale = Opening(id="d4", wall="w1", kind="door", offset=9000.0, width=900.0)
closures, omitted = opening_closures([wall], [stale])
assert closures == ()
assert omitted == (
{
"element": "d4",
"reason": "offset 9000.0 is outside wall w1 (axis length 5000); its gap is left open",
},
)
# ── pure: detect_rooms ──────────────────────────────────────────────────────
def test_detect_rooms_keeps_the_rooms_and_drops_the_wall_bodies():
rooms = detect_rooms(two_rooms(WALL))
# the exterior ring (mean width 2 x 5 440 000 / 54 400 = 200) and the shared
# wall body (2 x 1 000 000 / 10 400 = 192) are wall bodies, not rooms
assert [r["area"] for r in rooms] == [20_000_000.0, 20_000_000.0]
assert [r["centroid"] for r in rooms] == [(2200.0, 2700.0), (6400.0, 2700.0)]
assert {r["confidence"] for r in rooms} == {CONFIDENCE_OURS}
def test_one_foreign_edge_makes_the_face_foreign():
segments = two_rooms(WALL)
# the inner outline's bottom edge from (200,200) to (8400,200) becomes a plain line
segments[4] = (segments[4][0], segments[4][1], "0")
rooms = detect_rooms(segments)
assert [r["confidence"] for r in rooms] == [CONFIDENCE_FOREIGN, CONFIDENCE_FOREIGN]
def test_untagged_lines_are_foreign_and_min_area_filters():
rooms = detect_rooms(two_rooms())
assert [r["confidence"] for r in rooms] == [0.6, 0.6]
assert detect_rooms(two_rooms(), min_area=20_000_001.0) == ()
# ── async: label_room ───────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_a_4000_by_5000_room_between_200_mm_walls_is_labelled_20_m2(backend):
await draw_lines(
backend, [*rect(-200.0, -200.0, 4200.0, 5200.0), *rect(0.0, 0.0, 4000.0, 5000.0)], WALL
)
result = await label_room(
backend, {"name": "LIVING", "number": "01", "at": [2000.0, 2500.0]}, lang="en", scale=50
)
assert result["room"]["area"] == pytest.approx(20_000_000.0, abs=EPS)
assert result["room"]["area_text"] == "20.00 m²"
assert result["room"]["id"] == "R1"
assert result["loop"] == [[0.0, 0.0], [4000.0, 0.0], [4000.0, 5000.0], [0.0, 5000.0]]
labels = await texts(backend, ROOM)
assert set(labels) == {"LIVING", "01", "20.00 m²"}
raw = await backend.entity_get_xdata(result["handle"], APP_ID)
payload = decode(raw["xdata"][APP_ID])
assert payload["kind"] == "room"
assert payload["id"] == "R1"
assert payload["area"] == pytest.approx(20_000_000.0, abs=EPS)
@pytest.mark.asyncio
async def test_turkish_labels_use_a_decimal_comma(backend):
await draw_lines(
backend, [*rect(-200.0, -200.0, 4200.0, 5200.0), *rect(0.0, 0.0, 4000.0, 5000.0)], WALL
)
await label_room(backend, {"name": "SALON", "at": [2000.0, 2500.0]}, lang="tr")
assert "20,00 m²" in await texts(backend, ROOM)
@pytest.mark.asyncio
async def test_a_door_in_the_shared_wall_does_not_merge_the_two_rooms(backend):
door_gap = [
*rect(0.0, 0.0, 8600.0, 5400.0),
*rect(200.0, 200.0, 8400.0, 5200.0),
# the shared wall's faces, interrupted from y=1200 to y=2100, and its two jambs
((4200.0, 200.0), (4200.0, 1200.0)),
((4200.0, 2100.0), (4200.0, 5200.0)),
((4400.0, 200.0), (4400.0, 1200.0)),
((4400.0, 2100.0), (4400.0, 5200.0)),
((4200.0, 1200.0), (4400.0, 1200.0)),
((4200.0, 2100.0), (4400.0, 2100.0)),
]
await draw_lines(backend, door_gap, WALL)
await write_record(
backend, Wall(id="w5", axis=((4300.0, 100.0), (4300.0, 5300.0)), thickness=200.0), WALL
)
await write_record(
backend,
Opening(id="d1", wall="w5", kind="door", offset=1100.0, width=900.0),
ARCH_ROLE_LAYER["door"],
)
result = await label_room(backend, {"name": "A", "at": [2200.0, 2700.0]})
# without the closures room A would read 20 000 000 + 20 000 000 + 900 x 200
assert result["room"]["area"] == pytest.approx(20_000_000.0, abs=EPS)
assert result["closures"] == 2
@pytest.mark.asyncio
async def test_label_room_refusals_leave_the_drawing_untouched(backend):
await draw_lines(backend, rect(0.0, 0.0, 4000.0, 5000.0), WALL)
before = await backend.entity_count()
with pytest.raises(ValueError, match=re.escape("room.area: a room's area is measured")):
await label_room(backend, {"name": "A", "at": [100.0, 100.0], "area": 5.0})
with pytest.raises(ValueError, match=re.escape("room.at: (9000, 9000) lies in no closed face")):
await label_room(backend, {"name": "A", "at": [9000.0, 9000.0]})
with pytest.raises(ValueError, match=re.escape("lang: 'fr' is not a label language")):
await label_room(backend, {"name": "A", "at": [100.0, 100.0]}, lang="fr")
assert await backend.entity_count() == before
await label_room(backend, {"id": "R7", "name": "A", "at": [100.0, 100.0]})
with pytest.raises(ValueError, match=re.escape("room.id: 'R7' is already labelled")):
await label_room(backend, {"id": "R7", "name": "B", "at": [200.0, 200.0]})
@pytest.mark.asyncio
async def test_a_label_point_inside_a_wall_body_is_refused(backend):
await draw_lines(
backend, [*rect(-200.0, -200.0, 4200.0, 5200.0), *rect(0.0, 0.0, 4000.0, 5000.0)], WALL
)
with pytest.raises(
ValueError, match=re.escape("room.at: (-100, 2500) lies inside a wall body")
):
await label_room(backend, {"name": "A", "at": [-100.0, 2500.0]})
# ── async: rooms_detect ─────────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_a_foreign_plan_of_plain_lines_yields_the_same_areas_and_is_not_modified(backend):
await backend.entity_create_polyline(
[[0.0, 0.0], [8600.0, 0.0], [8600.0, 5400.0], [0.0, 5400.0]], True, layer="DUVAR"
)
await draw_lines(backend, two_rooms()[4:], "DUVAR")
await draw_lines(backend, [((-500.0, -500.0), (9000.0, -500.0))], "0") # not read
before = await backend.entity_count()
result = await rooms_detect(backend)
assert result["layers"] == ["DUVAR"]
assert [r["area"] for r in result["rooms"]] == [20_000_000.0, 20_000_000.0]
assert [r["centroid"] for r in result["rooms"]] == [[2200.0, 2700.0], [6400.0, 2700.0]]
assert result["confidence_min"] == CONFIDENCE_FOREIGN
assert result["modified"] is False
assert await backend.entity_count() == before
@pytest.mark.asyncio
async def test_our_own_walls_read_at_full_confidence_with_their_labels(backend):
await draw_lines(backend, two_rooms(), WALL)
await label_room(backend, {"name": "A", "number": "01", "at": [2200.0, 2700.0]})
result = await rooms_detect(backend)
assert result["layers"] == [WALL]
assert [r["confidence"] for r in result["rooms"]] == [1.0, 1.0]
first, second = result["rooms"]
assert first["label"] == {"id": "R1", "name": "A", "number": "01", "area": 20_000_000.0}
assert second["label"] is None
@pytest.mark.asyncio
async def test_the_record_reader_pages_the_drawing_once_not_once_per_arch_layer(
backend, monkeypatch
):
# a live seat pays one cross-process call per entity per scan; twelve arch
# layers must not mean twelve scans of the whole space
await write_record(
backend, Wall(id="w1", axis=((0.0, 0.0), (5000.0, 0.0)), thickness=200.0), WALL
)
await write_record(
backend,
Opening(id="d1", wall="w1", kind="door", offset=1000.0, width=900.0),
ARCH_ROLE_LAYER["door"],
)
await backend.entity_create_line(0.0, 0.0, 1.0, 1.0, layer="0") # not an arch layer
calls = []
original = backend.entity_list
async def counting(*args, **kwargs):
calls.append(kwargs)
return await original(*args, **kwargs)
monkeypatch.setattr(backend, "entity_list", counting)
records = await read_records(backend)
assert len(calls) == 1
assert [w.id for w in records["walls"]] == ["w1"]
assert [o.id for o in records["openings"]] == ["d1"]
assert records["unreadable"] == 0
@pytest.mark.asyncio
async def test_rooms_detect_refuses_a_layer_that_is_not_there(backend):
await draw_lines(backend, rect(0.0, 0.0, 4000.0, 5000.0), "WALLS")
with pytest.raises(ValueError, match=re.escape("layers: NOPE not in this drawing")):
await rooms_detect(backend, layers=["NOPE"])
@pytest.mark.asyncio
async def test_an_arc_on_a_wall_layer_is_reported_skipped_never_approximated(backend):
await draw_lines(backend, rect(0.0, 0.0, 4000.0, 5000.0), "WALLS")
arc = await backend.entity_create_arc(2000.0, 5000.0, 500.0, 0.0, 180.0, layer="WALLS")
result = await rooms_detect(backend)
assert [r["area"] for r in result["rooms"]] == [20_000_000.0]
assert [s["handle"] for s in result["skipped"]] == [arc.handle]
# ── what the reader cannot read is skipped, by name ─────────────────────────
def outline(x0, y0, x1, y1):
return [(x0, y0), (x1, y0), (x1, y1), (x0, y1)]
@pytest.mark.asyncio
async def test_an_old_style_polyline_is_skipped_by_name_not_dropped(backend):
# one room of old-style 2D POLYLINEs next to one of lightweight polylines;
# the headless engine reports the old-style kind with no vertices at all
msp = backend._doc.modelspace()
heavy = [
msp.add_polyline2d(outline(*box), close=True, dxfattribs={"layer": "WALLS"}).dxf.handle
for box in ((0.0, 0.0, 4400.0, 5400.0), (200.0, 200.0, 4200.0, 5200.0))
]
for box in ((10000.0, 0.0, 14400.0, 5400.0), (10200.0, 200.0, 14200.0, 5200.0)):
await backend.entity_create_polyline([list(p) for p in outline(*box)], True, layer="WALLS")
result = await rooms_detect(backend)
assert [r["area"] for r in result["rooms"]] == [20_000_000.0]
assert [(s["handle"], s["type"]) for s in result["skipped"]] == [
(heavy[0], "POLYLINE"),
(heavy[1], "POLYLINE"),
]
assert "old-style polyline" in result["skipped"][0]["reason"]
@pytest.mark.asyncio
async def test_walls_inside_a_block_reference_are_skipped_and_annotation_is_passed_over(
backend,
):
await draw_lines(backend, rect(0.0, 0.0, 4000.0, 5000.0), "WALLS")
block = backend._doc.blocks.new("ROOMBLK")
for (x1, y1), (x2, y2) in [
*rect(0.0, 0.0, 4400.0, 5400.0),
*rect(200.0, 200.0, 4200.0, 5200.0),
]:
block.add_line((x1, y1), (x2, y2))
insert = backend._doc.modelspace().add_blockref(
"ROOMBLK", (20000.0, 0.0), dxfattribs={"layer": "WALLS"}
)
await backend.entity_create_text("W1", 100.0, 100.0, 100.0, layer="WALLS")
result = await rooms_detect(backend)
assert [r["area"] for r in result["rooms"]] == [20_000_000.0]
assert [(s["handle"], s["type"]) for s in result["skipped"]] == [(insert.dxf.handle, "INSERT")]
assert "walls inside a block are not read" in result["skipped"][0]["reason"]
class LiveRows:
"""A backend that answers with the EntityInfo rows a live seat produced.
The rows are the ones `backends.com_backend._entity_info` returned on
AutoCAD 2026 (2026-09-23) for three closed ``ModelSpace.AddPolyline``
outlines on WALLS - outer 0..8600 x 0..5400, rooms 200..4200 and
4400..8400 x 200..5200 - whose ``Coordinates`` came back as x, y, z
triples and were read two at a time; plus an ``AddLightWeightPolyline``
pair (AcDbPolyline, reported POLYLINE) and a block reference. Read
through, the old-style rows gave five faces of 2.6 to 6.9 m².
"""
def __init__(self):
from backends.base import EntityInfo, LayerInfo
def row(handle, kind, points=None):
props = {} if points is None else {"points": points, "closed": True}
if points is not None:
props["bulges"] = [0.0] * len(points)
return EntityInfo(handle, kind, "WALLS", 256, "ByLayer", True, props)
self.rows = [
row(
"2A1",
"2DPOLYLINE",
[
[0.0, 0.0],
[0.0, 8600.0],
[0.0, 0.0],
[8600.0, 5400.0],
[0.0, 0.0],
[5400.0, 0.0],
],
),
row(
"2A2",
"2DPOLYLINE",
[
[200.0, 200.0],
[0.0, 4200.0],
[200.0, 0.0],
[4200.0, 5200.0],
[0.0, 200.0],
[5200.0, 0.0],
],
),
row(
"2A3",
"2DPOLYLINE",
[
[4400.0, 200.0],
[0.0, 8400.0],
[200.0, 0.0],
[8400.0, 5200.0],
[0.0, 4400.0],
[5200.0, 0.0],
],
),
row("2A4", "POLYLINE", [list(p) for p in outline(10000.0, 0.0, 14400.0, 5400.0)]),
row("2A5", "POLYLINE", [list(p) for p in outline(10200.0, 200.0, 14200.0, 5200.0)]),
row("2A6", "BLOCKREFERENCE"),
]
self.layers = [
LayerInfo(name, 7, "Continuous", 0.25, True, False, False, name == "0")
for name in ("0", "WALLS")
]
async def layer_list(self):
return self.layers
async def entity_list(self, type_filter=None, layer_filter=None, limit=200, offset=0):
rows = [r for r in self.rows if layer_filter is None or r.layer == layer_filter]
return rows[offset : offset + limit]
async def entity_get_xdata(self, handle, app_id):
return {"xdata": {}}
@pytest.mark.asyncio
async def test_live_old_style_polylines_are_skipped_not_read_as_their_garbled_points():
result = await rooms_detect(LiveRows())
# only the lightweight pair is a room; the garbled rows make no faces at all
assert [r["area"] for r in result["rooms"]] == [20_000_000.0]
assert result["confidence_min"] == CONFIDENCE_FOREIGN
assert [(s["handle"], s["type"]) for s in result["skipped"]] == [
("2A1", "2DPOLYLINE"),
("2A2", "2DPOLYLINE"),
("2A3", "2DPOLYLINE"),
("2A6", "BLOCKREFERENCE"),
]
# ── one face, one room record ───────────────────────────────────────────────
@pytest.mark.asyncio
async def test_a_second_label_in_a_labelled_room_is_refused_and_draws_nothing(backend):
await draw_lines(
backend, [*rect(0.0, 0.0, 4400.0, 5400.0), *rect(200.0, 200.0, 4200.0, 5200.0)], WALL
)
first = await label_room(backend, {"name": "LIVING", "at": [1000.0, 3000.0]})
before = await backend.entity_count()
with pytest.raises(
ValueError,
match=re.escape(
f"room.at: (2000, 3000) lies in a room that is already labelled - "
f"R1 'LIVING' (label {first['handle']})"
),
):
await label_room(backend, {"name": "LIVING2", "at": [2000.0, 3000.0]})
assert await backend.entity_count() == before
result = await rooms_detect(backend)
assert [r["labels"] for r in result["rooms"]] == [
[{"id": "R1", "name": "LIVING", "number": None, "area": 20_000_000.0}]
]
assert result["label_conflicts"] == []
@pytest.mark.asyncio
async def test_the_other_room_can_still_be_labelled(backend):
await draw_lines(backend, two_rooms(), WALL)
await label_room(backend, {"name": "A", "at": [2200.0, 2700.0]})
second = await label_room(backend, {"name": "B", "at": [6400.0, 2700.0]})
assert second["room"]["id"] == "R2"
assert second["room"]["area"] == pytest.approx(20_000_000.0, abs=EPS)
@pytest.mark.asyncio
async def test_rooms_detect_reports_every_record_in_a_face_and_the_conflict(backend):
# two records already on one face (written by other means than arch_room)
await draw_lines(backend, two_rooms(), WALL)
for rid in ("R1", "R2"):
room = Room(id=rid, name=rid, number=None, at=(2200.0, 2700.0), area=20_000_000.0)
await write_record(backend, room, ROOM)
result = await rooms_detect(backend)
first, second = result["rooms"]
assert [label["id"] for label in first["labels"]] == ["R1", "R2"]
assert first["label"]["id"] == "R1"
assert second["labels"] == [] and second["label"] is None
assert result["label_conflicts"] == [
{
"labels": ["R1", "R2"],
"centroid": [2200.0, 2700.0],
"reason": "one face carries 2 room records; a schedule would count its "
"20.00 m2 2 times",
}
]
+77
View File
@@ -0,0 +1,77 @@
"""arch_room and arch_rooms_detect over the MCP wire, on a fresh headless document."""
from __future__ import annotations
import pytest
import pytest_asyncio
from fastmcp import Client
import server
from engineering.arch.layers import ARCH_ROLE_LAYER
pytestmark = pytest.mark.asyncio
WALL = ARCH_ROLE_LAYER["wall"]
def rect(x0, y0, x1, y1):
return [((x0, y0), (x1, y0)), ((x1, y0), (x1, y1)), ((x1, y1), (x0, y1)), ((x0, y1), (x0, y0))]
@pytest_asyncio.fixture
async def client(monkeypatch):
"""In-memory client on a fresh headless document - never the operator's AutoCAD."""
monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
async with Client(server.mcp) as connected:
status = (await connected.call_tool("system_status", {})).structured_content or {}
assert status.get("backend") == "ezdxf", "these tests must never touch live AutoCAD"
await connected.call_tool("drawing_new", {})
yield connected
async def draw_room(client, layer):
"""A 4000 x 5000 room between 200 mm walls: outer and inner outline."""
for (x1, y1), (x2, y2) in [
*rect(-200.0, -200.0, 4200.0, 5200.0),
*rect(0.0, 0.0, 4000.0, 5000.0),
]:
await client.call_tool(
"entity_create_line", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "layer": layer}
)
async def test_arch_room_labels_the_measured_area(client):
await draw_room(client, WALL)
result = await client.call_tool(
"arch_room", {"name": "LIVING", "number": "01", "x": 2000.0, "y": 2500.0}
)
room = result.structured_content["room"]
assert room["area"] == pytest.approx(20_000_000.0)
assert room["area_text"] == "20.00 m²"
assert result.structured_content["layer"] == ARCH_ROLE_LAYER["room"]
async def test_arch_rooms_detect_reads_the_room_back_without_drawing(client):
await draw_room(client, "DUVAR")
before = (await client.call_tool("analysis_entity_stats", {})).structured_content
result = (await client.call_tool("arch_rooms_detect", {})).structured_content
assert result["layers"] == ["DUVAR"]
assert [r["area"] for r in result["rooms"]] == [pytest.approx(20_000_000.0)]
assert result["confidence_min"] == 0.6
after = (await client.call_tool("analysis_entity_stats", {})).structured_content
assert after["total_entities"] == before["total_entities"]
async def test_a_point_outside_every_room_is_refused_over_the_wire(client):
await draw_room(client, WALL)
result = await client.call_tool(
"arch_room", {"name": "X", "x": 9000.0, "y": 9000.0}, raise_on_error=False
)
assert result.is_error is True
assert "room.at: (9000, 9000) lies in no closed face" in result.content[0].text
async def test_the_detector_is_advertised_read_only(client):
tools = {tool.name: tool for tool in await client.list_tools()}
assert tools["arch_rooms_detect"].annotations.readOnlyHint is True
assert tools["arch_room"].annotations.destructiveHint is False
+200
View File
@@ -0,0 +1,200 @@
"""Door, window and room schedules: rows from the plan model, a TABLE from the drawing.
Records are written with `model.to_values` directly - in production the wall
engine (Task 4, another branch) and `arch_room` write them; the record shape is
pinned by the skeleton.
"""
from __future__ import annotations
import re
import pytest
from engineering.arch.lang import vocab
from engineering.arch.layers import ARCH_ROLE_LAYER
from engineering.arch.model import APP_ID, Opening, Room, Wall, to_payload, to_values
from engineering.arch.schedule import SCHEDULE_COLUMNS, draw_schedule, schedule_rows
WALLS = [
Wall(id="w1", axis=((0.0, 0.0), (6000.0, 0.0)), thickness=200.0),
Wall(id="w2", axis=((6000.0, 0.0), (6000.0, 4000.0)), thickness=200.0),
]
def plan():
return {
"walls": list(WALLS),
"openings": [
Opening(
id="da",
wall="w2",
kind="door",
offset=500.0,
width=800.0,
swing="out",
hand="right",
),
Opening(id="db", wall="w1", kind="door", offset=3000.0, width=912.5, height=2100.0),
Opening(
id="dc", wall="w1", kind="door", offset=1000.0, width=900.0, height=2100.0, tag="D2"
),
Opening(
id="wa",
wall="w1",
kind="window",
offset=4000.0,
width=1200.0,
height=1200.0,
sill=900.0,
),
],
"rooms": [
Room(id="R1", name="KITCHEN", number="2", at=(0.0, 0.0), area=15_000_000.0),
Room(id="R2", name="LIVING", number="10", at=(0.0, 0.0), area=20_000_000.0),
Room(id="R3", name="STORE", number=None, at=(0.0, 0.0), area=10_000_000.0),
],
}
def test_door_rows_generate_missing_tags_in_wall_order_and_sort_by_tag():
rows = schedule_rows("doors", plan(), lang="en")
en = vocab("en")
# wall order: w1 at 1000 (tagged D2), w1 at 3000 -> D1, w2 at 500 -> D3 (D2 is taken)
assert [(r["id"], r["tag"], r["generated"]) for r in rows] == [
("db", "D1", True),
("dc", "D2", False),
("da", "D3", True),
]
assert rows[0] == {
"id": "db",
"tag": "D1",
"generated": True,
"width": "912.5",
"height": "2100",
"swing": en["in"],
"hand": en["left"],
"wall": "w1",
}
# no height on the model: an empty cell, never a default
assert rows[2]["height"] == ""
assert (rows[2]["swing"], rows[2]["hand"]) == (en["out"], en["right"])
assert set(SCHEDULE_COLUMNS["doors"]) <= set(rows[0])
def test_turkish_door_rows_use_k_tags_decimal_commas_and_turkish_words():
rows = schedule_rows("doors", plan(), lang="tr")
tr = vocab("tr")
# explicit "D2" is never renamed; generated tags take the Turkish prefix
assert [r["tag"] for r in rows] == ["D2", "K1", "K2"]
k1 = next(r for r in rows if r["tag"] == "K1")
assert k1["width"] == "912,5"
assert (k1["swing"], k1["hand"]) == (tr["in"], tr["left"])
def test_window_rows_carry_sill_and_no_swing():
(row,) = schedule_rows("windows", plan(), lang="en")
assert row == {
"id": "wa",
"tag": "W1",
"generated": True,
"width": "1200",
"height": "1200",
"sill": "900",
"wall": "w1",
}
assert [r["tag"] for r in schedule_rows("windows", plan(), lang="tr")] == ["P1"]
def test_room_rows_sort_by_number_with_the_unnumbered_last_and_end_in_a_total():
rows = schedule_rows("rooms", plan(), lang="en")
assert [(r["number"], r["name"], r["area"]) for r in rows] == [
("2", "KITCHEN", "15.00 m²"),
("10", "LIVING", "20.00 m²"),
("", "STORE", "10.00 m²"),
("", vocab("en")["total"], "45.00 m²"), # 15 + 20 + 10
]
assert rows[-1]["total"] is True
assert rows[-1]["area_mm2"] == 45_000_000.0
def test_turkish_room_rows_write_the_total_with_a_decimal_comma():
rows = schedule_rows("rooms", plan(), lang="tr")
assert (rows[-1]["name"], rows[-1]["area"]) == (vocab("tr")["total"], "45,00 m²")
assert rows[0]["area"] == "15,00 m²"
def test_an_empty_plan_has_no_rows_and_no_total():
assert schedule_rows("rooms", {"walls": [], "openings": [], "rooms": []}) == ()
assert schedule_rows("doors", {"walls": [], "openings": [], "rooms": []}) == ()
@pytest.mark.parametrize(
("kind", "lang", "message"),
[
("stairs", "en", "kind: 'stairs' is not a schedule; schedules are doors, windows, rooms"),
("doors", "de", "lang: 'de' is not a schedule language; languages are en, tr"),
],
)
def test_an_unknown_kind_or_language_is_refused_with_the_list(kind, lang, message):
with pytest.raises(ValueError, match=re.escape(message)):
schedule_rows(kind, plan(), lang=lang)
# ── async: the TABLE ────────────────────────────────────────────────────────
async def write_record(backend, obj, role):
carrier = await backend.entity_create_line(0.0, 0.0, 1.0, 0.0, layer=ARCH_ROLE_LAYER[role])
await backend.entity_set_xdata(carrier.handle, APP_ID, to_values(to_payload(obj)))
async def write_plan(backend):
data = plan()
for wall in data["walls"]:
await write_record(backend, wall, "wall")
for opening in data["openings"]:
await write_record(backend, opening, opening.kind)
for room in data["rooms"]:
await write_record(backend, room, "room")
@pytest.mark.asyncio
async def test_a_door_schedule_is_a_table_with_title_header_and_one_row_per_door(backend):
await write_plan(backend)
result = await draw_schedule(backend, "doors", at=(10_000.0, 5_000.0), lang="en", scale=50)
assert result["row_count"] == 3
assert result["table_rows"] == 5 # title + header + 3 doors
assert result["representation"] == "composite"
assert result["layer"] == ARCH_ROLE_LAYER["symbol"]
assert result["title"] == vocab("en")["door_schedule"]
# 1:50: widths 15+20+20+20+20+20 = 115 mm -> 5750; rows 5 x 7 mm x 50 = 1750
assert result["bbox"] == {"min": [10_000.0, 3_250.0], "max": [15_750.0, 5_000.0]}
cells = {
e.properties["text"]
for e in await backend.entity_list(
type_filter="MTEXT", layer_filter=ARCH_ROLE_LAYER["symbol"], limit=1000
)
}
assert {"D1", "D2", "D3", "912.5", "w2"} <= cells
@pytest.mark.asyncio
async def test_a_room_schedule_carries_the_total_row(backend):
await write_plan(backend)
result = await draw_schedule(backend, "rooms", at=(0.0, 0.0), lang="tr")
assert result["row_count"] == 4 # three rooms + the total
assert result["table_rows"] == 6
assert result["rows"][-1]["area"] == "45,00 m²"
@pytest.mark.asyncio
async def test_nothing_to_schedule_is_refused_before_anything_is_drawn(backend):
before = await backend.entity_count()
with pytest.raises(ValueError, match=re.escape("kind: this drawing carries no window records")):
await draw_schedule(backend, "windows", at=(0.0, 0.0))
with pytest.raises(
ValueError, match=re.escape("scale: must be a finite number greater than zero")
):
await draw_schedule(backend, "doors", at=(0.0, 0.0), scale=-1)
assert await backend.entity_count() == before
+65
View File
@@ -0,0 +1,65 @@
"""arch_schedule over the MCP wire: rooms labelled by arch_room, scheduled as a TABLE."""
from __future__ import annotations
import pytest
import pytest_asyncio
from fastmcp import Client
import server
from engineering.arch.layers import ARCH_ROLE_LAYER
pytestmark = pytest.mark.asyncio
WALL = ARCH_ROLE_LAYER["wall"]
def rect(x0, y0, x1, y1):
return [((x0, y0), (x1, y0)), ((x1, y0), (x1, y1)), ((x1, y1), (x0, y1)), ((x0, y1), (x0, y0))]
@pytest_asyncio.fixture
async def client(monkeypatch):
"""In-memory client on a fresh headless document - never the operator's AutoCAD."""
monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
async with Client(server.mcp) as connected:
status = (await connected.call_tool("system_status", {})).structured_content or {}
assert status.get("backend") == "ezdxf", "these tests must never touch live AutoCAD"
await connected.call_tool("drawing_new", {})
yield connected
async def draw_two_rooms(client):
"""Rooms (200,200)-(4200,5200) and (4400,200)-(8400,5200) between 200 mm walls."""
for (x1, y1), (x2, y2) in [
*rect(0.0, 0.0, 8600.0, 5400.0),
*rect(200.0, 200.0, 8400.0, 5200.0),
((4200.0, 200.0), (4200.0, 5200.0)),
((4400.0, 200.0), (4400.0, 5200.0)),
]:
await client.call_tool(
"entity_create_line", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "layer": WALL}
)
async def test_a_room_schedule_reads_the_labels_arch_room_wrote(client):
await draw_two_rooms(client)
await client.call_tool("arch_room", {"name": "LIVING", "number": "1", "x": 2200.0, "y": 2700.0})
await client.call_tool("arch_room", {"name": "BED", "number": "2", "x": 6400.0, "y": 2700.0})
result = (
await client.call_tool("arch_schedule", {"kind": "rooms", "x": 9000.0, "y": 5400.0})
).structured_content
assert result["table_rows"] == 5 # title + header + two rooms + total
assert [(r["number"], r["area"]) for r in result["rows"]] == [
("1", "20.00 m²"),
("2", "20.00 m²"),
("", "40.00 m²"),
]
async def test_nothing_to_schedule_is_refused_over_the_wire(client):
result = await client.call_tool(
"arch_schedule", {"kind": "doors", "x": 0.0, "y": 0.0}, raise_on_error=False
)
assert result.is_error is True
assert "kind: this drawing carries no door records" in result.content[0].text
+74
View File
@@ -135,3 +135,77 @@ async def test_server_registers_table_and_mleader_tools():
names = {tool.name for tool in await server._registered_tools()}
assert "entity_create_table" in names
assert "leader_create_mleader" in names
class _MeasuredTable:
"""The IAcadTable members a live AutoCAD 2026 seat exposes (measured).
MEASURED on AutoCAD 2026 (track F task 6): the table has no ``TextHeight``
property - pywin32 answers ``table.TextHeight = h`` with AttributeError -
and a table left alone keeps the Standard style's heights (title 6.0, header
and data 4.5). ``SetCellTextHeight(row, col, h)`` is the member that sets it.
A fake that accepted ``TextHeight`` would hide exactly that defect.
"""
def __init__(self, rows, columns):
self.Handle = "A1"
self.Rows, self.Columns = rows, columns
self._heights = {
(r, c): (6.0 if r == 0 else 4.5) for r in range(rows) for c in range(columns)
}
def __setattr__(self, name, value):
if name == "TextHeight":
raise AttributeError(f"object has no attribute {name!r}")
object.__setattr__(self, name, value)
def SetColumnWidth(self, column, width):
pass
def SetRowHeight(self, row, height):
pass
def SetText(self, row, column, value):
pass
def SetCellTextHeight(self, row, column, height):
self._heights[(row, column)] = float(height)
def GetCellTextHeight(self, row, column):
return self._heights[(row, column)]
async def test_com_table_sets_every_cell_text_height_through_the_measured_member():
backend = ComBackend()
async def run_inline(func, *args, **kwargs):
return func(*args, **kwargs)
backend._run = run_inline
mspace = MagicMock()
made: list[_MeasuredTable] = []
def add_table(_point, rows, columns, _row_height, _column_width):
made.append(_MeasuredTable(rows, columns))
return made[-1]
mspace.AddTable.side_effect = add_table
info = EntityInfo("A1", "ACAD_TABLE", "TEXT", 256, "ByLayer", True, {})
with (
patch("backends.com_backend._msp", return_value=mspace),
patch("backends.com_backend._apoint", side_effect=lambda *p: tuple(p)),
patch("backends.com_backend._entity_info", return_value=info),
patch("backends.com_backend._apply_entity_attrs"),
patch("backends.com_backend._regen"),
):
await backend.entity_create_table(
0, 50, [["A", "1"]], headers=["Name", "Qty"], title="BOM", text_height=125.0
)
(table,) = made
assert {
(row, column): table.GetCellTextHeight(row, column)
for row in range(table.Rows)
for column in range(table.Columns)
} == {(row, column): 125.0 for row in range(3) for column in range(2)}
+10 -3
View File
@@ -495,7 +495,7 @@ async def test_tool_groups_is_byte_identical_to_the_source_declared_grouping():
async def test_tool_group_sizes_are_unchanged():
"""Frozen snapshot of the surface (233 tools, 26 groups).
"""Frozen snapshot of the surface (236 tools, 26 groups).
Taken before @cad_tool landed at 131 tools; `batch` moved 2 -> 3 when
v1.5.0's `cad_batch` joined `entity_batch_create`/`entity_batch_modify`, and
@@ -587,11 +587,18 @@ async def test_tool_group_sizes_are_unchanged():
`arch_dimension_chains`); the `arch` tag is ranked directly after `mech` in
`_GROUP_TAG_PRIORITY`, so those tools file under `architecture` rather
than under their secondary `create` / `dimension` tags.
Track F group R opened SECTION 25 with the rooms: `arch_room` and
`arch_schedule` are tagged `arch` and `create`, `arch_rooms_detect` `arch`
and `query`. `arch` is not in `_GROUP_TAG_PRIORITY` on this branch, so they
file under their secondary tags: `entity_creation` 18 -> 20,
`entity_query` 9 -> 10. This snapshot is branch-local: the track F merge
task recomputes it once the walls, rooms and catalogue branches have landed.
"""
sizes = {label: len(names) for label, names in (await server._tool_groups()).items()}
assert sizes == {
"analysis": 12,
"architecture": 4,
"architecture": 7,
"batch": 3,
"blocks": 9,
"corner_ops": 4,
@@ -617,4 +624,4 @@ async def test_tool_group_sizes_are_unchanged():
"validation": 1,
"view": 4,
}
assert sum(sizes.values()) == 233
assert sum(sizes.values()) == 236