diff --git a/README.md b/README.md index 2020760..9f44eef 100644 --- a/README.md +++ b/README.md @@ -27,8 +27,8 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co -> **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 | -233 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools and the 14 mechanical tools (TOOL_PACKS=core,mech keeps the last) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates. +236 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools and the 14 mechanical tools (TOOL_PACKS=core,mech keeps the last) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates. **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. diff --git a/backends/com_backend.py b/backends/com_backend.py index 07b5a1f..f224529 100644 --- a/backends/com_backend.py +++ b/backends/com_backend.py @@ -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) diff --git a/discovery/aliases.py b/discovery/aliases.py index e07703f..5738f41 100644 --- a/discovery/aliases.py +++ b/discovery/aliases.py @@ -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", + ), + ), } diff --git a/docs/tool-inventory.json b/docs/tool-inventory.json index 47a999b..4a1ade3 100644 --- a/docs/tool-inventory.json +++ b/docs/tool-inventory.json @@ -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", diff --git a/engineering/arch/__init__.py b/engineering/arch/__init__.py index 54238a8..f5f53c3 100644 --- a/engineering/arch/__init__.py +++ b/engineering/arch/__init__.py @@ -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", diff --git a/engineering/arch/faces.py b/engineering/arch/faces.py new file mode 100644 index 0000000..98a006d --- /dev/null +++ b/engineering/arch/faces.py @@ -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 diff --git a/engineering/arch/rooms.py b/engineering/arch/rooms.py new file mode 100644 index 0000000..ab403f4 --- /dev/null +++ b/engineering/arch/rooms.py @@ -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``. + """ + 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, + } diff --git a/engineering/arch/schedule.py b/engineering/arch/schedule.py new file mode 100644 index 0000000..689c205 --- /dev/null +++ b/engineering/arch/schedule.py @@ -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 ```` 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]}, + } diff --git a/server.py b/server.py index a1ac54e..6ac2627 100644 --- a/server.py +++ b/server.py @@ -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."), + ] = 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) ── diff --git a/tests/test_arch_faces.py b/tests/test_arch_faces.py new file mode 100644 index 0000000..620dcc2 --- /dev/null +++ b/tests/test_arch_faces.py @@ -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) diff --git a/tests/test_arch_rooms.py b/tests/test_arch_rooms.py new file mode 100644 index 0000000..e5c22c0 --- /dev/null +++ b/tests/test_arch_rooms.py @@ -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", + } + ] diff --git a/tests/test_arch_rooms_tools.py b/tests/test_arch_rooms_tools.py new file mode 100644 index 0000000..b09cd99 --- /dev/null +++ b/tests/test_arch_rooms_tools.py @@ -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 diff --git a/tests/test_arch_schedule.py b/tests/test_arch_schedule.py new file mode 100644 index 0000000..02076ab --- /dev/null +++ b/tests/test_arch_schedule.py @@ -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 diff --git a/tests/test_arch_schedule_tools.py b/tests/test_arch_schedule_tools.py new file mode 100644 index 0000000..2c52713 --- /dev/null +++ b/tests/test_arch_schedule_tools.py @@ -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 diff --git a/tests/test_table_leader.py b/tests/test_table_leader.py index d43d8e7..687eb68 100644 --- a/tests/test_table_leader.py +++ b/tests/test_table_leader.py @@ -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)} diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py index c6aad53..c1125be 100644 --- a/tests/test_tool_registry.py +++ b/tests/test_tool_registry.py @@ -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