"""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"]