merge: track H group U

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-24 12:30:30 +03:00
co-authored by Claude Opus 5.5
15 changed files with 2501 additions and 8 deletions
+11 -3
View File
@@ -495,7 +495,7 @@ async def test_tool_groups_is_byte_identical_to_the_source_declared_grouping():
async def test_tool_group_sizes_are_unchanged():
"""Frozen snapshot of the surface (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
+86
View File
@@ -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)
+301
View File
@@ -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
+192
View File
@@ -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
+100
View File
@@ -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
+82
View File
@@ -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
+161
View File
@@ -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