From 7aaead43d9aed3ab835288604db80042520cfe27 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Thu, 24 Sep 2026 11:55:57 +0300 Subject: [PATCH] feat(understand): drawing_scale_check - is this drawing to scale against that one Equipment tags found exactly once in each drawing are matched (a tag found twice - a plan copy, a legend - is reported ambiguous and takes no part; two plan copies resolve to the cluster that matches the most tags). For every pair of matched tags at least min_pair_mm apart on the reference, the ratio of their distances is taken in millimetres by each drawing's inferred unit; the median, the quartiles, the share within +/-10 % of the median, the same per room of the reference, and the verdict: to_scale (>= 80 %, with its factor), schematic (<= 50 %), partly, or insufficient under three matched tags. server.py grows SECTION 26 to two tools with drawing_scale_check (core, read-only, tagged analysis/query); alias record, inventory, README release snapshot 243, frozen group sizes (analysis 14). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- README.md | 6 +- discovery/aliases.py | 19 ++ docs/tool-inventory.json | 18 +- engineering/understand/scale.py | 272 +++++++++++++++++++++++++++ server.py | 87 ++++++++- tests/test_tool_registry.py | 13 +- tests/test_understand_scale.py | 192 +++++++++++++++++++ tests/test_understand_scale_tools.py | 100 ++++++++++ 8 files changed, 695 insertions(+), 12 deletions(-) create mode 100644 engineering/understand/scale.py create mode 100644 tests/test_understand_scale.py create mode 100644 tests/test_understand_scale_tools.py diff --git a/README.md b/README.md index b33ebea..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:** 242 tools · 8 resources · 5 prompt templates · 4516 collected tests. -> 242 is the **registered** count; a default install advertises 237 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 | -242 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 a0bda5e..9e00ef7 100644 --- a/discovery/aliases.py +++ b/discovery/aliases.py @@ -2494,6 +2494,25 @@ TOOL_ALIASES: dict[str, ToolAliases] = { "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 ff6e76b..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": 242, + "tools": 243, "resources": 8, "prompts": 5, "groups": 26 }, "groups": { - "analysis": 13, + "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", 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/server.py b/server.py index 29a0677..3284d6a 100644 --- a/server.py +++ b/server.py @@ -10555,7 +10555,7 @@ async def arch_plan_from_spec( # --------------------------------------------------------------------------- -# ── SECTION 26: Understanding & QA (1 tool) ───────────────────────────────── +# ── SECTION 26: Understanding & QA (2 tools) ──────────────────────────────── # --------------------------------------------------------------------------- @@ -10628,6 +10628,91 @@ async def drawing_understand( 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 c9ce514..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 (242 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 @@ -595,13 +595,14 @@ async def test_tool_group_sizes_are_unchanged(): 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. This snapshot is - branch-local: the track H merge task recomputes it once the understand, - network and diff branches have landed. + (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": 13, + "analysis": 14, "architecture": 12, "batch": 3, "blocks": 9, @@ -628,4 +629,4 @@ async def test_tool_group_sizes_are_unchanged(): "validation": 1, "view": 4, } - assert sum(sizes.values()) == 242 + assert sum(sizes.values()) == 243 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