anatoly/autocad-mcp-pro
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
307ccc799fd81cbe198f42958208a4734080f5ab
autocad-mcp-pro/tests/test_understand_takeoff.py
T
Umut CANandClaude Opus 5.5 8c1a6cb27c refactor(understand): one reading of units, plan copies, scale and rooms 
Groups U and N were built side by side, so the takeoffs carried narrow
stand-ins for what group U reads properly. On main they read through U:

- describe.drawing_units is the one call every reader makes (kept records,
  labelled room faces as evidence); describe, scale_check and the takeoffs
  all use it. Without the room faces, a layout that declares inches over
  millimetres can tie three ways and fall to the header.
- layout_tags takes its plan copies from describe.prepare, room_regions its
  rooms from describe.find_rooms, and pipe_rows its scale from scale_check,
  per-room groups included. scale_stand_in and the takeoff's own unit bands
  and cluster grid are gone.
- test_understand_one_reading pins the agreement: one unit, one plan copy,
  one verdict and one set of rooms per drawing, across drawing_understand,
  drawing_scale_check and the two takeoffs.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
2026-09-24 14:45:27 +03:00

426 lines
17 KiB
Python
Raw Blame History

"""Pipe takeoff rows: layout lengths, statuses, rooms, units and the scale rule.
The hand-built pair: a P&ID (mm) with three tanks and two runs, and a layout
whose INSUNITS says inches over millimetre geometry - two rooms split by a
wall at x = 20000, the tags at T4100 (2000, 3000), T4101 (30000, 3000) and
T4102 (30000, 15000). Every expected length is worked in the comment beside it.
The last four tests read the synthetic plant pair of Task 2 and compare every
run with the generator's own hand-computed rectilinear MST.
"""
from __future__ import annotations
import pytest
from engineering.understand.network import build_network
from engineering.understand.scale import scale_check
from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot
from engineering.understand.takeoff import (
DEFAULT_SERVICES,
layout_tags,
pipe_rows,
unit_of,
)
from engineering.understand.vocab import classify_layer, room_label
from tests.fixtures.plant_pair import build_plant_pair
EPS = 1e-9
PIPE = "PRODUCT"
WALLS = "WALLS"
def rec(handle, kind, layer, points=(), **extra):
return EntityRecord(
handle=handle, type=kind, layer=layer, space="Model", points=tuple(points), **extra
)
def text(handle, value, at, height, layer="TEXT"):
x, y = at
return rec(
handle,
"TEXT",
layer,
(at,),
text=value,
height=height,
bbox=(x, y, x + height * len(value) * 0.6, y + height),
)
def snap(records, insunits, source):
layers = {r.layer: {"color": 7, "linetype": "Continuous"} for r in records}
return Snapshot(
source=source,
insunits=insunits,
extmin=None,
extmax=None,
layers=layers,
layouts=(),
records=tuple(records),
)
def tank(handle, tag, x0, y0):
box = (x0, y0, x0 + 500.0, y0 + 500.0)
return [
rec(
handle,
"LWPOLYLINE",
"EQUIPMENT",
((x0, y0), (x0 + 500.0, y0), (x0 + 500.0, y0 + 500.0), (x0, y0 + 500.0)),
closed=True,
bbox=box,
),
text(handle + "T", tag, (x0 + 150.0, y0 + 225.0), 50.0),
]
def pid_records():
return [
*tank("A", "T4100", 0.0, 0.0),
*tank("B", "T4101", 3000.0, 0.0),
*tank("C", "T4102", 3000.0, 2000.0),
rec("L1", "LINE", PIPE, ((500.0, 250.0), (3000.0, 250.0))),
rec("L2", "LINE", PIPE, ((3250.0, 500.0), (3250.0, 2000.0))),
text("D1", "Ø51", (1500.0, 260.0), 50.0),
]
def layout_records(dx=0.0, extra=()):
def at(x, y):
return (x + dx, y)
return [
rec(
f"W{dx}",
"LWPOLYLINE",
WALLS,
(at(0, 0), at(40000, 0), at(40000, 20000), at(0, 20000)),
closed=True,
bbox=(dx, 0.0, dx + 40000.0, 20000.0),
),
rec(f"X{dx}", "LINE", WALLS, (at(20000, 0), at(20000, 20000))),
text(f"R1{dx}", "ROOM 101", at(5000, 10000), 250.0),
text(f"R2{dx}", "ROOM 102", at(30000, 10000), 250.0),
text(f"T0{dx}", "T4100", at(2000, 3000), 250.0),
text(f"T1{dx}", "T4101", at(30000, 3000), 250.0),
text(f"T2{dx}", "T4102", at(30000, 15000), 250.0),
*extra,
]
def room_key(label):
found = room_label(label)
assert found is not None, f"the vocabulary must read {label!r} as a room label"
return str(found.get("number") or found.get("name"))
@pytest.fixture
def pair():
assert classify_layer(PIPE)["service"] == "product", "PRODUCT must be a product layer"
assert classify_layer(WALLS)["discipline"] == "architecture", "WALLS must be a wall layer"
pid = snap(pid_records(), 4, "pid.dxf")
layout = snap(layout_records(), 1, "layout.dxf") # INSUNITS 1 = inches, geometry in mm
return pid, layout, build_network(pid, layers=[PIPE])
def run_by_tags(result, *tags):
(run,) = [r for r in result["runs"] if r["tags"] == list(tags)]
return run
def test_layout_lengths_are_manhattan_between_the_tags(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout")
a = run_by_tags(result, "T4100", "T4101")
b = run_by_tags(result, "T4101", "T4102")
assert abs(a["layout_m"] - 28.0) < EPS # |30000 - 2000| + 0 mm
assert abs(b["layout_m"] - 12.0) < EPS # 0 + |15000 - 3000| mm
assert (a["status"], b["status"]) == ("A", "B") # b has no diameter label
assert result["length_source"] == "layout"
def test_review_focus_5_inches_declared_over_millimetres(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout")
units = result["units"]["layout"]
assert (units["declared"], units["inferred"], units["m_per_unit"]) == ("in", "mm", 0.001)
a = run_by_tags(result, "T4100", "T4101")
assert abs(a["layout_m"] - 28.0) < EPS # not 28000 x 0.0254 = 711.2
assert any(
"INSUNITS declares Inches (1)" in w and "reads as mm" in w for w in result["warnings"]
)
assert unit_of(pid)["warning"] is None # the P&ID's mm is what its geometry says
def test_rooms_share_the_route_x_first(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout")
a = run_by_tags(result, "T4100", "T4101")
# (2000, 3000) -> (30000, 3000): 18000 mm left of the wall at x = 20000, 10000 right
assert a["rooms"] == pytest.approx(
{room_key("ROOM 101"): 18.0, room_key("ROOM 102"): 10.0}, abs=EPS
)
rows = {(r["room"], r["diameter"]): r for r in result["rows"]}
assert rows[(room_key("ROOM 101"), "Ø51")]["net_m"] == pytest.approx(18.0, abs=EPS)
assert rows[(room_key("ROOM 101"), "Ø51")]["direct_m"] == pytest.approx(18.0, abs=EPS)
assert rows[(room_key("ROOM 102"), "unassigned")]["unassigned_m"] == pytest.approx(
12.0, abs=EPS
)
def test_allowance_is_its_own_column_and_totals_add_up(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout", allowance=0.25)
for row in result["rows"]:
assert row["allowance"] == 0.25
assert row["allowance_m"] == pytest.approx(row["net_m"] * 0.25, abs=EPS)
assert row["total_m"] == pytest.approx(row["net_m"] * 1.25, abs=EPS)
totals = result["totals"]
assert totals["net_m"] == pytest.approx(40.0, abs=EPS) # 28 + 12
assert totals["total_m"] == pytest.approx(50.0, abs=EPS)
assert totals["flagged_runs"] == 0
def test_vertical_allowance_is_added_per_connection(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout", vertical_allowance=0.5)
a = run_by_tags(result, "T4100", "T4101")
assert a["vertical_m"] == 0.5 # one tree edge
assert a["length_m"] == pytest.approx(28.5, abs=EPS)
def test_the_schematic_length_is_kept_beside_the_answer(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, length_source="layout")
a = run_by_tags(result, "T4100", "T4101")
assert a["schematic_m"] == pytest.approx(2.5, abs=EPS) # the 2500 mm line on the P&ID
def test_a_tag_missing_from_the_layout_makes_the_run_c(pair):
pid, _layout, network = pair
layout = snap([r for r in layout_records() if r.text != "T4102"], 1, "layout.dxf")
result = pipe_rows(network, pid, layout, length_source="layout")
b = run_by_tags(result, "T4101", "T4102")
assert (b["status"], b["reason"]) == ("C", "tag_not_on_layout")
(row,) = [c for c in result["control"] if c["kind"] == "unroutable"]
assert row["schematic_m"] == pytest.approx(1.5, abs=EPS) # the 1500 mm line, drawn length only
assert result["totals"]["unroutable_m"] == pytest.approx(1.5, abs=EPS)
assert result["totals"]["flagged_runs"] == 1
def test_a_tag_written_twice_in_the_plan_is_ambiguous_not_picked(pair):
pid, _layout, network = pair
extra = [text("dup", "T4101", (35000.0, 18000.0), 250.0)]
layout = snap(layout_records(extra=extra), 1, "layout.dxf")
result = pipe_rows(network, pid, layout, length_source="layout")
assert {r["reason"] for r in result["runs"]} == {"tag_ambiguous"}
def test_of_two_plan_copies_one_is_chosen_and_nothing_is_ambiguous(pair):
pid, _layout, network = pair
layout = snap(layout_records() + layout_records(dx=100000.0), 1, "layout.dxf")
found = layout_tags(layout, ["T4100", "T4101", "T4102"])
assert found["clusters"] == 2
assert found["cluster"][2] < 100000.0 # the left copy: ties go to the leftmost
assert found["ambiguous"] == []
result = pipe_rows(network, pid, layout, length_source="layout")
assert run_by_tags(result, "T4100", "T4101")["layout_m"] == pytest.approx(28.0, abs=EPS)
def test_a_run_with_a_free_end_is_c(pair):
pid, layout, _network = pair
records = pid_records() + [rec("L3", "LINE", PIPE, ((0.0, 500.0), (0.0, 1500.0)))]
drawing = snap(records, 4, "pid.dxf")
result = pipe_rows(
build_network(drawing, layers=[PIPE]), drawing, layout, length_source="layout"
)
reasons = {r["reason"] for r in result["runs"] if r["status"] == "C"}
assert reasons == {"end_not_on_equipment"}
def test_other_services_are_left_out_unless_asked(pair):
pid, layout, network = pair
assert DEFAULT_SERVICES == ("product", "cip_supply", "cip_return")
result = pipe_rows(network, pid, layout, length_source="layout", services=["steam"])
assert result["rows"] == []
assert result["method"]["excluded_services"] == {"product": 2}
# -- the scale rule ----------------------------------------------------------
SCHEMATIC = {
"verdict": "schematic",
"within_10pct": 0.25,
"pair_count": 12,
"median": 1.3,
"iqr": 0.8,
"factor": None,
"matched": [],
"ambiguous": [],
}
def test_auto_measures_a_schematic_pid_on_the_layout(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, scale=SCHEMATIC)
assert result["length_source"] == "layout"
assert result["method"]["reason"] == "auto_schematic"
assert result["scale_verified"] is True
def test_pid_on_a_schematic_pid_is_refused_with_the_statistics(pair):
pid, layout, network = pair
with pytest.raises(ValueError) as excinfo:
pipe_rows(network, pid, layout, scale=SCHEMATIC, length_source="pid")
message = str(excinfo.value)
assert message.startswith("length_source='pid': the scale check calls this P&ID schematic")
assert "25% of 12 tag pairs" in message and "median ratio 1.3" in message
assert "force=True" in message
def test_force_measures_on_the_pid_and_says_it_is_unverified(pair):
pid, layout, network = pair
result = pipe_rows(network, pid, layout, scale=SCHEMATIC, length_source="pid", force=True)
assert result["length_source"] == "pid"
assert result["method"]["reason"] == "forced"
assert result["scale_verified"] is False
assert run_by_tags(result, "T4100", "T4101")["length_m"] == pytest.approx(2.5, abs=EPS)
assert any(w.startswith("scale_verified: false") for w in result["warnings"])
def test_a_to_scale_pid_is_measured_through_its_factor(pair):
pid, layout, network = pair
to_scale = dict(SCHEMATIC, verdict="to_scale", within_10pct=0.9, factor=0.1)
result = pipe_rows(network, pid, layout, scale=to_scale)
assert result["length_source"] == "pid"
# 2.5 m drawn on the P&ID / 0.1 P&ID mm per layout mm = 25 m
assert run_by_tags(result, "T4100", "T4101")["length_m"] == pytest.approx(25.0, abs=EPS)
def test_without_a_layout_auto_measures_the_pid_unverified(pair):
pid, _layout, network = pair
result = pipe_rows(network, pid, None)
assert (result["length_source"], result["scale_verified"]) == ("pid", False)
assert result["method"]["reason"] == "no_layout"
with pytest.raises(ValueError, match="^length_source='layout' needs a layout drawing"):
pipe_rows(network, pid, None, length_source="layout")
def test_scale_check_recognises_a_uniform_scale_and_a_rearranged_one():
places = {
"T4100": (2000.0, 3000.0),
"T4101": (30000.0, 3000.0),
"T4102": (30000.0, 15000.0),
"T4103": (8000.0, 17000.0),
}
layout = snap(
[text(t, t, p, 250.0) for t, p in places.items()]
+ [rec("W", "LINE", WALLS, ((0.0, 0.0), (40000.0, 20000.0)))],
4,
"l",
)
half = snap([text(t, t, (x * 0.5, y * 0.5), 50.0) for t, (x, y) in places.items()], 4, "p")
uniform = scale_check(half, layout)
assert uniform["verdict"] == "to_scale"
assert uniform["factor"] == pytest.approx(0.5, abs=EPS) # P&ID mm per layout mm
assert uniform["pair_count"] == 6 # four tags, every pair at least 2 m apart
shuffled = {
"T4100": (30000.0, 15000.0),
"T4101": (2000.0, 3000.0),
"T4102": (8000.0, 17000.0),
"T4103": (30000.0, 3000.0),
}
other = snap([text(t, t, p, 50.0) for t, p in shuffled.items()], 4, "p")
assert scale_check(other, layout)["verdict"] == "schematic"
def test_scale_check_calls_coinciding_pid_tags_insufficient_not_to_scale():
# every P&ID tag at one point: each ratio is 0, the median 0 - a factor of 0
# would divide every length by zero, so the verdict is insufficient (the
# same rule as group U's scale_check: a median that is not positive)
places = {"T4100": (2000.0, 3000.0), "T4101": (30000.0, 3000.0), "T4102": (30000.0, 15000.0)}
layout = snap([text(t, t, p, 250.0) for t, p in places.items()], 4, "l")
pid = snap([text(t, t, (100.0, 100.0), 50.0) for t in places], 4, "p")
result = scale_check(pid, layout)
assert result["pair_count"] == 3
assert (result["verdict"], result["factor"]) == ("insufficient", None)
def test_an_insunits_code_this_reader_does_not_convert_is_named_not_called_none():
drawing = snap(layout_records(), 14, "layout.dxf") # 14 = decimetres
found = unit_of(drawing)
assert (found["declared"], found["inferred"]) == (None, "mm")
assert "INSUNITS declares Decimeters (14) but the geometry reads as mm" in found["warning"]
assert "no unit" not in found["warning"]
@pytest.mark.parametrize(
("kwargs", "message"),
[
({"allowance": 1.5}, "allowance: must lie between 0 and 1"),
({"allowance": -0.1}, "allowance: must lie between 0 and 1"),
({"vertical_allowance": -1.0}, "vertical_allowance: must be zero or more metres"),
({"length_source": "tape"}, "length_source: 'tape' unknown"),
({"services": ["lava"]}, "services: lava unknown"),
({"services": []}, "services: an empty list takes nothing off"),
],
)
def test_refusals_name_the_option(pair, kwargs, message):
pid, layout, network = pair
with pytest.raises(ValueError, match="^" + message):
pipe_rows(network, pid, layout, **kwargs)
# -- the synthetic plant pair (Task 2) -------------------------------------------
@pytest.fixture(scope="module")
def plant(tmp_path_factory):
truth = build_plant_pair(tmp_path_factory.mktemp("plant"))
pid, layout = read_snapshot(truth["pid"]), read_snapshot(truth["layout"])
return truth, pid, layout, build_network(pid)
def truth_services(truth):
return sorted({run["service"] for run in truth["pipe_runs"]})
def test_every_synthetic_layout_length_is_the_hand_computed_rmst(plant):
truth, pid, layout, network = plant
result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout")
runs = {(tuple(sorted(r["tags"])), r["service"]): r for r in result["runs"]}
for expected in truth["pipe_runs"]:
run = runs[(tuple(sorted(expected["tags"])), expected["service"])]
assert run["status"] == expected["status"], run["id"]
if expected["status"] != "C":
assert run["layout_m"] * 1000.0 == pytest.approx(expected["layout_length_mm"], abs=1e-6)
assert sum(run["rooms"].values()) == pytest.approx(run["length_m"], abs=1e-9)
def test_auto_calls_the_synthetic_pid_schematic_and_measures_the_layout(plant):
truth, pid, layout, network = plant
result = pipe_rows(network, pid, layout, services=truth_services(truth))
assert result["scale"]["verdict"] == truth["scale_verdict"] == "schematic"
assert result["length_source"] == "layout"
with pytest.raises(ValueError, match="the scale check calls this P&ID schematic"):
pipe_rows(network, pid, layout, services=truth_services(truth), length_source="pid")
def test_review_focus_5_the_synthetic_layout_is_read_in_millimetres(plant):
truth, pid, layout, network = plant
assert layout.insunits == truth["layout_insunits"] == 1
result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout")
assert result["units"]["layout"]["inferred"] == truth["layout_true_unit"] == "mm"
assert result["units"]["layout"]["m_per_unit"] == 0.001
assert any("INSUNITS declares Inches (1)" in w for w in result["warnings"])
def test_routed_lengths_land_in_the_synthetic_rooms(plant):
truth, pid, layout, network = plant
result = pipe_rows(network, pid, layout, services=truth_services(truth), length_source="layout")
known = {str(room["number"] or room["name"]) for room in truth["rooms"]} | {""}
for run in result["runs"]:
assert set(run["rooms"]) <= known, run["id"]
Reference in New Issue View Git Blame Copy Permalink