anatoly/autocad-mcp-pro
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity
Files
e640def4ffa689858c2441b82eff2334d1e1b93e
autocad-mcp-pro/tests/test_understand_takeoff.py
T
Umut CANandClaude Opus 5.5 e640def4ff feat(understand): orthogonal lengths and pipe takeoff rows 
route.py: Manhattan distance, the rectilinear minimum spanning tree (Prim
over |dx| + |dy|, exact on the points given) and the x-then-y route.

pipe_rows turns the P&ID's runs into rows by room x service x diameter.
length_source auto measures on the P&ID only when the scale check calls it
to scale; otherwise each run is the RMST of its tags' layout positions plus
vertical_allowance per connection, with the drawn length kept beside it as
schematic_m. length_source=pid on a schematic P&ID is refused with the
statistics unless force=True. Status A / B / C; a C run goes to control
with its schematic length, counted and flagged. Routed lengths are shared
among the layout rooms their route crosses (track F's face finder); the
allowance is its own column. Lengths are metres from the unit the geometry
implies: a layout declaring inches over millimetres is read in mm and warned
about, never multiplied by 25.4.

The takeoff carries narrow stand-ins for the unit inference, the plan-copy
clusters and the scale check (group U's modules are on another branch); a
scale result passed in is used first. Tags written twice in the chosen
cluster are ambiguous, never picked.

Beyond the plan: the scale stand-in calls a non-positive median ratio
insufficient (as group U's scale_check does) instead of to_scale with a
factor of 0; an INSUNITS code the reader does not convert is named in the
unit warning instead of being called "no unit"; wall layers are classified
once per layer, not once per record.

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

424 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.snapshot import EntityRecord, Snapshot, read_snapshot
from engineering.understand.takeoff import (
DEFAULT_SERVICES,
layout_tags,
pipe_rows,
scale_stand_in,
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 in" in w and "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_the_stand_in_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_stand_in(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_stand_in(other, layout)["verdict"] == "schematic"
def test_the_stand_in_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_stand_in(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 code 14 (a unit this reader does not convert)" 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 in" 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