diff --git a/README.md b/README.md index 5f00738..303ea23 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:** 241 tools · 8 resources · 5 prompt templates · 4516 collected tests. -> 241 is the **registered** count; a default install advertises 236 over `tools/list`, +> **v1.5 release snapshot:** 243 tools · 8 resources · 5 prompt templates · 4516 collected tests. +> 243 is the **registered** count; a default install advertises 238 over `tools/list`, > because `ENABLE_3D` is unset. `system_about` is the runtime authority. ## Why this exists @@ -122,7 +122,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 | -241 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) 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. +243 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) 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/discovery/aliases.py b/discovery/aliases.py index 7286ead..9e00ef7 100644 --- a/discovery/aliases.py +++ b/discovery/aliases.py @@ -2465,6 +2465,54 @@ TOOL_ALIASES: dict[str, ToolAliases] = { "mimari plan", ), ), + # ── Understanding & QA (SECTION 26, track H group U) ──────────────────── + # No AutoCAD command reads a foreign drawing back as a report, so `acad` + # is empty. UNITS stays with drawing_settings (it sets units; this reads + # what the geometry implies), and no synonym here says "layers", "blocks" + # or "how big", the words the golden layer_list / block_list / + # analysis_bounding_box queries lean on. + "drawing_understand": ToolAliases( + acad=(), + synonyms=( + "understand this drawing", + "read a foreign drawing", + "summarise a client drawing", + "drawing report", + "which unit is this drawing really in", + "wrong insunits", + "inches or millimetres", + "two copies of the plan", + "stray entity far away", + "what is on this p&id", + "birim", + "birimi ne", + "çizimi anla", + "cizimi anla", + "çizim raporu", + "cizim raporu", + "yabancı çizim", + "yabanci cizim", + ), + ), + "drawing_scale_check": ToolAliases( + acad=(), + synonyms=( + "scale check", + "is the p&id to scale", + "is this drawing to scale", + "schematic or to scale", + "unscaled p&id", + "tag distances against the layout", + "can i measure pipe on the p&id", + "ölçek", + "olcek", + "ölçek kontrolü", + "olcek kontrolu", + "ölçekli mi", + "olcekli mi", + "масштаб", + ), + ), } diff --git a/docs/tool-inventory.json b/docs/tool-inventory.json index 6874746..f7f220a 100644 --- a/docs/tool-inventory.json +++ b/docs/tool-inventory.json @@ -3,13 +3,13 @@ "schema_version": 1, "package_version": "1.5.1", "totals": { - "tools": 241, + "tools": 243, "resources": 8, "prompts": 5, "groups": 26 }, "groups": { - "analysis": 12, + "analysis": 14, "architecture": 12, "batch": 3, "blocks": 9, @@ -1116,6 +1116,20 @@ "title": "Save As" } }, + { + "name": "drawing_scale_check", + "group": "analysis", + "cost": "read", + "summary": "Is this drawing to scale against a reference? Tag distances, ratios, a verdict.", + "tags": [ + "analysis", + "query" + ], + "annotations": { + "title": "Understand: Scale Check", + "readOnlyHint": true + } + }, { "name": "drawing_settings", "group": "system", @@ -1161,6 +1175,20 @@ "destructiveHint": false } }, + { + "name": "drawing_understand", + "group": "analysis", + "cost": "read", + "summary": "Read a foreign drawing: units, extents, clusters, layers, tags, rooms, languages.", + "tags": [ + "analysis", + "query" + ], + "annotations": { + "title": "Understand: Describe Drawing", + "readOnlyHint": true + } + }, { "name": "drawing_undo", "group": "drawing", diff --git a/engineering/understand/clusters.py b/engineering/understand/clusters.py new file mode 100644 index 0000000..0722e4c --- /dev/null +++ b/engineering/understand/clusters.py @@ -0,0 +1,194 @@ +"""Separate bodies of model-space content: sheets, plan copies, details. + +A foreign model space is often several drawings side by side - two copies of +one plan, a detail beside it, a legend. Every reader after this one must know +which body a tag or a room belongs to, or it matches a tag in one copy against +the same tag in the other. + +The algorithm, in the order it runs: + +1. **Raster.** The box of the records is divided into square cells of + ``cell`` drawing units (default :data:`CELL_FRACTION` of its diagonal, so + the grid never exceeds about 50 x 50 cells). Every record marks the cells + its box covers. +2. **Connect.** Occupied cells that touch, corners included, are one + component; a record belongs to the component of its cells. Two bodies stay + apart when an empty cell separates them - a gap of two cells is always + enough. +3. **Absorb.** Components whose boxes intersect are merged, repeatedly: a tag + floating in the middle of a room lies inside the plan's box and belongs to + the plan, not to a cluster of its own. + +Ids are ``C1``, ``C2``, ... by entity count, largest first (ties left to right, +then bottom to top). Outliers are the caller's to exclude: one entity thousands of +kilometres away would otherwise set the cell size. Pure; the drawing is never touched. +""" + +from __future__ import annotations + +import math +from collections.abc import Iterable, Sequence + +from engineering.understand.snapshot import EntityRecord +from engineering.understand.units import entity_box + +__all__ = ["CELL_FRACTION", "find_clusters"] + +#: Cell side as a fraction of the content's diagonal. +CELL_FRACTION = 0.02 +#: Labels listed per cluster. +LABELS_PER_CLUSTER = 5 + +_TEXT_TYPES = frozenset({"TEXT", "MTEXT", "MULTILEADER"}) + + +def _intersect(a, b) -> bool: + return a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3] + + +def _merge(a, b): + return (min(a[0], b[0]), min(a[1], b[1]), max(a[2], b[2]), max(a[3], b[3])) + + +def _root(parent: list[int], i: int) -> int: + """Union-find root with path halving.""" + while parent[i] != i: + parent[i] = parent[parent[i]] + i = parent[i] + return i + + +def _absorb(bodies: list[tuple[tuple, list[int]]]) -> list[tuple[tuple, list[int]]]: + """Merge bodies whose boxes intersect until none do. + + One pass is a sweep over boxes sorted by their left edge (a pair can only + intersect while the next left edge is inside the current box) feeding a + union-find; a merged box can reach a body it did not touch before, so + passes repeat until one merges nothing. + """ + while True: + order = sorted(range(len(bodies)), key=lambda i: bodies[i][0][0]) + parent = list(range(len(bodies))) + joined = False + for n, i in enumerate(order): + box = bodies[i][0] + for j in order[n + 1 :]: + if bodies[j][0][0] > box[2]: + break + ri, rj = _root(parent, i), _root(parent, j) + if ri != rj and _intersect(box, bodies[j][0]): + parent[rj] = ri + joined = True + if not joined: + return bodies + merged: dict[int, tuple[tuple, list[int]]] = {} + for i, (box, members) in enumerate(bodies): + r = _root(parent, i) + if r in merged: + merged[r] = (_merge(merged[r][0], box), merged[r][1] + members) + else: + merged[r] = (box, list(members)) + bodies = list(merged.values()) + + +def find_clusters( + records: Sequence[EntityRecord], + *, + exclude: Iterable[str] = (), + cell: float | None = None, +) -> tuple[list[dict], dict[str, str]]: + """``(clusters, membership)``: one row per body of content, and handle -> id. + + A row is ``{"id", "box": [x0, y0, x1, y1], "size": [w, h], "entities", + "layers": [the three commonest], "labels": [the tallest texts]}``. + ``exclude`` names handles to leave out (the outliers). A record without a + box (no geometry the snapshot could read) belongs to no cluster. + """ + skip = set(exclude) + items = [(rec, box) for rec in records if rec.handle not in skip and (box := entity_box(rec))] + if not items: + return [], {} + x0 = min(b[0] for _r, b in items) + y0 = min(b[1] for _r, b in items) + x1 = max(b[2] for _r, b in items) + y1 = max(b[3] for _r, b in items) + if cell is None: + diag = math.hypot(x1 - x0, y1 - y0) + cell = diag * CELL_FRACTION if diag > 0.0 else 1.0 + if not cell > 0.0: + raise ValueError(f"cell: must be greater than zero, got {cell!r}") + + def span(lo: float, hi: float, origin: float) -> range: + return range( + int(math.floor((lo - origin) / cell)), int(math.floor((hi - origin) / cell)) + 1 + ) + + owner: dict[tuple[int, int], list[int]] = {} + first_cell: list[tuple[int, int]] = [] + for k, (_rec, b) in enumerate(items): + cols, rows = span(b[0], b[2], x0), span(b[1], b[3], y0) + first_cell.append((cols[0], rows[0])) + for i in cols: + for j in rows: + owner.setdefault((i, j), []).append(k) + + component: dict[tuple[int, int], int] = {} + count = 0 + for start in owner: + if start in component: + continue + component[start] = count + stack = [start] + while stack: + ci, cj = stack.pop() + for di in (-1, 0, 1): + for dj in (-1, 0, 1): + nxt = (ci + di, cj + dj) + if nxt in owner and nxt not in component: + component[nxt] = count + stack.append(nxt) + count += 1 + + groups: dict[int, list[int]] = {} + for k in range(len(items)): + groups.setdefault(component[first_cell[k]], []).append(k) + bodies = [] + for members in groups.values(): + box = items[members[0]][1] + for k in members[1:]: + box = _merge(box, items[k][1]) + bodies.append((box, members)) + bodies = _absorb(bodies) + + bodies.sort(key=lambda body: (-len(body[1]), body[0][0], body[0][1])) + clusters: list[dict] = [] + membership: dict[str, str] = {} + for n, (box, members) in enumerate(bodies, start=1): + cid = f"C{n}" + layers: dict[str, int] = {} + texts = [] + for k in members: + rec = items[k][0] + membership[rec.handle] = cid + layers[rec.layer] = layers.get(rec.layer, 0) + 1 + if rec.type in _TEXT_TYPES and rec.text and rec.text.strip(): + texts.append((-(rec.height or 0.0), rec.text.strip())) + labels: list[str] = [] + for _h, text in sorted(texts): + if text not in labels: + labels.append(text) + if len(labels) == LABELS_PER_CLUSTER: + break + clusters.append( + { + "id": cid, + "box": list(box), + "size": [box[2] - box[0], box[3] - box[1]], + "entities": len(members), + "layers": [ + name for name, _c in sorted(layers.items(), key=lambda kv: (-kv[1], kv[0]))[:3] + ], + "labels": labels, + } + ) + return clusters, membership diff --git a/engineering/understand/describe.py b/engineering/understand/describe.py new file mode 100644 index 0000000..1cd2f83 --- /dev/null +++ b/engineering/understand/describe.py @@ -0,0 +1,463 @@ +"""drawing_understand: one read of a foreign drawing, every claim with its evidence. + +`describe` assembles the report from the pure readers - units and extents +(`units`), bodies of content (`clusters`), and here: layers and what they +carry, equipment tags, rooms, text languages, blocks, layouts and the title +block. Each claim carries its evidence (what was measured or matched) and a +confidence; nothing is corrected silently, and the drawing is never modified +(the snapshot is a frozen copy). + +Confidences, in one place: + +* a layer name matching the vocabulary: `vocab.classify_layer`'s own (0.9 for + a direct keyword); the texts on an unnamed layer voting instead: + :data:`TEXT_VOTE_CONFIDENCE` when most of them agree, :data:`WEAK_VOTE_CONFIDENCE` + otherwise; +* a tag read from a block attribute :data:`ATTRIBUTE_TAG_CONFIDENCE`, from a + text :data:`TEXT_TAG_CONFIDENCE`; +* a room label that sits in a face of the wall layers + :data:`ROOM_FACE_CONFIDENCE`, a label alone :data:`ROOM_LABEL_CONFIDENCE`; +* a title block whose name says so :data:`TITLE_NAME_CONFIDENCE`, one chosen by + its attribute count alone :data:`TITLE_ATTRIBUTE_CONFIDENCE`; +* character classes are exact: a text either holds a Cyrillic letter or not. +""" + +from __future__ import annotations + +import math +import re +from collections import Counter +from collections.abc import Sequence + +from engineering.arch.faces import Face, face_containing, planar_faces +from engineering.pid.graph import ATTRIBUTE_TAGS, KEYWORDS +from engineering.understand.clusters import find_clusters +from engineering.understand.labels import parse_tag +from engineering.understand.snapshot import EntityRecord, Snapshot, length_of +from engineering.understand.units import infer_units, robust_extents, wall_segments +from engineering.understand.vocab import classify_layer, room_label + +__all__ = [ + "describe", + "find_rooms", + "prepare", + "tag_occurrences", + "text_language", +] + +TEXT_VOTE_CONFIDENCE = 0.6 +WEAK_VOTE_CONFIDENCE = 0.4 +ATTRIBUTE_TAG_CONFIDENCE = 0.95 +TEXT_TAG_CONFIDENCE = 0.8 +ROOM_FACE_CONFIDENCE = 0.9 +ROOM_LABEL_CONFIDENCE = 0.6 +TITLE_NAME_CONFIDENCE = 0.9 +TITLE_ATTRIBUTE_CONFIDENCE = 0.6 +#: The face finder pairs every wall segment with every other; above this many +#: the rooms are listed without faces and a warning says so. +FACE_SEGMENT_CAP = 3000 +#: Endpoints closer than this fraction of the robust diagonal are one corner +#: (1 mm on a 50 m plan drawn in millimetres). +FACE_TOL_FRACTION = 2e-5 +#: The same tag lettered twice closer than this fraction of the robust diagonal +#: (an attribute and a text beside one symbol) is one occurrence. +TAG_MERGE_FRACTION = 0.01 +#: Rows listed for named blocks, most inserted first. +BLOCK_ROW_CAP = 100 +#: Block-name words that mark a title block (EN / NL / DE / TR / RU). +TITLE_WORDS = ("TITLE", "TITEL", "STEMPEL", "SCHRIFTFELD", "ANTET", "ШТАМП") + +_TEXT_TYPES = frozenset({"TEXT", "MTEXT", "MULTILEADER"}) +_TOKEN_RE = re.compile(r"[A-Z]+") +_CYRILLIC = re.compile(r"[\u0400-\u04FF]") # the Cyrillic block +#: Letters only Turkish (of the five languages) writes; ç, ö and ü are shared +#: with German and French and are not evidence on their own. +_TURKISH = re.compile(r"[ĞğİıŞş]") +_LATIN = re.compile(r"[A-Za-z\u00C0-\u024F]") # ASCII and Latin-1 to Latin Extended-B + + +def _anchor(rec: EntityRecord) -> tuple[float, float] | None: + if rec.points: + return (float(rec.points[0][0]), float(rec.points[0][1])) + if rec.bbox is not None: + b = rec.bbox + return ((b[0] + b[2]) / 2.0, (b[1] + b[3]) / 2.0) + return None + + +def prepare(snap: Snapshot) -> dict: + """The shared first pass: model-space records, extents and outliers, the + records kept, clusters and membership. `describe` and `scale.scale_check` + both start here, so they agree on which entity is an outlier and which + cluster a tag sits in.""" + model = [r for r in snap.records if r.space == "Model"] + extents = robust_extents(model) + outliers = {o["handle"] for o in extents["outliers"]} + kept = [r for r in model if r.handle not in outliers] + clusters, membership = find_clusters(kept) + robust = extents["robust"] + diag = math.hypot(robust[2] - robust[0], robust[3] - robust[1]) if robust else 0.0 + return { + "model": model, + "extents": extents, + "kept": kept, + "clusters": clusters, + "membership": membership, + "diag": diag, + } + + +def tag_occurrences(snap: Snapshot, *, prep: dict | None = None) -> list[dict]: + """Every equipment tag on the drawing, one row per occurrence. + + A tag is read with `labels.parse_tag` from a TEXT / MTEXT / MULTILEADER, + or from an INSERT attribute named like a tag (`pid.graph.ATTRIBUTE_TAGS`). + Two readings of one tag in one space closer than + :data:`TAG_MERGE_FRACTION` of the diagonal are one occurrence (the + attribute's position and confidence win). Rows carry ``tag``, ``handle``, + ``at``, ``space``, ``layer``, ``cluster`` (model space only), ``source`` + and ``confidence``. + """ + prep = prep or prepare(snap) + radius = prep["diag"] * TAG_MERGE_FRACTION + found: list[dict] = [] + for rec in snap.records: + at = _anchor(rec) + if at is None: + continue + readings = [] + if rec.type in _TEXT_TYPES and rec.text: + readings.append((parse_tag(rec.text), "text", TEXT_TAG_CONFIDENCE)) + elif rec.type == "INSERT": + for name, value in rec.attribs: + if name.upper() in ATTRIBUTE_TAGS and value: + readings.append((parse_tag(value), "attribute", ATTRIBUTE_TAG_CONFIDENCE)) + for tag, source, confidence in readings: + if not tag: + continue + found.append( + { + "tag": tag, + "handle": rec.handle, + "at": [at[0], at[1]], + "space": rec.space, + "layer": rec.layer, + "cluster": prep["membership"].get(rec.handle) if rec.space == "Model" else None, + "source": source, + "confidence": confidence, + } + ) + found.sort(key=lambda o: (-o["confidence"], o["tag"], o["handle"])) + kept: list[dict] = [] + for occ in found: + twin = next( + ( + k + for k in kept + if k["tag"] == occ["tag"] + and k["space"] == occ["space"] + and math.dist(k["at"], occ["at"]) <= radius + ), + None, + ) + if twin is None: + kept.append(occ) + kept.sort(key=lambda o: (o["tag"], o["space"], o["at"][0], o["at"][1])) + return kept + + +def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict) -> dict: + """Room labels and the faces of the wall layers they sit in. + + A label is any text `vocab.room_label` recognises. Wall layers are the + ones `vocab.classify_layer` files under ``architecture``; their straight + segments go through `engineering.arch.faces.planar_faces` (arcs are left + out and counted). ``labelled`` pairs each room row with its `Face` for a + caller that needs the geometry; ``rooms`` is JSON-ready. + """ + labels = [] + for rec in records: + if rec.type in _TEXT_TYPES and rec.text: + parsed = room_label(rec.text) + at = _anchor(rec) + if parsed and at is not None: + labels.append((rec, parsed, at)) + segments, bulged = wall_segments(records) + faces: tuple[Face, ...] = () + skipped = None + if len(segments) > FACE_SEGMENT_CAP: + skipped = ( + f"{len(segments)} wall segments exceed the {FACE_SEGMENT_CAP}-segment cap of the " + "face finder; rooms are listed by label only" + ) + elif segments: + faces = planar_faces(segments, tol=tol) + rooms: list[dict] = [] + labelled: list[tuple[dict, Face]] = [] + used: set[int] = set() + for rec, parsed, at in labels: + face = face_containing(faces, at) if faces else None + row = { + "number": parsed.get("number"), + "name": parsed.get("name"), + "language": parsed.get("language"), + "text": rec.text, + "handle": rec.handle, + "at": [at[0], at[1]], + "cluster": membership.get(rec.handle), + "face": None, + "source": "label", + "confidence": ROOM_LABEL_CONFIDENCE, + } + if face is not None: + used.add(id(face)) + row["face"] = { + "area": face.area, + "polygon": [[p[0], p[1]] for p in face.loop], + "centroid": [face.centroid[0], face.centroid[1]], + } + row["source"] = "label+face" + row["confidence"] = ROOM_FACE_CONFIDENCE + labelled.append((row, face)) + rooms.append(row) + rooms.sort(key=lambda r: (r["cluster"] or "", r["number"] or "", r["name"] or "", r["at"])) + return { + "rooms": rooms, + "labelled": labelled, + "faces": len(faces), + "faces_unlabelled": sum(1 for f in faces if id(f) not in used), + "wall_segments": len(segments), + "arcs_skipped": bulged, + "skipped": skipped, + } + + +def text_language(text: str) -> str: + """``cyrillic`` | ``turkish`` | ``latin`` | ``none`` by character class, in + that precedence (a Russian label with a Latin tag in it is Cyrillic).""" + if _CYRILLIC.search(text): + return "cyrillic" + if _TURKISH.search(text): + return "turkish" + if _LATIN.search(text): + return "latin" + return "none" + + +def _layers(snap: Snapshot, unit_mm: float | None) -> list[dict]: + counts: Counter = Counter() + lengths: dict[str, float] = {} + texts: dict[str, list[str]] = {} + for rec in snap.records: + counts[rec.layer] += 1 + lengths[rec.layer] = lengths.get(rec.layer, 0.0) + length_of(rec) + if rec.type in _TEXT_TYPES and rec.text: + texts.setdefault(rec.layer, []).append(rec.text) + rows = [] + for name in sorted(set(snap.layers) | set(counts)): + props = snap.layers.get(name, {}) + c = classify_layer(name) + row = { + "name": name, + "entities": counts.get(name, 0), + "length": lengths.get(name, 0.0), + "length_m": lengths.get(name, 0.0) * unit_mm / 1000.0 if unit_mm else None, + "color": props.get("color"), + "linetype": props.get("linetype"), + "discipline": c.get("discipline", "unknown"), + "service": c.get("service", "unknown"), + "keyword": c.get("keyword"), + "language": c.get("language"), + "confidence": c.get("confidence", 0.0), + "evidence": f"layer name keyword {c.get('keyword')!r}" if c.get("keyword") else None, + } + if row["discipline"] == "unknown" and row["service"] == "unknown": + votes: Counter = Counter() + said = texts.get(name, []) + for text in said: + t = classify_layer(text) + if ( + t.get("discipline", "unknown") != "unknown" + or t.get("service", "unknown") != "unknown" + ): + votes[(t.get("discipline", "unknown"), t.get("service", "unknown"))] += 1 + if votes: + (discipline, service), n = sorted(votes.items(), key=lambda kv: (-kv[1], kv[0]))[0] + row.update( + discipline=discipline, + service=service, + confidence=( + TEXT_VOTE_CONFIDENCE if n * 2 > len(said) else WEAK_VOTE_CONFIDENCE + ), + evidence=f"{n} of {len(said)} texts on the layer read as {service} / {discipline}", + ) + rows.append(row) + rows.sort(key=lambda r: (-r["entities"], r["name"])) + return rows + + +def _block_kind(name: str) -> str | None: + for token in _TOKEN_RE.findall(name.upper()): + kind = KEYWORDS.get(token) + if kind: + return kind + return None + + +def _blocks(snap: Snapshot) -> dict: + inserts = [r for r in snap.records if r.type == "INSERT" and r.block] + named = Counter(r.block for r in inserts if not r.block.startswith("*")) + anonymous = Counter(r.block for r in inserts if r.block.startswith("*")) + rows = [ + { + "name": name, + "inserts": n, + "kind": _block_kind(name), + "discipline": classify_layer(name).get("discipline", "unknown"), + } + for name, n in sorted(named.items(), key=lambda kv: (-kv[1], kv[0])) + ] + kinds = Counter(row["kind"] for row in rows if row["kind"]) + return { + "inserts": len(inserts), + "named": rows[:BLOCK_ROW_CAP], + "named_total": len(rows), + "anonymous": {"names": len(anonymous), "inserts": sum(anonymous.values())}, + "kinds": dict(sorted(kinds.items())), + } + + +def _title_block(snap: Snapshot) -> dict: + best = None + for rec in snap.records: + if rec.type != "INSERT" or len(rec.attribs) < 3: + continue + upper = (rec.block or "").upper() + named = any(word in upper for word in TITLE_WORDS) + key = (named, rec.space != "Model", len(rec.attribs)) + if best is None or key > best[0]: + best = (key, rec, named) + if best is None: + return {"found": False, "reason": "no block reference with three or more attributes"} + _key, rec, named = best + return { + "found": True, + "handle": rec.handle, + "block": rec.block, + "space": rec.space, + "attributes": {tag: value for tag, value in rec.attribs}, + "confidence": TITLE_NAME_CONFIDENCE if named else TITLE_ATTRIBUTE_CONFIDENCE, + "evidence": ( + f"block name {rec.block!r} reads as a title block" + if named + else f"the block reference with the most attributes ({len(rec.attribs)})" + ), + } + + +def _languages(snap: Snapshot, layer_rows: list[dict]) -> dict: + by_class: Counter = Counter() + samples: dict[str, list[str]] = {} + for rec in snap.records: + if rec.type in _TEXT_TYPES and rec.text: + lang = text_language(rec.text) + by_class[lang] += 1 + if len(samples.setdefault(lang, [])) < 3: + samples[lang].append(rec.handle) + total = sum(by_class.values()) + layer_langs = Counter(row["language"] for row in layer_rows if row.get("language")) + return { + "texts": total, + "by_class": {k: by_class.get(k, 0) for k in ("latin", "turkish", "cyrillic", "none")}, + "share": { + k: (by_class.get(k, 0) / total if total else 0.0) + for k in ("latin", "turkish", "cyrillic", "none") + }, + "samples": samples, + "layer_name_languages": dict(sorted(layer_langs.items())), + "confidence": 1.0, + } + + +def describe(snap: Snapshot) -> dict: + """The whole report; see the module docstring for the confidences.""" + prep = prepare(snap) + warnings: list[str] = [] + tol = prep["diag"] * FACE_TOL_FRACTION if prep["diag"] > 0 else 1.0 + rooms = find_rooms(prep["kept"], tol=tol, membership=prep["membership"]) + units = infer_units( + snap, + records=prep["kept"], + room_areas=[row["face"]["area"] for row, _face in rooms["labelled"]], + ) + if units["warning"]: + warnings.append(units["warning"]) + extents = dict(prep["extents"]) + extents["declared"] = ( + {"min": list(snap.extmin), "max": list(snap.extmax)} + if snap.extmin is not None and snap.extmax is not None + else None + ) + if extents["outliers"]: + handles = ", ".join(o["handle"] for o in extents["outliers"][:5]) + stretch = extents["stretch"] + warnings.append( + f"{len(extents['outliers'])} entit{'y' if len(extents['outliers']) == 1 else 'ies'} " + f"far outside the content stretch the extents" + + (f" {stretch:.3g}x" if stretch else "") + + f": {handles}" + ) + if rooms["skipped"]: + warnings.append(rooms["skipped"]) + + occurrences = tag_occurrences(snap, prep=prep) + by_tag: dict[str, list[dict]] = {} + for occ in occurrences: + by_tag.setdefault(occ["tag"], []).append(occ) + tags = [ + { + "tag": tag, + "count": len(rows), + "ambiguous": len(rows) > 1, + "confidence": max(r["confidence"] for r in rows), + "occurrences": [ + {k: r[k] for k in ("handle", "at", "space", "cluster", "layer", "source")} + for r in rows + ], + } + for tag, rows in sorted(by_tag.items()) + ] + repeated = [t["tag"] for t in tags if t["ambiguous"]] + if repeated: + warnings.append( + f"{len(repeated)} tag(s) appear more than once (plan copies, legends, second " + f"sheets): {', '.join(repeated[:5])}" + ) + + layer_rows = _layers(snap, units["mm_per_unit"]) + layout_names = list(snap.layouts) + if "Model" not in layout_names: + layout_names.insert(0, "Model") + space_counts = Counter(r.space for r in snap.records) + return { + "source": snap.source, + "entities": len(snap.records), + "units": units, + "extents": extents, + "clusters": prep["clusters"], + "layers": layer_rows, + "equipment_tags": tags, + "rooms": rooms["rooms"], + "room_faces": { + "faces": rooms["faces"], + "unlabelled": rooms["faces_unlabelled"], + "wall_segments": rooms["wall_segments"], + "arcs_skipped": rooms["arcs_skipped"], + "tolerance": tol, + }, + "languages": _languages(snap, layer_rows), + "blocks": _blocks(snap), + "layouts": [{"name": n, "entities": space_counts.get(n, 0)} for n in layout_names], + "title_block": _title_block(snap), + "warnings": warnings, + } diff --git a/engineering/understand/scale.py b/engineering/understand/scale.py new file mode 100644 index 0000000..a649793 --- /dev/null +++ b/engineering/understand/scale.py @@ -0,0 +1,272 @@ +"""drawing_scale_check: is this drawing to scale against that one? + +A P&ID and a layout of one plant carry the same equipment tags. If the P&ID +were drawn to scale, the distance between two tanks on it would be the distance +on the layout times one factor, for every pair. On a schematic it is not - and +a pipe length measured on a schematic is a length of drawn line, not of pipe. + +The check, in the order it runs: + +1. **Scope.** Model-space tags are read (`describe.tag_occurrences`) and the + drawings split into clusters (`clusters.find_clusters`, outliers left out). + Every pair of scopes - the whole drawing or one of its eight largest + clusters, on each side - is tried, and the pair that matches the most tags + wins (ties: the whole drawing first, then the larger cluster). Two copies + of one plan side by side therefore resolve to one copy instead of making + every tag ambiguous. +2. **Match.** Only a tag found exactly once in each scope is matched. A tag + found more than once in either (a plan copy, a legend, a second sheet) is + ``ambiguous`` and takes no part: picking one of its positions would be a + guess. +3. **Pairs.** For every two matched tags whose distance on the reference + drawing ``b`` is at least ``min_pair_mm``, the ratio of their distance on + ``a`` to that on ``b`` is taken, both converted to millimetres by each + drawing's inferred unit (`units.infer_units`). +4. **Statistics.** The median ratio, the quartiles and their range, and the + share of pairs within +/-10 % of the median - overall, and per room of the + reference drawing (the labelled faces `describe.find_rooms` finds) or, for + tags in no room, per cluster. +5. **Verdict.** ``to_scale`` when at least 80 % of the pairs are within the + band (``factor`` is then the median: a drawing at a different but uniform + scale is recognised); ``schematic`` at 50 % or less; ``partly`` in between; + ``insufficient`` with fewer than three matched tags or no pair above the + floor. + +A tag label is not an equipment centre: the band and the floor are there to +absorb the offset between the two, and the result says so. Pure; neither +drawing is touched. +""" + +from __future__ import annotations + +import math +import statistics +from itertools import combinations + +from engineering.arch.faces import face_containing +from engineering.understand.describe import FACE_TOL_FRACTION, find_rooms, prepare, tag_occurrences +from engineering.understand.snapshot import Snapshot +from engineering.understand.units import infer_units + +__all__ = ["MIN_TAGS", "SCHEMATIC_SHARE", "TO_SCALE_SHARE", "WITHIN", "match_tags", "scale_check"] + +#: The band around the median a pair must fall in to count as consistent. +WITHIN = 0.10 +TO_SCALE_SHARE = 0.80 +SCHEMATIC_SHARE = 0.50 +MIN_TAGS = 3 +#: Clusters per drawing tried as a scope, largest first. +MAX_SCOPE_CLUSTERS = 8 +#: Pair rows listed in the result, farthest from the median first. +PAIR_ROW_CAP = 200 +NOTE = ( + "Tag labels are not equipment centres; the +/-10 % band and the min_pair_mm floor absorb " + "the offset between a label and what it labels." +) + + +def _side(snap: Snapshot) -> dict: + prep = prepare(snap) + prep["tags"] = [o for o in tag_occurrences(snap, prep=prep) if o["space"] == "Model"] + return prep + + +def _scoped(side: dict, cluster: str | None, name: str) -> list[dict]: + if cluster is None: + return side["tags"] + ids = [c["id"] for c in side["clusters"]] + if cluster not in ids: + raise ValueError(f"{name}: no cluster {cluster!r}; the drawing has {ids}") + return [o for o in side["tags"] if o["cluster"] == cluster] + + +def _match(side_a: dict, side_b: dict, cluster_a: str | None, cluster_b: str | None) -> dict: + occ_a = _scoped(side_a, cluster_a, "cluster_a") + occ_b = _scoped(side_b, cluster_b, "cluster_b") + count_a: dict[str, list[dict]] = {} + count_b: dict[str, list[dict]] = {} + for o in occ_a: + count_a.setdefault(o["tag"], []).append(o) + for o in occ_b: + count_b.setdefault(o["tag"], []).append(o) + matched = { + tag: (tuple(count_a[tag][0]["at"]), tuple(count_b[tag][0]["at"])) + for tag in sorted(set(count_a) & set(count_b)) + if len(count_a[tag]) == 1 and len(count_b[tag]) == 1 + } + ambiguous = sorted( + tag + for tag in set(count_a) | set(count_b) + if len(count_a.get(tag, ())) > 1 or len(count_b.get(tag, ())) > 1 + ) + return { + "matched": matched, + "ambiguous": ambiguous, + "occurrences": { + tag: {"a": len(count_a.get(tag, ())), "b": len(count_b.get(tag, ()))} + for tag in ambiguous + }, + "only_a": sorted(t for t in count_a if t not in count_b and len(count_a[t]) == 1), + "only_b": sorted(t for t in count_b if t not in count_a and len(count_b[t]) == 1), + "cluster_a": cluster_a, + "cluster_b": cluster_b, + } + + +def match_tags(a: Snapshot, b: Snapshot, *, cluster_a=None, cluster_b=None) -> dict: + """``{"matched": {tag: (pa, pb)}, "ambiguous": [tag, ...], ...}`` within the + given scopes (a cluster id such as ``"C1"``, or None for the whole model + space). ``pa`` / ``pb`` are the tag's positions in drawing units. Also + ``occurrences`` (per ambiguous tag, how often each side has it), + ``only_a`` / ``only_b`` and the scopes used. An unknown cluster id is + refused with the ids the drawing has.""" + return _match(_side(a), _side(b), cluster_a, cluster_b) + + +def _stats(ratios: list[float]) -> dict: + if not ratios: + return {"median": None, "q1": None, "q3": None, "iqr": None, "within_10pct": None} + median = statistics.median(ratios) + if len(ratios) >= 2: + q1, _q2, q3 = statistics.quantiles(ratios, n=4, method="inclusive") + else: + q1 = q3 = ratios[0] + within = sum(1 for r in ratios if abs(r - median) <= WITHIN * median) / len(ratios) + return {"median": median, "q1": q1, "q3": q3, "iqr": q3 - q1, "within_10pct": within} + + +def _verdict(tag_count: int, pair_count: int, stats: dict) -> str: + if tag_count < MIN_TAGS or pair_count == 0 or not stats["median"] or stats["median"] <= 0: + return "insufficient" + if stats["within_10pct"] >= TO_SCALE_SHARE: + return "to_scale" + if stats["within_10pct"] <= SCHEMATIC_SHARE: + return "schematic" + return "partly" + + +def _summary(tags: list[str], rows: list[dict]) -> dict: + stats = _stats([r["ratio"] for r in rows]) + verdict = _verdict(len(tags), len(rows), stats) + return { + "verdict": verdict, + "factor": stats["median"] if verdict == "to_scale" else None, + **stats, + "pair_count": len(rows), + } + + +def scale_check(a: Snapshot, b: Snapshot, *, min_pair_mm=2000.0) -> dict: + """Is drawing ``a`` to scale against the reference drawing ``b``? + + Returns ``{"verdict", "factor", "median", "q1", "q3", "iqr", "within_10pct", + "pair_count", "pairs", "pairs_truncated", "pairs_below_floor", "matched", + "ambiguous", "occurrences", "only_a", "only_b", "groups", "scope", "units", + "min_pair_mm", "thresholds", "notes"}``. ``pairs`` lists at most 200 rows, + farthest from the median first; ``matched`` is ``[{"tag", "a": [x, y], + "b": [x, y]}]`` in drawing units. Refused before anything is computed: a + ``min_pair_mm`` that is not a positive finite number. + """ + try: + floor = float(min_pair_mm) + except (TypeError, ValueError): + raise ValueError(f"min_pair_mm: expected a number, got {min_pair_mm!r}") from None + if not math.isfinite(floor) or floor <= 0.0: + raise ValueError(f"min_pair_mm: must be a positive number of millimetres, got {floor:g}") + side_a, side_b = _side(a), _side(b) + + best = None + for ca in [None] + [c["id"] for c in side_a["clusters"][:MAX_SCOPE_CLUSTERS]]: + for cb in [None] + [c["id"] for c in side_b["clusters"][:MAX_SCOPE_CLUSTERS]]: + m = _match(side_a, side_b, ca, cb) + if best is None or len(m["matched"]) > len(best["matched"]): + best = m + + tol_b = side_b["diag"] * FACE_TOL_FRACTION if side_b["diag"] > 0 else 1.0 + rooms_b = find_rooms(side_b["kept"], tol=tol_b, membership=side_b["membership"]) + units = {} + notes = [NOTE] + for key, snap, side, rooms in ( + ("a", a, side_a, None), + ("b", b, side_b, rooms_b), + ): + areas = [row["face"]["area"] for row, _f in rooms["labelled"]] if rooms else [] + u = infer_units(snap, records=side["kept"], room_areas=areas) + units[key] = { + "unit": u["inferred"], + "mm_per_unit": u["mm_per_unit"], + "declared": u["declared"], + } + if u["mm_per_unit"] is None: + notes.append( + f"drawing {key}: no unit could be settled; its drawing units are taken as mm" + ) + fa = units["a"]["mm_per_unit"] or 1.0 + fb = units["b"]["mm_per_unit"] or 1.0 + + matched = best["matched"] + tags = sorted(matched) + rows: list[dict] = [] + below = 0 + for ti, tj in combinations(tags, 2): + d_b = math.dist(matched[ti][1], matched[tj][1]) * fb + if d_b < floor: + below += 1 + continue + d_a = math.dist(matched[ti][0], matched[tj][0]) * fa + rows.append({"tags": [ti, tj], "a_mm": d_a, "b_mm": d_b, "ratio": d_a / d_b}) + overall = _summary(tags, rows) + + faces = [face for _row, face in rooms_b["labelled"]] + room_of = {id(face): row for row, face in rooms_b["labelled"]} + groups: dict[str, dict] = {} + for tag in tags: + pb = matched[tag][1] + face = face_containing(faces, pb) if faces else None + if face is not None: + row = room_of[id(face)] + key = " ".join(p for p in ("room", row["number"], row["name"]) if p) + kind = "room" + else: + cluster = next( + (o["cluster"] for o in side_b["tags"] if o["tag"] == tag and tuple(o["at"]) == pb), + None, + ) + key, kind = (cluster or "unclustered"), "cluster" + groups.setdefault(key, {"group": key, "kind": kind, "tags": []})["tags"].append(tag) + group_rows = [] + for key in sorted(groups): + g = groups[key] + if len(g["tags"]) < 2: + continue + members = set(g["tags"]) + inside = [r for r in rows if r["tags"][0] in members and r["tags"][1] in members] + group_rows.append({**g, **_summary(g["tags"], inside)}) + + median = overall["median"] + listed = sorted( + rows, + key=lambda r: (-(abs(r["ratio"] - median) if median else 0.0), r["tags"]), + )[:PAIR_ROW_CAP] + return { + **overall, + "pairs": listed, + "pairs_truncated": len(rows) > PAIR_ROW_CAP, + "pairs_below_floor": below, + "matched": [{"tag": t, "a": list(matched[t][0]), "b": list(matched[t][1])} for t in tags], + "ambiguous": best["ambiguous"], + "occurrences": best["occurrences"], + "only_a": best["only_a"], + "only_b": best["only_b"], + "groups": group_rows, + "scope": {"cluster_a": best["cluster_a"], "cluster_b": best["cluster_b"]}, + "units": units, + "min_pair_mm": floor, + "thresholds": { + "within": WITHIN, + "to_scale_share": TO_SCALE_SHARE, + "schematic_share": SCHEMATIC_SHARE, + "min_tags": MIN_TAGS, + }, + "notes": notes, + } diff --git a/engineering/understand/units.py b/engineering/understand/units.py new file mode 100644 index 0000000..d17bf7b --- /dev/null +++ b/engineering/understand/units.py @@ -0,0 +1,399 @@ +"""Declared versus inferred drawing units, and extents that one stray entity cannot stretch. + +`$INSUNITS` is a claim the drawing makes about itself, and a foreign drawing +can make it wrongly: a layout that declares inches over millimetre geometry +reads 25.4 times too large if the header is believed. This module never corrects +the claim silently. It measures what the geometry implies and reports both, +with the numbers that decided it. + +**Evidence.** Each item is one median measured in drawing units and the band a +real drawing puts it in, in metres. A candidate unit is *plausible* for an item +when the measurement, converted by that unit, lands in the band. An item that +every candidate or no candidate satisfies says nothing and is dropped as +inconclusive. + +======================== ====================== ====== ==================================== +item band weight basis +======================== ====================== ====== ==================================== +``room_area`` 1 m² … 5 000 m² 1.0 a labelled room's face (this reader's + heuristic) +``wall_thickness`` 0.05 m … 1.0 m 1.0 nearest parallel wall line (heuristic) +``door_leaf`` 0.5 m … 1.5 m 1.0 radius of a quarter-circle arc, a door + swing (heuristic) +``text_height`` 1.8 mm … 4.0 m 0.5 ISO 3098-0 lettering heights 1.8-20 mm + at ISO 5455 scales 1:1 to 1:200 (the + 1:200 ceiling is this reader's choice + for plant drawings) +``overall_size`` 0.01 m … 2 000 m 0.25 the larger side of the robust extents + (heuristic: a plant or a building) +======================== ====================== ====== ==================================== + +**Decision.** A candidate scores the weights of the items it is plausible for; +the header is evidence too and adds :data:`HEADER_WEIGHT` to the declared unit. +The highest score wins and a tie goes to the declared unit, then to the order of +:data:`PREFERENCE`. So the header is overruled only when the geometry +outweighs it - one weak item (the overall size) never does. + +**Extents.** The body of the drawing is grown from its middle: the seed is +the quartile box of the entity centres (Q1..Q3 on each axis), and every +entity whose box lies within :data:`OUTLIER_REACH` body diagonals of the body +joins it, pass after pass, until a pass adds nothing. What never joins is an +outlier - an entity farther from everything else than the drawing is large. +The robust extents are the body's box. This is this reader's heuristic, not +a statistical standard: a per-axis fence on centres (Tukey's) flags the edge +walls of a plan whose centres crowd its middle, and a quantile range is +dragged by the outlier itself in a small drawing. Fewer than four entities +have no outliers. + +Pure: records in, dicts out. The drawing is never touched. +""" + +from __future__ import annotations + +import math +import statistics +from collections.abc import Iterable, Sequence + +from ezdxf import units as ezdxf_units + +from engineering.understand.snapshot import EntityRecord, Snapshot +from engineering.understand.vocab import classify_layer + +__all__ = [ + "BANDS", + "CANDIDATES", + "HEADER_WEIGHT", + "MM_PER_UNIT", + "PREFERENCE", + "entity_box", + "infer_units", + "robust_extents", + "wall_segments", +] + +#: Millimetres per drawing unit, exact by definition: the SI prefixes, and the +#: inch (25.4 mm) and foot (304.8 mm) of the 1959 international yard and pound +#: agreement. ezdxf's own ``METER_FACTOR`` is rounded (39.37007874 in/m), so it +#: is not used for conversion. +MM_PER_UNIT: dict[str, float] = {"mm": 1.0, "cm": 10.0, "m": 1000.0, "in": 25.4, "ft": 304.8} +CANDIDATES: tuple[str, ...] = ("mm", "cm", "m", "in", "ft") +#: Tie-break after the declared unit. +PREFERENCE: tuple[str, ...] = ("mm", "m", "cm", "in", "ft") +#: What the header's own claim weighs against the geometry. +HEADER_WEIGHT = 0.5 +#: (low, high, weight, power): the band in metres (m² for power 2). +BANDS: dict[str, tuple[float, float, float, int]] = { + "room_area": (1.0, 5000.0, 1.0, 2), + "wall_thickness": (0.05, 1.0, 1.0, 1), + "door_leaf": (0.5, 1.5, 1.0, 1), + "text_height": (0.0018, 4.0, 0.5, 1), + "overall_size": (0.01, 2000.0, 0.25, 1), +} +#: How far (in body diagonals) an entity may lie from the body and still join it. +OUTLIER_REACH = 1.0 +#: A door swing is a quarter circle; the sweep may be drawn a little off. +DOOR_SWEEP = (85.0, 95.0) +#: Wall-thickness pairing reads at most this many wall segments. +WALL_SEGMENT_CAP = 3000 + +_TEXT_TYPES = frozenset({"TEXT", "MTEXT"}) +_POLY_TYPES = frozenset({"LWPOLYLINE", "POLYLINE"}) + + +def entity_box(rec: EntityRecord) -> tuple[float, float, float, float] | None: + """The record's WCS box: its own ``bbox``, else its points (a circle or arc + grown by its radius about its centre), else None.""" + if rec.bbox is not None: + return tuple(float(v) for v in rec.bbox) + if not rec.points: + return None + xs = [float(p[0]) for p in rec.points] + ys = [float(p[1]) for p in rec.points] + grow = float(rec.radius) if rec.type in ("CIRCLE", "ARC") and rec.radius else 0.0 + return (min(xs) - grow, min(ys) - grow, max(xs) + grow, max(ys) + grow) + + +def _union(boxes: Iterable[tuple[float, float, float, float]]): + boxes = list(boxes) + if not boxes: + return None + return ( + min(b[0] for b in boxes), + min(b[1] for b in boxes), + max(b[2] for b in boxes), + max(b[3] for b in boxes), + ) + + +def _diag(box) -> float: + return math.hypot(box[2] - box[0], box[3] - box[1]) if box else 0.0 + + +def _box_gap(a, b) -> float: + """Distance between two boxes; 0 when they touch or overlap.""" + dx = max(a[0] - b[2], b[0] - a[2], 0.0) + dy = max(a[1] - b[3], b[1] - a[3], 0.0) + return math.hypot(dx, dy) + + +def _grow_body(boxes: Sequence[tuple[float, float, float, float]]) -> set[int]: + """Indices of the boxes that never join the body grown from the centre.""" + if len(boxes) < 4: + return set() + xs = [(b[0] + b[2]) / 2.0 for b in boxes] + ys = [(b[1] + b[3]) / 2.0 for b in boxes] + qx = statistics.quantiles(xs, n=4, method="inclusive") + qy = statistics.quantiles(ys, n=4, method="inclusive") + body = (qx[0], qy[0], qx[2], qy[2]) + floor = statistics.median(_diag(b) for b in boxes) + pending = set(range(len(boxes))) + while pending: + reach = OUTLIER_REACH * max(_diag(body), floor) + joined = [i for i in pending if _box_gap(boxes[i], body) <= reach] + if not joined: + break + for i in joined: + body = _union([body, boxes[i]]) + pending.difference_update(joined) + return pending + + +def _distance_to_box(x: float, y: float, box) -> float: + dx = max(box[0] - x, 0.0, x - box[2]) + dy = max(box[1] - y, 0.0, y - box[3]) + return math.hypot(dx, dy) + + +def robust_extents(records: Sequence[EntityRecord]) -> dict: + """Computed and robust extents of ``records`` and the outliers between them. + + ``outliers`` are sorted farthest first, each with its handle, type, layer, + centre and its distance from the robust box. ``stretch`` is the computed + diagonal over the robust one (None when the robust box is a point). + """ + items = [(rec, box) for rec in records if (box := entity_box(rec)) is not None] + centres = [((b[0] + b[2]) / 2.0, (b[1] + b[3]) / 2.0) for _rec, b in items] + skip = _grow_body([b for _rec, b in items]) + outside = sorted(skip) + computed = _union(b for _rec, b in items) + robust = _union(b for i, (_rec, b) in enumerate(items) if i not in skip) + outliers = [] + for i in outside: + rec, _box = items[i] + cx, cy = centres[i] + outliers.append( + { + "handle": rec.handle, + "type": rec.type, + "layer": rec.layer, + "at": [cx, cy], + "distance": _distance_to_box(cx, cy, robust) if robust else None, + "confidence": 0.9, + } + ) + outliers.sort(key=lambda o: (-(o["distance"] or 0.0), o["handle"])) + robust_diag = _diag(robust) + return { + "computed": list(computed) if computed else None, + "robust": list(robust) if robust else None, + "outliers": outliers, + "stretch": (_diag(computed) / robust_diag) if robust_diag > 0.0 else None, + "method": ( + "body grown from the quartile box of entity centres; an entity farther from it " + f"than {OUTLIER_REACH:g} body diagonal(s) never joins and is an outlier" + ), + } + + +def wall_segments( + records: Sequence[EntityRecord], *, is_wall=None +) -> tuple[list[tuple[tuple[float, float], tuple[float, float]]], int]: + """Straight segments of LINE and polyline records on wall layers, plus the + number of bulged (arc) polyline segments left out. + + ``is_wall(layer)`` defaults to `vocab.classify_layer` naming the + ``architecture`` discipline. + """ + cache: dict[str, bool] = {} + + def wall(layer: str) -> bool: + if layer not in cache: + if is_wall is not None: + cache[layer] = bool(is_wall(layer)) + else: + cache[layer] = classify_layer(layer).get("discipline") == "architecture" + return cache[layer] + + segments: list[tuple[tuple[float, float], tuple[float, float]]] = [] + bulged = 0 + for rec in records: + if not wall(rec.layer): + continue + if rec.type == "LINE" and len(rec.points) >= 2: + segments.append((tuple(rec.points[0]), tuple(rec.points[1]))) + elif rec.type in _POLY_TYPES and len(rec.points) >= 2: + pts = list(rec.points) + count = len(pts) if rec.closed else len(pts) - 1 + for i in range(count): + a, b = pts[i], pts[(i + 1) % len(pts)] + bulge = rec.bulges[i] if i < len(rec.bulges) else 0.0 + if abs(bulge) > 1e-12: + bulged += 1 + continue + if a != b: + segments.append((tuple(a), tuple(b))) + return segments, bulged + + +def _wall_thickness(segments) -> tuple[float | None, int]: + """Median distance from an axis-parallel wall segment to its nearest parallel + neighbour on either side that overlaps it; (None, 0) without a pair. + + Both sides, because the inner face of a double-line wall has the room on + one side and its own outer face on the other: looking one way only reads + half the faces as room widths and the median lands on neither.""" + horizontal: list[tuple[float, float, float]] = [] # (y, x0, x1) + vertical: list[tuple[float, float, float]] = [] # (x, y0, y1) + for (ax, ay), (bx, by) in segments[:WALL_SEGMENT_CAP]: + length = math.hypot(bx - ax, by - ay) + if length <= 0.0: + continue + if abs(by - ay) <= 1e-9 * length: + horizontal.append((ay, min(ax, bx), max(ax, bx))) + elif abs(bx - ax) <= 1e-9 * length: + vertical.append((ax, min(ay, by), max(ay, by))) + gaps: list[float] = [] + for group in (horizontal, vertical): + group.sort() + for i, (c, lo, hi) in enumerate(group): + best = None + for step in (1, -1): + j = i + step + while 0 <= j < len(group): + c2, lo2, hi2 = group[j] + d = abs(c2 - c) + if best is not None and d >= best: + break + if d > 0.0 and lo2 < hi and lo < hi2: + best = d + j += step + if best is not None: + gaps.append(best) + if not gaps: + return None, 0 + return statistics.median(gaps), len(gaps) + + +def _declared(code: int) -> dict: + try: + name = ezdxf_units.decode(int(code)) + label = ezdxf_units.unit_name(int(code)) + except (IndexError, ValueError, TypeError): + name, label = None, None + return {"code": int(code), "name": name, "label": label or "Unitless"} + + +def infer_units( + snap: Snapshot, + *, + records: Sequence[EntityRecord] | None = None, + room_areas: Sequence[float] = (), +) -> dict: + """The unit the geometry implies, against the one ``$INSUNITS`` declares. + + ``records`` defaults to the snapshot's model-space records; pass the ones + left after the outliers are removed. ``room_areas`` are labelled room faces + in square drawing units. Every item of ``evidence`` carries its median, + its sample count, what it is in metres under each candidate, the + candidates it is plausible for and whether it was informative. + """ + recs = [r for r in snap.records if r.space == "Model"] if records is None else list(records) + measured: dict[str, tuple[float, int]] = {} + + heights = [float(r.height) for r in recs if r.type in _TEXT_TYPES and r.height and r.height > 0] + if heights: + measured["text_height"] = (statistics.median(heights), len(heights)) + radii = [] + for r in recs: + if r.type == "ARC" and r.radius and r.angles: + sweep = (float(r.angles[1]) - float(r.angles[0])) % 360.0 + if DOOR_SWEEP[0] <= sweep <= DOOR_SWEEP[1]: + radii.append(float(r.radius)) + if radii: + measured["door_leaf"] = (statistics.median(radii), len(radii)) + thickness, pairs = _wall_thickness(wall_segments(recs)[0]) + if thickness is not None: + measured["wall_thickness"] = (thickness, pairs) + areas = [float(a) for a in room_areas if a and a > 0] + if areas: + measured["room_area"] = (statistics.median(areas), len(areas)) + box = robust_extents(recs)["robust"] + if box: + side = max(box[2] - box[0], box[3] - box[1]) + if side > 0: + measured["overall_size"] = (side, len(recs)) + + evidence = [] + for kind, (value, count) in measured.items(): + low, high, weight, power = BANDS[kind] + physical = {u: value * (MM_PER_UNIT[u] / 1000.0) ** power for u in CANDIDATES} + plausible = [u for u in CANDIDATES if low <= physical[u] <= high] + evidence.append( + { + "kind": kind, + "value": value, + "count": count, + "band": [low, high], + "band_unit": "m2" if power == 2 else "m", + "weight": weight, + "as": physical, + "plausible": plausible, + "informative": 0 < len(plausible) < len(CANDIDATES), + } + ) + + declared = _declared(snap.insunits) + name = declared["name"] + informative = [e for e in evidence if e["informative"]] + scores = {u: sum(e["weight"] for e in informative if u in e["plausible"]) for u in CANDIDATES} + if name in scores: + scores[name] += HEADER_WEIGHT + if informative: + best = max(scores.values()) + order = ([name] if name in scores else []) + [u for u in PREFERENCE if u != name] + inferred = next(u for u in order if scores[u] == best) + total = sum(e["weight"] for e in informative) + (HEADER_WEIGHT if name in scores else 0.0) + confidence = round(0.9 * scores[inferred] / total, 2) + elif name in MM_PER_UNIT: + inferred, confidence = name, 0.4 + else: + inferred, confidence = None, 0.0 + + warning = None + if inferred is not None and inferred != name: + support = [e for e in informative if inferred in e["plausible"]] + detail = "; ".join( + f"{e['kind']} {e['value']:g} = {e['as'][inferred]:.4g} {e['band_unit']} as {inferred}" + + (f", {e['as'][name]:.4g} {e['band_unit']} as {name}" if name in MM_PER_UNIT else "") + for e in support + ) + warning = ( + f"INSUNITS declares {declared['label']} ({declared['code']}) but the geometry reads " + f"as {inferred}: {detail}. Lengths are measured in {inferred}; the header is not " + "changed." + ) + elif inferred is None: + warning = ( + f"INSUNITS declares {declared['label']} ({declared['code']}) and no measurement " + "settles the unit; lengths stay in drawing units." + ) + return { + "declared": declared, + "inferred": inferred, + "mm_per_unit": MM_PER_UNIT.get(inferred) if inferred else None, + "agree": inferred is not None and inferred == name, + "confidence": confidence, + "scores": scores, + "evidence": evidence, + "warning": warning, + } diff --git a/server.py b/server.py index 9e3704d..3284d6a 100644 --- a/server.py +++ b/server.py @@ -10554,6 +10554,165 @@ async def arch_plan_from_spec( raise ToolError(str(exc)) from exc +# --------------------------------------------------------------------------- +# ── SECTION 26: Understanding & QA (2 tools) ──────────────────────────────── +# --------------------------------------------------------------------------- + + +@cad_tool( + summary="Read a foreign drawing: units, extents, clusters, layers, tags, rooms, languages.", + cost="read", +) +@mcp.tool( + annotations={"title": "Understand: Describe Drawing", "readOnlyHint": True}, + tags={"analysis", "query"}, +) +async def drawing_understand( + path: Annotated[ + str | None, + Field( + default=None, + description="DXF file to read (a DWG only on a live seat); omitted: the current " + "document.", + ), + ] = None, + ctx: Context = None, +) -> dict: + """One report on a drawing somebody else made, every claim with its evidence + and a confidence. Never modifies the drawing: a file is read as it is, and + the current document through a snapshot (headless: the document in memory; + live: a `Document.Export` DXF copy in a temporary folder, deleted after). + An xref's own entities are not read - the host drawing is. + + - `units`: `$INSUNITS` as declared against the unit the geometry implies + (text heights, wall thickness, door swings, labelled room areas, overall + size - each item with its median and its value under every candidate + unit). A disagreement is a warning with the numbers, never a silent + correction. + - `extents`: declared, computed and robust boxes and the `outliers` that + stretch them, farthest first, by handle. + - `clusters`: separate bodies of model-space content (plan copies, sheets, + details), each with its box, entity count, commonest layers and tallest + labels. + - `layers`: entity count, drawn length (and metres in the inferred unit) + and the discipline / service each most likely carries - the vocabulary's + confidence (0.9) for a name keyword, 0.6 when most texts on the layer + agree, 0.4 when fewer do. + - `equipment_tags` (a tag found more than once is `ambiguous`), `rooms` + (room labels in five languages, each with the wall-layer face it sits in + and its measured area), `languages` (Latin / Turkish / Cyrillic character + classes), `blocks` (named / anonymous, kinds by name), `layouts`, + `title_block` and `warnings`. + + Refused before anything is read: a `path` that does not exist, a DXF over + `MAX_DXF_BYTES` (the refusal names the size and the variable), and a file + the engine cannot parse (a DWG headlessly: ezdxf reads DXF only). + """ + import asyncio + + from ezdxf.lldxf.const import DXFError + + from engineering.understand.describe import describe + from engineering.understand.snapshot import take_snapshot + + target = None + if path: + target = validate_path(path) + if not target.is_file(): + raise ToolError(f"path: no such file {str(target)!r}") + await ctx.info(f"drawing_understand: {target or 'current document'}") + try: + snap = await take_snapshot(_backend(ctx), str(target) if target else None) + except (ValueError, OSError, RuntimeError, DXFError) as exc: + raise ToolError(str(exc)) from exc + return await asyncio.to_thread(describe, snap) + + +@cad_tool( + summary="Is this drawing to scale against a reference? Tag distances, ratios, a verdict.", + cost="read", +) +@mcp.tool( + annotations={"title": "Understand: Scale Check", "readOnlyHint": True}, + tags={"analysis", "query"}, +) +async def drawing_scale_check( + reference: Annotated[ + str, + Field( + description="The reference drawing of the same plant (a DXF, usually the layout): " + "distances on it are the ones the other drawing is measured against." + ), + ], + path: Annotated[ + str | None, + Field( + default=None, + description="The drawing to check (usually the P&ID); omitted: the current document.", + ), + ] = None, + min_pair_mm: Annotated[ + float, + Field( + default=2000.0, + gt=0, + description="Tag pairs closer than this on the reference (mm) are left out: a tag " + "label is not an equipment centre.", + ), + ] = 2000.0, + ctx: Context = None, +) -> dict: + """Is the drawing to scale against the reference - or a schematic whose lengths + are lengths of drawn line, not of pipe? Read-only on both drawings. + + Equipment tags are matched between the two (a Cyrillic look-alike such as + `Т4100` is the Latin `T4100`). Only a tag found exactly once in each is used; + a tag found more than once (plan copies, a legend, a second sheet) is listed + in `ambiguous` and takes no part. When a drawing holds several bodies of + content (two copies of one plan), the pair of scopes - whole drawing or one + cluster on each side - that matches the most tags is used and named in + `scope`. For every two matched tags at least `min_pair_mm` apart on the + reference, the ratio of their distances is taken (both in millimetres by + each drawing's inferred unit, see `drawing_understand`). The result gives + the matched tags, the median ratio, the quartiles and their range, the + share of pairs within +/-10 % of the median, the same per room of the + reference (or per cluster for tags in no room) in `groups`, and a verdict: + `to_scale` (at least 80 % within the band; `factor` is the median, so a + uniform other scale is recognised), `schematic` (50 % or less), `partly` + in between, `insufficient` with fewer than three matched tags or no pair + above the floor. Tag labels are not equipment centres; the band and the + floor absorb that offset, and the result's `notes` say so. + + Refused before anything is read: a `reference` or `path` that does not + exist, a DXF over `MAX_DXF_BYTES` (the refusal names the size and the + variable), a file the engine cannot parse (a DWG headlessly), and a + `min_pair_mm` that is not a positive finite number. + """ + import asyncio + + from ezdxf.lldxf.const import DXFError + + from engineering.understand.scale import scale_check + from engineering.understand.snapshot import take_snapshot + + ref = validate_path(reference) + if not ref.is_file(): + raise ToolError(f"reference: no such file {str(ref)!r}") + target = None + if path: + target = validate_path(path) + if not target.is_file(): + raise ToolError(f"path: no such file {str(target)!r}") + await ctx.info(f"drawing_scale_check: {target or 'current document'} against {ref}") + backend = _backend(ctx) + try: + checked = await take_snapshot(backend, str(target) if target else None) + against = await take_snapshot(backend, str(ref)) + return await asyncio.to_thread(scale_check, checked, against, min_pair_mm=min_pair_mm) + except (ValueError, OSError, RuntimeError, DXFError) as exc: + raise ToolError(str(exc)) from exc + + # --------------------------------------------------------------------------- # ── RESOURCES ─────────────────────────────────────────────────────────────── # --------------------------------------------------------------------------- diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py index e8a1ecb..df20db3 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 (241 tools, 26 groups). + """Frozen snapshot of the surface (243 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 @@ -591,10 +591,18 @@ async def test_tool_group_sizes_are_unchanged(): exactly `PACK_TOOL_NAMES["arch"]`, and every group the track-F branches had grown while the tag was unranked on them returned to its pre-track-F value. + + Track H group U opened SECTION 26 (Understanding & QA) with + `drawing_understand`, tagged `analysis` and `query`: `analysis` outranks + `query` in `_GROUP_TAG_PRIORITY`, so the reader files under `analysis` + (12 -> 13) beside the other whole-drawing queries, and + `drawing_scale_check`, tagged the same, followed it (13 -> 14). This + snapshot is branch-local: the track H merge task recomputes it once the + understand, network and diff branches have landed. """ sizes = {label: len(names) for label, names in (await server._tool_groups()).items()} assert sizes == { - "analysis": 12, + "analysis": 14, "architecture": 12, "batch": 3, "blocks": 9, @@ -621,4 +629,4 @@ async def test_tool_group_sizes_are_unchanged(): "validation": 1, "view": 4, } - assert sum(sizes.values()) == 241 + assert sum(sizes.values()) == 243 diff --git a/tests/test_understand_clusters.py b/tests/test_understand_clusters.py new file mode 100644 index 0000000..3e12077 --- /dev/null +++ b/tests/test_understand_clusters.py @@ -0,0 +1,86 @@ +"""Bodies of model-space content: two plan copies are two clusters.""" + +from __future__ import annotations + +import pytest + +from engineering.understand.clusters import find_clusters +from engineering.understand.snapshot import EntityRecord + +EPS = 1e-9 + + +def line(handle, a, b, layer="WALL"): + return EntityRecord(handle=handle, type="LINE", layer=layer, space="Model", points=(a, b)) + + +def text(handle, value, x, y, height=250.0): + return EntityRecord( + handle=handle, + type="TEXT", + layer="TEXT", + space="Model", + points=((x, y),), + text=value, + height=height, + ) + + +def plan(prefix, dx): + """A 20 000 x 12 000 outline with a partition, a tag in each half and a + title under it, shifted ``dx`` along x. The tags float 4 000 from every + line - far more than one cell - so only the absorb step keeps them in.""" + return [ + line(f"{prefix}1", (dx, 0.0), (dx + 20000.0, 0.0)), + line(f"{prefix}2", (dx + 20000.0, 0.0), (dx + 20000.0, 12000.0)), + line(f"{prefix}3", (dx + 20000.0, 12000.0), (dx, 12000.0)), + line(f"{prefix}4", (dx, 12000.0), (dx, 0.0)), + line(f"{prefix}5", (dx + 10000.0, 0.0), (dx + 10000.0, 12000.0)), + text(f"{prefix}T1", "T4100", dx + 4000.0, 6000.0), + text(f"{prefix}T2", "T4200", dx + 14000.0, 6000.0), + text(f"{prefix}T3", "PLANT LAYOUT", dx + 2000.0, 12500.0, height=500.0), + ] + + +def test_two_plan_copies_side_by_side_are_two_clusters(): + # Copy A spans x 0..20 000, copy B 40 000..60 000: a 20 000 gap. The + # content box is 60 000 x 12 500 (the title's point box ends at its + # insertion, y 12 500), so a cell is 2 % (CELL_FRACTION) of the + # hypot(60 000, 12 500) = 61 288 diagonal, about 1 226 units, and the gap + # is over 16 empty cells. + clusters, membership = find_clusters([*plan("A", 0.0), *plan("B", 40000.0)]) + assert len(clusters) == 2 + left, right = sorted(clusters, key=lambda c: c["box"][0]) + assert left["entities"] == right["entities"] == 8 + assert left["box"] == pytest.approx([0.0, 0.0, 20000.0, 12500.0], abs=EPS) + assert right["box"] == pytest.approx([40000.0, 0.0, 60000.0, 12500.0], abs=EPS) + assert membership["AT1"] == membership["A1"] == left["id"] + assert membership["BT2"] == membership["B5"] == right["id"] + # The tallest text leads the labels. + assert left["labels"][0] == "PLANT LAYOUT" + assert left["layers"][0] == "WALL" + + +def test_ids_run_largest_first_then_left_to_right(): + small = [line("S1", (100000.0, 0.0), (100500.0, 0.0))] + clusters, _ = find_clusters([*small, *plan("A", 40000.0), *plan("B", 0.0)]) + assert [c["id"] for c in clusters] == ["C1", "C2", "C3"] + assert [c["box"][0] for c in clusters] == pytest.approx([0.0, 40000.0, 100000.0], abs=EPS) + assert [c["entities"] for c in clusters] == [8, 8, 1] + + +def test_excluded_handles_take_no_part(): + far = line("FAR", (-5_000_000_000.0, 0.0), (-4_999_999_000.0, 0.0)) + clusters, membership = find_clusters([*plan("A", 0.0), far], exclude={"FAR"}) + assert len(clusters) == 1 + assert "FAR" not in membership + assert clusters[0]["box"] == pytest.approx([0.0, 0.0, 20000.0, 12500.0], abs=EPS) + + +def test_nothing_to_cluster_is_an_empty_answer(): + assert find_clusters([]) == ([], {}) + + +def test_a_non_positive_cell_is_refused(): + with pytest.raises(ValueError, match="cell: must be greater than zero"): + find_clusters(plan("A", 0.0), cell=-1.0) diff --git a/tests/test_understand_describe.py b/tests/test_understand_describe.py new file mode 100644 index 0000000..82e5542 --- /dev/null +++ b/tests/test_understand_describe.py @@ -0,0 +1,301 @@ +"""drawing_understand's report: the synthetic plant pair, then hand-built snapshots. + +The plant-pair facts come from the generator's truth dict (Task 2), never from +a real drawing; the hand-built cases compute their expectations in comments. +""" + +from __future__ import annotations + +import copy + +import pytest + +from engineering.understand.describe import describe, tag_occurrences, text_language +from engineering.understand.labels import parse_tag +from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot +from engineering.understand.vocab import classify_layer, room_label +from tests.fixtures.plant_pair import build_plant_pair + +EPS = 1e-9 + + +def shoelace(polygon): + total = 0.0 + for (x1, y1), (x2, y2) in zip(polygon, polygon[1:] + polygon[:1], strict=True): + total += x1 * y2 - x2 * y1 + return abs(total) / 2.0 + + +@pytest.fixture(scope="module") +def plant(tmp_path_factory): + truth = build_plant_pair(tmp_path_factory.mktemp("plant")) + return { + "truth": truth, + "pid": describe(read_snapshot(truth["pid"])), + "layout": describe(read_snapshot(truth["layout"])), + } + + +# ── the synthetic plant pair ──────────────────────────────────────────────── + + +def test_the_layout_declares_inches_over_millimetre_geometry(plant): + units = plant["layout"]["units"] + assert units["declared"]["code"] == plant["truth"]["layout_insunits"] + assert units["inferred"] == plant["truth"]["layout_true_unit"] + assert units["agree"] is False + assert any(w.startswith("INSUNITS declares Inches (1)") for w in plant["layout"]["warnings"]) + + +def test_the_far_outlier_is_named_by_handle(plant): + outliers = plant["layout"]["extents"]["outliers"] + assert outliers[0]["handle"] == plant["truth"]["outlier_handle"] + assert any(plant["truth"]["outlier_handle"] in w for w in plant["layout"]["warnings"]) + + +def test_the_two_plan_copies_are_two_clusters(plant): + assert len(plant["layout"]["clusters"]) == plant["truth"]["cluster_count"] + + +def test_service_layers_are_classified(plant): + rows = {r["name"]: r for r in plant["layout"]["layers"]} + rows.update({r["name"]: r for r in plant["pid"]["layers"]}) + for layer, service in plant["truth"]["layer_services"].items(): + assert rows[layer]["service"] == service, layer + assert rows[layer]["confidence"] > 0.0, layer + + +def test_every_equipment_tag_of_the_truth_is_found_on_the_pid(plant): + found = {t["tag"] for t in plant["pid"]["equipment_tags"]} + expected = {tag for run in plant["truth"]["pipe_runs"] for tag in run["tags"]} + expected |= {row["tag"] for row in plant["truth"]["cable_rows"]} + assert expected <= found + + +def test_layout_rooms_carry_the_truth_numbers_names_and_measured_faces(plant): + for room in plant["truth"]["rooms"]: + rows = [ + r + for r in plant["layout"]["rooms"] + if r["number"] == room["number"] and r["name"] == room["name"] + ] + assert rows, room["number"] + expected = shoelace([tuple(p) for p in room["polygon"]]) + for row in rows: + assert row["face"] is not None, room["number"] + assert abs(row["face"]["area"] - expected) <= 1e-9 * expected, room["number"] + assert row["confidence"] == 0.9 + + +def test_the_pid_texts_are_read_as_cyrillic(plant): + assert plant["pid"]["languages"]["by_class"]["cyrillic"] > 0 + + +# ── hand-built snapshots ──────────────────────────────────────────────────── + + +def line(handle, a, b, layer="WALL", space="Model"): + return EntityRecord(handle=handle, type="LINE", layer=layer, space=space, points=(a, b)) + + +def text(handle, value, x, y, layer="TEXT", height=250.0): + return EntityRecord( + handle=handle, + type="TEXT", + layer=layer, + space="Model", + points=((x, y),), + text=value, + height=height, + ) + + +def insert(handle, block, x, y, attribs=(), space="Model"): + return EntityRecord( + handle=handle, + type="INSERT", + layer="0", + space=space, + points=((x, y),), + block=block, + attribs=tuple(attribs), + ) + + +def rect(prefix, x0, y0, x1, y1, layer="WALL"): + return [ + line(f"{prefix}1", (x0, y0), (x1, y0), layer), + line(f"{prefix}2", (x1, y0), (x1, y1), layer), + line(f"{prefix}3", (x1, y1), (x0, y1), layer), + line(f"{prefix}4", (x0, y1), (x0, y0), layer), + ] + + +def snapshot(records, insunits=4, layers=None, layouts=("Model",)): + return Snapshot( + source="test", + insunits=insunits, + extmin=None, + extmax=None, + layers=layers or {}, + layouts=layouts, + records=tuple(records), + ) + + +def test_a_room_label_inside_the_wall_faces_is_measured(): + assert classify_layer("WALL")["discipline"] == "architecture" + parsed = room_label("ROOM 101 PROCESS") + assert parsed is not None + # Inner face (200,200)-(6200,4200): 6000 x 4000 = 24 000 000 unit^2. + records = [ + *rect("O", 0.0, 0.0, 6400.0, 4400.0), + *rect("I", 200.0, 200.0, 6200.0, 4200.0), + text("R", "ROOM 101 PROCESS", 1000.0, 2000.0), + ] + report = describe(snapshot(records)) + (room,) = report["rooms"] + assert room["number"] == parsed["number"] + assert room["name"] == parsed["name"] + assert room["face"]["area"] == pytest.approx(24_000_000.0, abs=EPS) + assert room["face"]["polygon"] == [ + [200.0, 200.0], + [6200.0, 200.0], + [6200.0, 4200.0], + [200.0, 4200.0], + ] + assert room["source"] == "label+face" + assert room["confidence"] == 0.9 + # The wall ring between the outlines is the one unlabelled face. + assert report["room_faces"]["faces"] == 2 + assert report["room_faces"]["unlabelled"] == 1 + + +def test_an_attribute_and_a_text_of_one_tag_are_one_occurrence_and_a_copy_is_two(): + assert parse_tag("T4100") == "T4100" + records = [ + line("L", (0.0, 0.0), (20000.0, 0.0), "0"), + insert("B1", "TANK", 1000.0, 0.0, attribs=[("TAG", "T4100")]), + # 150 units away: inside 1 % of the 20 000 diagonal (200), so the same one. + text("X1", "T4100", 1150.0, 0.0), + # 18 000 away: a second occurrence. + text("X2", "T4100", 19000.0, 0.0), + ] + snap = snapshot(records) + occurrences = tag_occurrences(snap) + assert [(o["handle"], o["source"]) for o in occurrences] == [ + ("B1", "attribute"), + ("X2", "text"), + ] + report = describe(snap) + (row,) = report["equipment_tags"] + assert row["tag"] == "T4100" + assert row["count"] == 2 + assert row["ambiguous"] is True + assert row["confidence"] == 0.95 + assert any("appear more than once" in w for w in report["warnings"]) + + +def test_text_languages_are_counted_by_character_class(): + assert text_language("ПОДАЧА CIP") == "cyrillic" + assert text_language("DÖNÜŞ") == "turkish" + assert text_language("DÖNÜS") == "latin" # ö and ü alone are German too + assert text_language("SUPPLY") == "latin" + assert text_language("3,5") == "none" + records = [ + text("A", "ПОДАЧА", 0.0, 0.0), + text("B", "BESLEME DÖNÜŞ", 0.0, 500.0), + text("C", "SUPPLY", 0.0, 1000.0), + text("D", "12", 0.0, 1500.0), + ] + languages = describe(snapshot(records))["languages"] + assert languages["by_class"] == {"latin": 1, "turkish": 1, "cyrillic": 1, "none": 1} + assert languages["share"]["cyrillic"] == 0.25 + assert languages["samples"]["cyrillic"] == ["A"] + + +def test_blocks_are_split_named_and_anonymous_with_kinds_by_name(): + records = [ + insert("A1", "*U12", 0.0, 0.0), + insert("A2", "*U12", 100.0, 0.0), + insert("G1", "GATE_VALVE", 200.0, 0.0), + insert("G2", "GATE_VALVE", 300.0, 0.0), + insert("G3", "GATE_VALVE", 400.0, 0.0), + insert("P1", "PUMP_A", 500.0, 0.0), + ] + blocks = describe(snapshot(records))["blocks"] + assert blocks["inserts"] == 6 + assert [(r["name"], r["inserts"], r["kind"]) for r in blocks["named"]] == [ + ("GATE_VALVE", 3, "valve"), + ("PUMP_A", 1, "equipment"), + ] + assert blocks["anonymous"] == {"names": 1, "inserts": 2} + assert blocks["kinds"] == {"equipment": 1, "valve": 1} + + +def test_the_title_block_is_the_named_paper_space_block_with_its_attributes(): + title = insert( + "TB", + "A1_TITLE", + 0.0, + 0.0, + attribs=[("DWG_NO", "001"), ("REV", "B"), ("SCALE", "1:100"), ("DATE", "2026-09-24")], + space="Layout1", + ) + other = insert( + "M", + "NOTE_BOX", + 0.0, + 0.0, + attribs=[("A", "1"), ("B", "2"), ("C", "3"), ("D", "4"), ("E", "5")], + ) + report = describe(snapshot([title, other], layouts=("Model", "Layout1"))) + block = report["title_block"] + assert block["found"] is True + assert block["handle"] == "TB" + assert block["attributes"] == { + "DWG_NO": "001", + "REV": "B", + "SCALE": "1:100", + "DATE": "2026-09-24", + } + assert block["confidence"] == 0.9 + assert report["layouts"] == [ + {"name": "Model", "entities": 1}, + {"name": "Layout1", "entities": 1}, + ] + + +def test_no_title_block_is_said_not_invented(): + block = describe(snapshot([insert("M", "X", 0.0, 0.0, attribs=[("A", "1")])]))["title_block"] + assert block == {"found": False, "reason": "no block reference with three or more attributes"} + + +def test_a_layer_without_a_keyword_is_classified_by_its_texts(): + expected = classify_layer("CIP SUPPLY") + assert expected["service"] != "unknown" + assert classify_layer("L-17")["service"] == "unknown" + assert classify_layer("L-17")["discipline"] == "unknown" + records = [ + line("P1", (0.0, 0.0), (3000.0, 0.0), "L-17"), + line("P2", (3000.0, 0.0), (3000.0, 4000.0), "L-17"), + text("T1", "CIP SUPPLY", 1000.0, 100.0, layer="L-17"), + text("T2", "CIP SUPPLY", 3100.0, 2000.0, layer="L-17"), + ] + layers = {r["name"]: r for r in describe(snapshot(records))["layers"]} + row = layers["L-17"] + assert row["service"] == expected["service"] + assert row["discipline"] == expected["discipline"] + assert row["confidence"] == 0.6 + assert row["entities"] == 4 + assert row["length"] == pytest.approx(7000.0, abs=EPS) # 3000 + 4000 + assert row["length_m"] == pytest.approx(7.0, abs=EPS) + assert row["evidence"].startswith("2 of 2 texts on the layer read as") + + +def test_describe_leaves_the_snapshot_as_it_was(): + records = [*rect("O", 0.0, 0.0, 6400.0, 4400.0), text("R", "T4100", 100.0, 100.0)] + snap = snapshot(records) + before = copy.deepcopy(snap) + describe(snap) + assert snap == before diff --git a/tests/test_understand_scale.py b/tests/test_understand_scale.py new file mode 100644 index 0000000..52ca2f6 --- /dev/null +++ b/tests/test_understand_scale.py @@ -0,0 +1,192 @@ +"""drawing_scale_check: tag matching and ratio statistics. + +The hand-built plans put six tags at coordinates written below; every pair +distance and ratio in the comments is worked out from them. +""" + +from __future__ import annotations + +import math +from itertools import combinations + +import pytest + +from engineering.understand.scale import match_tags, scale_check +from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot +from tests.fixtures.plant_pair import build_plant_pair + +EPS = 1e-12 + +#: Six tags of one plant, in millimetres. The closest pair is T4100-T4500, +#: hypot(6000, 4000) = 7211.1, so every pair clears the 2 m floor. +PLAN = { + "T4100": (0.0, 0.0), + "T4200": (12000.0, 0.0), + "T4300": (12000.0, 9000.0), + "T4400": (0.0, 9000.0), + "T4500": (6000.0, 4000.0), + "T4600": (20000.0, 15000.0), +} + + +def text(handle, value, x, y): + return EntityRecord( + handle=handle, type="TEXT", layer="TAGS", space="Model", points=((x, y),), text=value + ) + + +def outline(handle, x0, y0, x1, y1): + return EntityRecord( + handle=handle, + type="LWPOLYLINE", + layer="0", + space="Model", + points=((x0, y0), (x1, y0), (x1, y1), (x0, y1)), + bulges=(0.0, 0.0, 0.0, 0.0), + closed=True, + ) + + +def snapshot(records): + return Snapshot( + source="test", + insunits=4, + extmin=None, + extmax=None, + layers={}, + layouts=("Model",), + records=tuple(records), + ) + + +def tags(positions, prefix="H", factor=1.0, dx=0.0): + return [ + text(f"{prefix}{i}", tag, x * factor + dx, y * factor) + for i, (tag, (x, y)) in enumerate(sorted(positions.items())) + ] + + +def test_a_drawing_at_exactly_half_scale_is_to_scale_with_factor_one_half(): + result = scale_check(snapshot(tags(PLAN, "A", 0.5)), snapshot(tags(PLAN, "B"))) + assert result["verdict"] == "to_scale" + assert result["factor"] == pytest.approx(0.5, abs=EPS) + assert result["median"] == pytest.approx(0.5, abs=EPS) + assert result["iqr"] == pytest.approx(0.0, abs=EPS) + assert result["within_10pct"] == 1.0 + assert result["pair_count"] == 15 # C(6, 2) + assert result["pairs_below_floor"] == 0 + assert [m["tag"] for m in result["matched"]] == sorted(PLAN) + assert result["ambiguous"] == [] + assert result["units"]["a"]["unit"] == result["units"]["b"]["unit"] == "mm" + + +def test_a_rearranged_drawing_is_schematic(): + # The same six tags, shuffled onto each other's places and one moved off. + rearranged = { + "T4100": (12000.0, 9000.0), + "T4200": (0.0, 0.0), + "T4300": (20000.0, 15000.0), + "T4400": (6000.0, 4000.0), + "T4500": (12000.0, 0.0), + "T4600": (0.0, 20000.0), + } + result = scale_check(snapshot(tags(rearranged, "A")), snapshot(tags(PLAN, "B"))) + ratios = sorted( + math.dist(rearranged[i], rearranged[j]) / math.dist(PLAN[i], PLAN[j]) + for i, j in combinations(sorted(PLAN), 2) + ) + median = ratios[7] # 15 ratios: the 8th + within = sum(1 for r in ratios if abs(r - median) <= 0.1 * median) / 15 + assert result["median"] == pytest.approx(median, abs=EPS) + assert result["within_10pct"] == pytest.approx(within, abs=EPS) + assert within <= 0.5 + assert result["verdict"] == "schematic" + assert result["factor"] is None + + +def test_a_tag_found_twice_is_ambiguous_and_takes_no_part(): + # T4400 appears twice on the reference: neither copy is matched. + reference = [*tags(PLAN, "B"), text("DUP", "T4400", 3000.0, 12000.0)] + matched = match_tags(snapshot(tags(PLAN, "A", 0.5)), snapshot(reference)) + assert "T4400" not in matched["matched"] + assert matched["ambiguous"] == ["T4400"] + assert matched["occurrences"] == {"T4400": {"a": 1, "b": 2}} + assert matched["matched"]["T4200"] == ((6000.0, 0.0), (12000.0, 0.0)) + + result = scale_check(snapshot(tags(PLAN, "A", 0.5)), snapshot(reference)) + assert result["ambiguous"] == ["T4400"] + assert "T4400" not in [m["tag"] for m in result["matched"]] + assert result["pair_count"] == 10 # C(5, 2): the ambiguous tag is in no pair + assert all("T4400" not in row["tags"] for row in result["pairs"]) + assert result["verdict"] == "to_scale" + assert result["factor"] == pytest.approx(0.5, abs=EPS) + + +def test_two_plan_copies_on_the_reference_resolve_to_one_copy(): + # The reference holds the plan twice, 60 000 apart, each inside its own + # outline: over the whole drawing every tag is ambiguous, inside one copy + # every tag is unique. + reference = [ + outline("O1", -1000.0, -1000.0, 21000.0, 16000.0), + *tags(PLAN, "L"), + outline("O2", 59000.0, -1000.0, 81000.0, 16000.0), + *tags(PLAN, "R", dx=60000.0), + ] + whole = match_tags(snapshot(tags(PLAN, "A", 0.5)), snapshot(reference)) + assert whole["matched"] == {} + assert whole["ambiguous"] == sorted(PLAN) + result = scale_check(snapshot(tags(PLAN, "A", 0.5)), snapshot(reference)) + assert result["scope"] == {"cluster_a": None, "cluster_b": "C1"} + assert len(result["matched"]) == 6 + assert result["ambiguous"] == [] + assert result["verdict"] == "to_scale" + assert result["factor"] == pytest.approx(0.5, abs=EPS) + + +def test_cyrillic_look_alike_tags_match_their_latin_twins(): + # "Т4100" typed with a Cyrillic Т (U+0422) is the layout's Latin T4100. + cyrillic = {("Т" + tag[1:] if tag == "T4100" else tag): xy for tag, xy in PLAN.items()} + matched = match_tags(snapshot(tags(cyrillic, "A")), snapshot(tags(PLAN, "B"))) + assert "T4100" in matched["matched"] + assert len(matched["matched"]) == 6 + + +def test_fewer_than_three_matched_tags_is_insufficient(): + two = {k: PLAN[k] for k in ("T4100", "T4200")} + result = scale_check(snapshot(tags(two, "A")), snapshot(tags(PLAN, "B"))) + assert result["verdict"] == "insufficient" + assert result["factor"] is None + assert result["pair_count"] == 1 + + +def test_pairs_closer_than_the_floor_on_the_reference_are_left_out(): + # With a 13 m floor, the pairs of PLAN at least 13 000 apart on the + # reference are T4100-T4300 (15 000), T4100-T4600 (25 000), T4200-T4600 + # (hypot(8000, 15000) = 17 000), T4400-T4600 (hypot(20000, 6000) = 20 881), + # T4500-T4600 (hypot(14000, 11000) = 17 804), T4200-T4400 (15 000): six. + result = scale_check( + snapshot(tags(PLAN, "A", 0.5)), snapshot(tags(PLAN, "B")), min_pair_mm=13000.0 + ) + assert result["pair_count"] == 6 + assert result["pairs_below_floor"] == 9 + assert result["verdict"] == "to_scale" + + +@pytest.mark.parametrize("floor", [0.0, -5.0, float("nan"), float("inf")]) +def test_a_floor_that_is_not_a_positive_number_is_refused(floor): + with pytest.raises(ValueError, match="min_pair_mm: must be a positive number"): + scale_check(snapshot(tags(PLAN, "A")), snapshot(tags(PLAN, "B")), min_pair_mm=floor) + + +def test_an_unknown_cluster_is_refused_with_the_ids_the_drawing_has(): + with pytest.raises(ValueError, match=r"cluster_b: no cluster 'C99'"): + match_tags(snapshot(tags(PLAN, "A")), snapshot(tags(PLAN, "B")), cluster_b="C99") + + +def test_the_synthetic_pid_is_schematic_against_its_layout(tmp_path): + truth = build_plant_pair(tmp_path) + result = scale_check(read_snapshot(truth["pid"]), read_snapshot(truth["layout"])) + assert result["verdict"] == truth["scale_verdict"] + assert len(result["matched"]) >= 3 + # The layout's two plan copies: the check must pick one of them. + assert result["scope"]["cluster_b"] is not None diff --git a/tests/test_understand_scale_tools.py b/tests/test_understand_scale_tools.py new file mode 100644 index 0000000..5a2b631 --- /dev/null +++ b/tests/test_understand_scale_tools.py @@ -0,0 +1,100 @@ +"""drawing_scale_check over the MCP wire, on a fresh headless document.""" + +from __future__ import annotations + +import ezdxf +import pytest +import pytest_asyncio +from fastmcp import Client + +import server +from tests.fixtures.plant_pair import build_plant_pair + +pytestmark = pytest.mark.asyncio + +#: Six tags of one plant in millimetres; the closest pair is 7 211 apart. +PLAN = { + "T4100": (0.0, 0.0), + "T4200": (12000.0, 0.0), + "T4300": (12000.0, 9000.0), + "T4400": (0.0, 9000.0), + "T4500": (6000.0, 4000.0), + "T4600": (20000.0, 15000.0), +} + + +@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 + + +def reference_file(tmp_path): + """The plan's tags at full size in a millimetre DXF (INSUNITS 4).""" + doc = ezdxf.new(dxfversion="R2010", units=4) + msp = doc.modelspace() + for tag, (x, y) in PLAN.items(): + msp.add_text(tag, height=250.0, dxfattribs={"insert": (x, y), "layer": "TAGS"}) + path = tmp_path / "reference.dxf" + doc.saveas(path) + return str(path) + + +async def test_the_synthetic_pid_is_schematic_against_its_layout(client, tmp_path): + truth = build_plant_pair(tmp_path) + result = await client.call_tool( + "drawing_scale_check", {"path": truth["pid"], "reference": truth["layout"]} + ) + assert result.structured_content["verdict"] == truth["scale_verdict"] + + +async def test_the_current_document_at_half_scale_is_to_scale_with_factor_one_half( + client, tmp_path +): + # The same tags at half the coordinates in the current document. Its + # header says metres (drawing_new keeps ezdxf's default INSUNITS 6), but + # 250-unit text and a 10 000-unit spread read as millimetres, and the + # check converts by the inferred unit: the factor stays 0.5. + for tag, (x, y) in PLAN.items(): + await client.call_tool( + "entity_create_text", + {"text": tag, "x": x * 0.5, "y": y * 0.5, "height": 250.0, "layer": "TAGS"}, + ) + before = (await client.call_tool("analysis_entity_stats", {})).structured_content + result = await client.call_tool("drawing_scale_check", {"reference": reference_file(tmp_path)}) + after = (await client.call_tool("analysis_entity_stats", {})).structured_content + assert after == before + report = result.structured_content + assert report["units"]["a"]["unit"] == report["units"]["b"]["unit"] == "mm" + assert report["verdict"] == "to_scale" + assert report["factor"] == pytest.approx(0.5, abs=1e-12) + assert report["pair_count"] == 15 + assert report["ambiguous"] == [] + + +async def test_a_missing_reference_is_refused_by_name(client, tmp_path): + result = await client.call_tool( + "drawing_scale_check", {"reference": str(tmp_path / "none.dxf")}, raise_on_error=False + ) + assert result.is_error is True + assert "reference: no such file" in result.content[0].text + + +async def test_a_non_positive_floor_is_refused(client, tmp_path): + result = await client.call_tool( + "drawing_scale_check", + {"reference": reference_file(tmp_path), "min_pair_mm": 0.0}, + raise_on_error=False, + ) + assert result.is_error is True + assert "min_pair_mm" in result.content[0].text + + +async def test_drawing_scale_check_is_advertised_read_only(client): + tools = {tool.name: tool for tool in await client.list_tools()} + assert tools["drawing_scale_check"].annotations.readOnlyHint is True diff --git a/tests/test_understand_tools.py b/tests/test_understand_tools.py new file mode 100644 index 0000000..8722ab8 --- /dev/null +++ b/tests/test_understand_tools.py @@ -0,0 +1,82 @@ +"""drawing_understand 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 tests.fixtures.plant_pair import build_plant_pair + +pytestmark = pytest.mark.asyncio + + +@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 test_drawing_understand_reads_the_synthetic_layout_file(client, tmp_path): + truth = build_plant_pair(tmp_path) + result = await client.call_tool("drawing_understand", {"path": truth["layout"]}) + report = result.structured_content + assert report["units"]["declared"]["code"] == truth["layout_insunits"] + assert report["units"]["inferred"] == truth["layout_true_unit"] + assert report["extents"]["outliers"][0]["handle"] == truth["outlier_handle"] + assert len(report["clusters"]) == truth["cluster_count"] + + +async def test_drawing_understand_reads_the_current_document_and_leaves_it_alone(client): + # A 6400 x 4400 outline on WALL (perimeter 2 x (6400 + 4400) = 21 600) and + # one tag. + corners = [(0.0, 0.0), (6400.0, 0.0), (6400.0, 4400.0), (0.0, 4400.0)] + for (x1, y1), (x2, y2) in zip(corners, corners[1:] + corners[:1], strict=True): + await client.call_tool( + "entity_create_line", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "layer": "WALL"} + ) + await client.call_tool( + "entity_create_text", + {"text": "T4100", "x": 1000.0, "y": 1000.0, "height": 250.0, "layer": "TAGS"}, + ) + before = (await client.call_tool("analysis_entity_stats", {})).structured_content + report = (await client.call_tool("drawing_understand", {})).structured_content + after = (await client.call_tool("analysis_entity_stats", {})).structured_content + assert after == before + layers = {row["name"]: row for row in report["layers"]} + assert layers["WALL"]["entities"] == 4 + assert layers["WALL"]["length"] == pytest.approx(21600.0, abs=1e-9) + assert [t["tag"] for t in report["equipment_tags"]] == ["T4100"] + + +async def test_a_missing_file_is_refused_by_path(client, tmp_path): + missing = tmp_path / "nowhere.dxf" + result = await client.call_tool( + "drawing_understand", {"path": str(missing)}, raise_on_error=False + ) + assert result.is_error is True + assert "path: no such file" in result.content[0].text + assert "nowhere.dxf" in result.content[0].text + + +async def test_a_file_the_engine_cannot_parse_is_refused(client, tmp_path): + # Not a DXF: the snapshot's parse error comes back as a tool error that + # names the file, never as a traceback. + garbage = tmp_path / "not_a_drawing.dxf" + garbage.write_text("this is not a drawing", encoding="utf-8") + result = await client.call_tool( + "drawing_understand", {"path": str(garbage)}, raise_on_error=False + ) + assert result.is_error is True + assert "not_a_drawing.dxf" in result.content[0].text + + +async def test_drawing_understand_is_advertised_read_only(client): + tools = {tool.name: tool for tool in await client.list_tools()} + assert tools["drawing_understand"].annotations.readOnlyHint is True diff --git a/tests/test_understand_units.py b/tests/test_understand_units.py new file mode 100644 index 0000000..4434ea4 --- /dev/null +++ b/tests/test_understand_units.py @@ -0,0 +1,161 @@ +"""Declared versus inferred units, and robust extents, on hand-built snapshots. + +Every expected number is worked out in the comment beside it from the +coordinates written here; nothing is read back from the implementation. +""" + +from __future__ import annotations + +import math + +import pytest + +from engineering.understand.snapshot import EntityRecord, Snapshot +from engineering.understand.units import entity_box, infer_units, robust_extents + +EPS = 1e-9 + + +def line(handle, a, b, layer="WALL"): + return EntityRecord(handle=handle, type="LINE", layer=layer, space="Model", points=(a, b)) + + +def text(handle, value, x, y, height, layer="TEXT"): + return EntityRecord( + handle=handle, + type="TEXT", + layer=layer, + space="Model", + points=((x, y),), + text=value, + height=height, + ) + + +def rect(prefix, x0, y0, x1, y1, layer="WALL"): + return [ + line(f"{prefix}1", (x0, y0), (x1, y0), layer), + line(f"{prefix}2", (x1, y0), (x1, y1), layer), + line(f"{prefix}3", (x1, y1), (x0, y1), layer), + line(f"{prefix}4", (x0, y1), (x0, y0), layer), + ] + + +def snapshot(records, insunits): + return Snapshot( + source="test", + insunits=insunits, + extmin=None, + extmax=None, + layers={}, + layouts=("Model",), + records=tuple(records), + ) + + +def room_in_mm(): + """One 6000 x 4000 room between 200 mm double-line walls, labelled with + 250-unit text: outer outline (0,0)-(6400,4400), inner (200,200)-(6200,4200).""" + return [ + *rect("O", 0.0, 0.0, 6400.0, 4400.0), + *rect("I", 200.0, 200.0, 6200.0, 4200.0), + text("T1", "ROOM 101 PROCESS", 1000.0, 2000.0, 250.0), + text("T2", "T4100", 3000.0, 1000.0, 250.0), + ] + + +def test_inches_declared_over_millimetre_geometry_is_flagged(): + # Inner room face 6000 x 4000 = 2.4e7 unit^2: 24 m2 as mm (plausible), + # 15 484 m2 as inches (out of the 1-5000 m2 band). Wall pairs 200 apart: + # 0.2 m as mm, 5.08 m as inches. Text 250: 0.25 m as mm, 6.35 m as inches. + result = infer_units(snapshot(room_in_mm(), insunits=1), room_areas=[6000.0 * 4000.0]) + assert result["declared"] == {"code": 1, "name": "in", "label": "Inches"} + assert result["inferred"] == "mm" + assert result["mm_per_unit"] == 1.0 + assert result["agree"] is False + assert result["warning"].startswith("INSUNITS declares Inches (1) but the geometry reads as mm") + kinds = {e["kind"]: e for e in result["evidence"]} + assert kinds["wall_thickness"]["value"] == pytest.approx(200.0, abs=EPS) + assert kinds["wall_thickness"]["plausible"] == ["mm"] + assert kinds["room_area"]["as"]["mm"] == pytest.approx(24.0, abs=EPS) + assert kinds["room_area"]["as"]["in"] == pytest.approx(24_000_000.0 * 0.0254**2, abs=EPS) + assert "in" not in kinds["text_height"]["plausible"] # 6.35 m of lettering + + +def test_millimetres_declared_over_the_same_geometry_agree(): + result = infer_units(snapshot(room_in_mm(), insunits=4), room_areas=[24_000_000.0]) + assert result["inferred"] == "mm" + assert result["agree"] is True + assert result["warning"] is None + assert 0.0 < result["confidence"] <= 0.9 + + +def test_one_weak_item_never_overrules_the_header(): + # Only an overall size: 20 000 units is 20 m as mm, 200 m as cm, 508 m as + # inches (all in the 0.01-2000 m band) and 20 km as metres (out of it). + # mm, cm and in score the size's 0.25 each; "m" scores the header's 0.5 + # alone and still wins - the declared metre stands. + records = [ + line("A", (0.0, 0.0), (20000.0, 0.0), "0"), + line("B", (0.0, 0.0), (0.0, 5000.0), "0"), + ] + result = infer_units(snapshot(records, insunits=6)) + assert result["inferred"] == "m" + assert result["agree"] is True + + +def test_no_evidence_keeps_the_declared_unit_at_low_confidence(): + result = infer_units(snapshot([], insunits=4)) + assert result["inferred"] == "mm" + assert result["confidence"] == 0.4 + assert result["evidence"] == [] + + +def test_unitless_with_no_evidence_settles_nothing(): + result = infer_units(snapshot([], insunits=0)) + assert result["inferred"] is None + assert result["mm_per_unit"] is None + assert result["warning"].endswith("lengths stay in drawing units.") + + +def test_a_far_outlier_is_found_by_handle_and_left_out_of_the_robust_box(): + records = [ + line(f"L{i}", (i * 1000.0, 0.0), (i * 1000.0 + 500.0, 800.0), "0") for i in range(12) + ] + far = line("FAR", (-5_000_000_000.0, 0.0), (-5_000_000_000.0 + 10.0, 0.0), "0") + result = robust_extents([*records, far]) + assert [o["handle"] for o in result["outliers"]] == ["FAR"] + # Robust box: x from 0 to 11 * 1000 + 500 = 11500, y from 0 to 800. + assert result["robust"] == pytest.approx([0.0, 0.0, 11500.0, 800.0], abs=EPS) + assert result["computed"][0] == pytest.approx(-5_000_000_000.0, abs=EPS) + # The outlier's centre (-4 999 999 995, 0) is 4 999 999 995 from the box. + assert result["outliers"][0]["distance"] == pytest.approx(4_999_999_995.0, abs=1e-3) + # Stretch: hypot(5 000 011 500, 800) / hypot(11 500, 800) - the computed + # diagonal over the robust one. + expected = math.hypot(5_000_011_500.0, 800.0) / math.hypot(11500.0, 800.0) + assert result["stretch"] == pytest.approx(expected, rel=1e-12) + + +def test_the_edge_walls_of_a_plan_are_not_outliers(): + # The room's centres crowd its middle (the vertical walls centre on + # y 2200, the label on y 2000), so a per-axis fence on centres would cut + # the top and bottom walls off; grown from the middle, every wall joins. + result = robust_extents(room_in_mm()) + assert result["outliers"] == [] + assert result["robust"] == pytest.approx([0.0, 0.0, 6400.0, 4400.0], abs=EPS) + assert result["stretch"] == pytest.approx(1.0, abs=EPS) + + +def test_fewer_than_four_entities_have_no_outliers(): + records = [line("A", (0.0, 0.0), (1.0, 0.0), "0"), line("B", (1e9, 0.0), (1e9 + 1.0, 0.0), "0")] + assert robust_extents(records)["outliers"] == [] + + +def test_entity_box_grows_a_circle_by_its_radius(): + circle = EntityRecord( + handle="C", type="CIRCLE", layer="0", space="Model", points=((10.0, 20.0),), radius=5.0 + ) + assert entity_box(circle) == (5.0, 15.0, 15.0, 25.0) + boxed = EntityRecord(handle="B", type="INSERT", layer="0", space="Model", bbox=(1, 2, 3, 4)) + assert entity_box(boxed) == (1.0, 2.0, 3.0, 4.0) + assert entity_box(EntityRecord(handle="N", type="HATCH", layer="0", space="Model")) is None