merge: track H group N
The merge agent stopped this branch on an add/add conflict it had not been told to expect: groups U and N each created tests/test_understand_tools.py. Resolved by hand: - tests/test_understand_tools.py stays U's (the drawing_understand wire tests), and N's takeoff wire tests move whole to tests/test_understand_takeoff_tools.py. Each file keeps its own client fixture: U's opens a fresh document, N's reads files only. - server.py: SECTION 26 (4 tools, U then D) is kept as merged, and SECTION 27 (Plant Takeoffs, 2 tools) follows it whole, above the RESOURCES banner. - discovery/aliases.py: additive; the two takeoff records follow the four readers. - tests/test_tool_registry.py: all three branches' paragraphs are kept. The snapshot is recomputed from the live registry: analysis 18, 247 tools. Task 11 ranks `understand` and `plant`. - docs/tool-inventory.json regenerated (247 tools, 26 groups); README counts are 247 registered / 242 advertised. The collected-tests figure is left to Task 11. ruff check passes. Suite: 4919 passed, 1 xfailed. The one deselected test is the collected-tests figure. 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
@@ -609,10 +609,24 @@ async def test_tool_group_sizes_are_unchanged():
|
||||
|
||||
Merged, the two SECTION 26 halves file all four readers under `analysis`
|
||||
(12 -> 16).
|
||||
|
||||
`analysis` moved 12 -> 13 when track H's SECTION 27 opened with
|
||||
`pipe_takeoff` on the network branch. It is tagged `plant` and
|
||||
`analysis`, and `plant` is not in `_GROUP_TAG_PRIORITY` on that
|
||||
branch, so it files under `analysis`. This snapshot is branch-local:
|
||||
the track H integration task gives `plant` its pack and its group
|
||||
once every branch has landed.
|
||||
|
||||
It moved 13 -> 14 when `cable_takeoff` joined the same section, for
|
||||
the same branch-local reason.
|
||||
|
||||
With SECTION 27 merged beside them, the four readers and the two
|
||||
takeoffs all file under `analysis` (12 -> 18) until the track H
|
||||
integration task ranks `understand` and `plant`.
|
||||
"""
|
||||
sizes = {label: len(names) for label, names in (await server._tool_groups()).items()}
|
||||
assert sizes == {
|
||||
"analysis": 16,
|
||||
"analysis": 18,
|
||||
"architecture": 12,
|
||||
"batch": 3,
|
||||
"blocks": 9,
|
||||
@@ -639,4 +653,4 @@ async def test_tool_group_sizes_are_unchanged():
|
||||
"validation": 1,
|
||||
"view": 4,
|
||||
}
|
||||
assert sum(sizes.values()) == 245
|
||||
assert sum(sizes.values()) == 247
|
||||
|
||||
@@ -0,0 +1,386 @@
|
||||
"""Cable takeoff rows: loads from the P&ID, panels and Manhattan lengths from the layout.
|
||||
|
||||
The hand-built pair. On the P&ID: M82 with 'P = 18,5 кВт' and '3 ф. 400 В +
|
||||
N + PE' under it, M87 with only '3 ф. 400 В + N + PE' (no power stated), the
|
||||
package panel CP-M88 with 'P = 15 кВт' and its child motor M88 with
|
||||
'P = 4 кВт'. On the layout (mm): two rooms split at x = 20000, the panel CP-1
|
||||
at (15000, 1000), M82 (2000, 3000), M87 (4000, 3000), CP-M88 (25000, 5000),
|
||||
M88 (26000, 8000), each with a 'Wiring to ...' callout 300 mm below it.
|
||||
Every expected length is worked in the comment beside it. The last two tests
|
||||
read the synthetic plant pair of Task 2 and compare every cable row with the
|
||||
generator's own Manhattan lengths.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
|
||||
import pytest
|
||||
|
||||
from engineering.understand.labels import normalize_panel
|
||||
from engineering.understand.report import write_workbook
|
||||
from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot
|
||||
from engineering.understand.takeoff import cable_rows, roundup_m
|
||||
from engineering.understand.vocab import classify_layer, room_label
|
||||
from tests.fixtures.plant_pair import build_plant_pair
|
||||
|
||||
EPS = 1e-9
|
||||
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 pid_records():
|
||||
return [
|
||||
text("a", "M82", (0.0, 0.0), 50.0),
|
||||
text("a1", "P = 18,5 кВт", (0.0, -80.0), 50.0),
|
||||
text("a2", "3 ф. 400 В + N + PE", (0.0, -160.0), 50.0),
|
||||
text("b", "M87", (1000.0, 0.0), 50.0),
|
||||
text("b1", "3 ф. 400 В + N + PE", (1000.0, -80.0), 50.0),
|
||||
text("c", "CP-M88", (2000.0, 0.0), 50.0),
|
||||
text("c1", "P = 15 кВт", (2000.0, -80.0), 50.0),
|
||||
text("d", "M88", (3000.0, 0.0), 50.0),
|
||||
text("d1", "P = 4 кВт", (3000.0, -80.0), 50.0),
|
||||
]
|
||||
|
||||
|
||||
def layout_records(skip=()):
|
||||
places = {
|
||||
"M82": (2000.0, 3000.0),
|
||||
"M87": (4000.0, 3000.0),
|
||||
"CP-M88": (25000.0, 5000.0),
|
||||
"M88": (26000.0, 8000.0),
|
||||
}
|
||||
panels = {"M82": "CP1", "M87": "CP 1", "CP-M88": "CP-1", "M88": "CP-M88"}
|
||||
records = [
|
||||
rec(
|
||||
"W",
|
||||
"LWPOLYLINE",
|
||||
WALLS,
|
||||
((0, 0), (40000, 0), (40000, 20000), (0, 20000)),
|
||||
closed=True,
|
||||
bbox=(0.0, 0.0, 40000.0, 20000.0),
|
||||
),
|
||||
rec("X", "LINE", WALLS, ((20000, 0), (20000, 20000))),
|
||||
text("R1", "ROOM 101", (5000.0, 10000.0), 250.0),
|
||||
text("R2", "ROOM 102", (30000.0, 10000.0), 250.0),
|
||||
text("P1", "CP-1", (15000.0, 1000.0), 250.0),
|
||||
]
|
||||
for tag, (x, y) in places.items():
|
||||
if tag in skip:
|
||||
continue
|
||||
records.append(text("t" + tag, tag, (x, y), 250.0))
|
||||
records.append(text("w" + tag, f"Wiring to {panels[tag]}", (x, y - 300.0), 250.0))
|
||||
return records
|
||||
|
||||
|
||||
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(WALLS)["discipline"] == "architecture", "WALLS must be a wall layer"
|
||||
return snap(pid_records(), 4, "pid.dxf"), snap(layout_records(), 4, "layout.dxf")
|
||||
|
||||
|
||||
def by_tag(result):
|
||||
return {row["tag"]: row for row in result["rows"]}
|
||||
|
||||
|
||||
def test_every_load_gets_its_panel_length_and_rounded_cable(pair):
|
||||
pid, layout = pair
|
||||
rows = by_tag(cable_rows(pid, layout))
|
||||
assert set(rows) == {"M82", "M87", "CP-M88", "M88"}
|
||||
expected = {
|
||||
# tag: (panel, Manhattan mm, ROUNDUP(m x 1.2))
|
||||
"M82": ("CP-1", 15000.0, 18), # |15000-2000| + |1000-3000| = 13000 + 2000
|
||||
"M87": ("CP-1", 13000.0, 16), # 11000 + 2000 -> 15.6 -> 16
|
||||
"CP-M88": ("CP-1", 14000.0, 17), # 10000 + 4000 -> 16.8 -> 17
|
||||
"M88": ("CP-M88", 4000.0, 5), # 1000 + 3000 -> 4.8 -> 5
|
||||
}
|
||||
for tag, (panel, mm, cable) in expected.items():
|
||||
row = rows[tag]
|
||||
assert row["panel"] == panel # 'CP1', 'CP 1' and 'CP-1' are one panel
|
||||
assert row["panel_source"] == "layout"
|
||||
assert row["length_m"] * 1000.0 == pytest.approx(mm, abs=1e-6)
|
||||
assert row["cable_m"] == cable
|
||||
|
||||
|
||||
def test_power_and_supply_are_read_from_the_pid_texts(pair):
|
||||
pid, layout = pair
|
||||
m82 = by_tag(cable_rows(pid, layout))["M82"]
|
||||
assert m82["kw"] == pytest.approx(18.5)
|
||||
assert (m82["phases"], m82["voltage"], m82["neutral"]) == (3, 400, True)
|
||||
|
||||
|
||||
def test_power_is_never_invented(pair):
|
||||
pid, layout = pair
|
||||
result = cable_rows(pid, layout)
|
||||
m87 = by_tag(result)["M87"]
|
||||
assert m87["kw"] is None # the P&ID states phases and voltage, no kW
|
||||
assert m87["cable_m"] == 16 # the cable is still measured
|
||||
assert {
|
||||
"tag": "M87",
|
||||
"item": "missing_power",
|
||||
"detail": "no power is stated near the tag on the P&ID",
|
||||
} in result["open_items"]
|
||||
assert all(item["tag"] == "M87" for item in result["open_items"])
|
||||
|
||||
|
||||
def test_room_totals_count_a_package_and_not_its_children(pair):
|
||||
pid, layout = pair
|
||||
result = cable_rows(pid, layout)
|
||||
rows = by_tag(result)
|
||||
assert rows["M88"]["package"] == "CP-M88"
|
||||
assert rows["M82"]["room"] == room_key("ROOM 101")
|
||||
assert rows["CP-M88"]["room"] == room_key("ROOM 102")
|
||||
rooms = {item["key"]: item for item in result["summary"]["by_room"]}
|
||||
assert rooms[room_key("ROOM 101")]["kw"] == pytest.approx(18.5) # M82; M87 has none
|
||||
assert rooms[room_key("ROOM 102")]["kw"] == pytest.approx(15.0) # CP-M88, not + M88's 4
|
||||
assert rooms[room_key("ROOM 102")]["loads"] == 2
|
||||
assert result["totals"]["known_kw"] == pytest.approx(33.5)
|
||||
assert result["totals"]["cable_m"] == 18 + 16 + 17 + 5
|
||||
|
||||
|
||||
def test_roundup_does_not_buy_a_metre_of_binary_noise():
|
||||
assert 50.0 * 1.1 > 55.0 # 55.00000000000001 in floating point
|
||||
assert roundup_m(50.0, 0.10) == 55
|
||||
assert roundup_m(15.0, 0.20) == 18
|
||||
assert roundup_m(15.001, 0.20) == 19 # 18.0012 -> 19
|
||||
|
||||
|
||||
def test_without_a_layout_lengths_stay_empty_with_an_open_item(pair):
|
||||
pid, _layout = pair
|
||||
result = cable_rows(pid, None)
|
||||
assert all(row["length_m"] is None and row["cable_m"] is None for row in result["rows"])
|
||||
no_layout = [item["tag"] for item in result["open_items"] if item["item"] == "no_layout"]
|
||||
assert sorted(no_layout) == ["CP-M88", "M82", "M87", "M88"]
|
||||
# no callout on the P&ID either: the panel is open, never guessed
|
||||
assert all(row["panel"] is None for row in result["rows"])
|
||||
|
||||
|
||||
def test_a_tag_missing_from_the_layout_is_an_open_item(pair):
|
||||
pid, _layout = pair
|
||||
layout = snap(layout_records(skip=("M87",)), 4, "layout.dxf")
|
||||
result = cable_rows(pid, layout)
|
||||
items = {(item["tag"], item["item"]) for item in result["open_items"]}
|
||||
assert ("M87", "tag_not_on_layout") in items
|
||||
assert ("M87", "missing_panel") in items # its callout went with it
|
||||
assert by_tag(result)["M87"]["cable_m"] is None
|
||||
|
||||
|
||||
def test_sections_come_only_from_the_callers_rules(pair):
|
||||
pid, layout = pair
|
||||
assert all(row["section"] is None for row in cable_rows(pid, layout)["rows"])
|
||||
rules = [{"max_kw": 40, "section": "5x10"}, {"max_kw": 5, "section": "5x2.5"}]
|
||||
rows = by_tag(cable_rows(pid, layout, section_rules=rules))
|
||||
assert rows["M82"]["section"] == "5x10" # 18.5 kW
|
||||
assert rows["M88"]["section"] == "5x2.5" # 4 kW: the smaller rule is tried first
|
||||
assert rows["M87"]["section"] is None # no power, no section
|
||||
|
||||
|
||||
def test_a_load_beyond_every_rule_is_an_open_item(pair):
|
||||
pid, layout = pair
|
||||
result = cable_rows(pid, layout, section_rules=[{"max_kw": 10, "section": "5x4"}])
|
||||
assert {"tag": "M82", "item": "no_section_rule", "detail": "no rule covers 18.5 kW"} in (
|
||||
result["open_items"]
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("kwargs", "message"),
|
||||
[
|
||||
({"allowance": 2.0}, "allowance: must lie between 0 and 1"),
|
||||
({"section_rules": {"max_kw": 5}}, "section_rules: expected a list"),
|
||||
(
|
||||
{"section_rules": [{"max_kw": 0, "section": "x"}]},
|
||||
r"section_rules\[0\]\.max_kw: must be",
|
||||
),
|
||||
({"section_rules": [{"max_kw": 5}]}, r"section_rules\[0\]\.section: expected a non-empty"),
|
||||
(
|
||||
{"section_rules": [{"max_kw": 5, "section": "x", "phases": 2}]},
|
||||
r"section_rules\[0\]\.phases: 1 or 3",
|
||||
),
|
||||
(
|
||||
{"section_rules": [{"max_kw": 5, "section": "x", "iec": True}]},
|
||||
r"section_rules\[0\]: unknown key\(s\) iec",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_refusals_name_the_option(pair, kwargs, message):
|
||||
pid, layout = pair
|
||||
with pytest.raises(ValueError, match="^" + message):
|
||||
cable_rows(pid, layout, **kwargs)
|
||||
|
||||
|
||||
CABLE_HEADERS = {
|
||||
"tr": [
|
||||
"Etiket",
|
||||
"Mahal",
|
||||
"Güç (kW)",
|
||||
"Faz",
|
||||
"Gerilim (V)",
|
||||
"Nötr",
|
||||
"Sürücü (VFD)",
|
||||
"Pano",
|
||||
"Mesafe (m)",
|
||||
"Pay oranı",
|
||||
"Kablo (m)",
|
||||
"Kesit",
|
||||
"Paket",
|
||||
],
|
||||
"en": [
|
||||
"Tag",
|
||||
"Room",
|
||||
"Power (kW)",
|
||||
"Phases",
|
||||
"Voltage (V)",
|
||||
"Neutral",
|
||||
"VFD",
|
||||
"Panel",
|
||||
"Distance (m)",
|
||||
"Allowance",
|
||||
"Cable (m)",
|
||||
"Section",
|
||||
"Package",
|
||||
],
|
||||
"ru": [
|
||||
"Тег",
|
||||
"Помещение",
|
||||
"Мощность (кВт)",
|
||||
"Фазы",
|
||||
"Напряжение (В)",
|
||||
"Нейтраль",
|
||||
"ЧРП",
|
||||
"Шкаф",
|
||||
"Расстояние (м)",
|
||||
"Запас",
|
||||
"Кабель (м)",
|
||||
"Сечение",
|
||||
"Комплект",
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lang", ["tr", "en", "ru"])
|
||||
def test_the_cable_workbook_speaks_three_languages(pair, tmp_path, lang):
|
||||
pid, layout = pair
|
||||
files = write_workbook(
|
||||
cable_rows(pid, layout), kind="cable", lang=lang, path=str(tmp_path / "cables.xlsx")
|
||||
)
|
||||
with open(files["csv"][0], encoding="utf-8-sig", newline="") as handle:
|
||||
metraj = list(csv.reader(handle))
|
||||
assert metraj[0] == CABLE_HEADERS[lang]
|
||||
assert metraj[-1][10] == str(18 + 16 + 17 + 5)
|
||||
with open(files["csv"][2], encoding="utf-8-sig", newline="") as handle:
|
||||
kontrol = list(csv.reader(handle))
|
||||
assert [row[0] for row in kontrol[1:]] == ["M87"] # the one open item: no power
|
||||
|
||||
|
||||
def test_the_method_sheet_states_the_package_rule_and_that_power_is_never_invented(pair, tmp_path):
|
||||
pid, layout = pair
|
||||
files = write_workbook(
|
||||
cable_rows(pid, layout), kind="cable", lang="en", path=str(tmp_path / "cables.xlsx")
|
||||
)
|
||||
with open(files["csv"][3], encoding="utf-8-sig", newline="") as handle:
|
||||
method = {row[0]: row[1] for row in list(csv.reader(handle))[1:]}
|
||||
assert "never invented" in method["Rule: power"]
|
||||
assert "counted twice" in method["Rule: panel packages"]
|
||||
assert method["Section rules (the caller's)"] == "—"
|
||||
|
||||
|
||||
# -- the synthetic plant pair (Task 2) -------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def plant(tmp_path_factory):
|
||||
truth = build_plant_pair(tmp_path_factory.mktemp("plant"))
|
||||
return truth, read_snapshot(truth["pid"]), read_snapshot(truth["layout"])
|
||||
|
||||
|
||||
def inside(point, polygon):
|
||||
"""Even-odd ray cast over the generator's own room polygon."""
|
||||
x, y = point
|
||||
hit = False
|
||||
for (x1, y1), (x2, y2) in zip(polygon, polygon[1:] + polygon[:1], strict=False):
|
||||
if (y1 > y) != (y2 > y) and x < x1 + (y - y1) * (x2 - x1) / (y2 - y1):
|
||||
hit = not hit
|
||||
return hit
|
||||
|
||||
|
||||
def test_every_synthetic_cable_row_equals_the_generators_truth(plant):
|
||||
truth, pid, layout = plant
|
||||
result = cable_rows(pid, layout)
|
||||
rows = by_tag(result)
|
||||
assert set(rows) == {expected["tag"] for expected in truth["cable_rows"]}
|
||||
for expected in truth["cable_rows"]:
|
||||
row = rows[expected["tag"]]
|
||||
assert row["panel"] == expected["panel"], expected["tag"]
|
||||
if expected["kw"] is None:
|
||||
assert row["kw"] is None, expected["tag"]
|
||||
assert {
|
||||
"tag": expected["tag"],
|
||||
"item": "missing_power",
|
||||
"detail": "no power is stated near the tag on the P&ID",
|
||||
} in result["open_items"]
|
||||
else:
|
||||
assert row["kw"] == pytest.approx(expected["kw"]), expected["tag"]
|
||||
assert row["length_m"] * 1000.0 == pytest.approx(expected["length_mm"], abs=1e-6)
|
||||
assert row["cable_m"] == expected["roundup_m"], expected["tag"]
|
||||
|
||||
|
||||
def test_the_synthetic_room_totals_count_each_power_once(plant):
|
||||
truth, pid, layout = plant
|
||||
result = cable_rows(pid, layout)
|
||||
rooms = [
|
||||
(str(room["number"] or room["name"]), [tuple(p) for p in room["polygon"]])
|
||||
for room in truth["rooms"]
|
||||
]
|
||||
for row in result["rows"]:
|
||||
homes = [key for key, polygon in rooms if row["at"] and inside(row["at"], polygon)]
|
||||
if homes:
|
||||
assert row["room"] == homes[0], row["tag"]
|
||||
stated = {
|
||||
normalize_panel(r["tag"]) or r["tag"] for r in truth["cable_rows"] if r["kw"] is not None
|
||||
}
|
||||
|
||||
def child(r):
|
||||
return r["panel"] in stated and (normalize_panel(r["tag"]) or r["tag"]) != r["panel"]
|
||||
|
||||
expected_kw = sum(r["kw"] for r in truth["cable_rows"] if r["kw"] is not None and not child(r))
|
||||
assert result["totals"]["known_kw"] == pytest.approx(expected_kw)
|
||||
by_room = sum(item["kw"] for item in result["summary"]["by_room"])
|
||||
assert by_room == pytest.approx(expected_kw)
|
||||
@@ -0,0 +1,635 @@
|
||||
"""The line-network engine: runs, duplicates once, fittings, diameters, services, equipment.
|
||||
|
||||
Every drawing below is built record by record, so each expected length is the
|
||||
hand sum written beside it; nothing is read back from the implementation. The
|
||||
last three tests read the synthetic plant pair of Task 2 and compare every run
|
||||
with the generator's own truth.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from engineering.understand.network import (
|
||||
Run,
|
||||
build_network,
|
||||
default_layers,
|
||||
is_reducer,
|
||||
label_search_default,
|
||||
)
|
||||
from engineering.understand.snapshot import EntityRecord, Snapshot, read_snapshot
|
||||
from engineering.understand.vocab import classify_layer, equipment_kind
|
||||
from tests.fixtures.plant_pair import build_plant_pair
|
||||
|
||||
EPS = 1e-9
|
||||
PIPE = "PRODUCT"
|
||||
|
||||
|
||||
def rec(handle, kind, layer=PIPE, points=(), **extra):
|
||||
return EntityRecord(
|
||||
handle=handle, type=kind, layer=layer, space="Model", points=tuple(points), **extra
|
||||
)
|
||||
|
||||
|
||||
def line(handle, a, b, layer=PIPE):
|
||||
return rec(handle, "LINE", layer, (a, b))
|
||||
|
||||
|
||||
def text(handle, value, at, height=5.0, 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 tank(handle, tag, box, layer="EQUIPMENT"):
|
||||
"""A closed rectangle with its tag written inside it."""
|
||||
x0, y0, x1, y1 = box
|
||||
outline = rec(
|
||||
handle, "LWPOLYLINE", layer, ((x0, y0), (x1, y0), (x1, y1), (x0, y1)), closed=True, bbox=box
|
||||
)
|
||||
return [outline, text(handle + "T", tag, ((x0 + x1) / 2.0 - 10.0, (y0 + y1) / 2.0))]
|
||||
|
||||
|
||||
def insert(handle, block, box, layer=PIPE):
|
||||
return rec(
|
||||
handle,
|
||||
"INSERT",
|
||||
layer,
|
||||
(((box[0] + box[2]) / 2.0, (box[1] + box[3]) / 2.0),),
|
||||
block=block,
|
||||
bbox=box,
|
||||
)
|
||||
|
||||
|
||||
def snap(*records, insunits=4):
|
||||
flat = []
|
||||
for item in records:
|
||||
flat.extend(item if isinstance(item, list) else [item])
|
||||
layers = {r.layer: {"color": 7, "linetype": "Continuous"} for r in flat}
|
||||
return Snapshot(
|
||||
source="test",
|
||||
insunits=insunits,
|
||||
extmin=None,
|
||||
extmax=None,
|
||||
layers=layers,
|
||||
layouts=(),
|
||||
records=tuple(flat),
|
||||
)
|
||||
|
||||
|
||||
def total(run: Run) -> float:
|
||||
return sum(edge["length"] for edge in run.edges)
|
||||
|
||||
|
||||
def net(*records, **kw):
|
||||
return build_network(snap(*records), layers=[PIPE], **kw)
|
||||
|
||||
|
||||
# -- topology ------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_ends_within_tol_merge_into_one_run():
|
||||
result = net(line("1", (0.0, 0.0), (1000.0, 0.0)), line("2", (1000.4, 0.0), (2000.0, 0.0)))
|
||||
(run,) = result["runs"]
|
||||
# the second line starts at the merged vertex (1000, 0): 1000 + 1000
|
||||
assert abs(total(run) - 2000.0) < EPS
|
||||
assert run.ends == ((0.0, 0.0), (2000.0, 0.0))
|
||||
|
||||
|
||||
def test_a_branch_ending_on_a_header_is_a_t_junction():
|
||||
result = net(line("1", (0.0, 0.0), (2000.0, 0.0)), line("2", (1000.0, 0.0), (1000.0, 800.0)))
|
||||
(run,) = result["runs"]
|
||||
assert abs(total(run) - 2800.0) < EPS # 2000 header + 800 branch
|
||||
assert len(run.edges) == 3 # header split at the tee
|
||||
assert run.ends == ((0.0, 0.0), (1000.0, 800.0), (2000.0, 0.0))
|
||||
|
||||
|
||||
def test_two_lines_crossing_without_an_end_do_not_connect():
|
||||
result = net(line("1", (0.0, 0.0), (2000.0, 0.0)), line("2", (1000.0, -500.0), (1000.0, 500.0)))
|
||||
assert len(result["runs"]) == 2
|
||||
|
||||
|
||||
# -- Review Focus 4: double-drawn pipe counts once ----------------------------
|
||||
|
||||
|
||||
def test_a_line_drawn_twice_is_counted_once():
|
||||
result = net(line("1", (0.0, 0.0), (2000.0, 0.0)), line("2", (0.0, 0.0), (2000.0, 0.0)))
|
||||
(run,) = result["runs"]
|
||||
assert abs(total(run) - 2000.0) < EPS # not 4000
|
||||
assert abs(result["stats"]["overlap_length"] - 2000.0) < EPS
|
||||
assert run.edges[0]["handles"] == ("1", "2")
|
||||
|
||||
|
||||
def test_a_line_drawn_over_a_polyline_is_counted_once():
|
||||
poly = rec("P", "LWPOLYLINE", PIPE, ((0.0, 0.0), (1000.0, 0.0), (3000.0, 0.0)))
|
||||
result = net(line("L", (0.0, 0.0), (3000.0, 0.0)), poly)
|
||||
(run,) = result["runs"]
|
||||
# the line splits at the polyline's vertex (1000, 0): pieces 1000 + 2000, each once
|
||||
assert abs(total(run) - 3000.0) < EPS
|
||||
assert abs(result["stats"]["overlap_length"] - 3000.0) < EPS
|
||||
assert all(edge["handles"] == ("L", "P") for edge in run.edges)
|
||||
|
||||
|
||||
def test_a_partial_collinear_overlap_counts_the_shared_stretch_once():
|
||||
result = net(line("1", (0.0, 0.0), (2000.0, 0.0)), line("2", (1500.0, 0.0), (3000.0, 0.0)))
|
||||
(run,) = result["runs"]
|
||||
assert abs(total(run) - 3000.0) < EPS # 0..3000 once; 1500..2000 was drawn twice
|
||||
assert abs(result["stats"]["overlap_length"] - 500.0) < EPS
|
||||
|
||||
|
||||
def test_an_arc_keeps_its_true_length():
|
||||
# a quarter circle of radius 1000 centred on the origin, from 0 deg to 90 deg
|
||||
arc = rec("A", "ARC", PIPE, ((0.0, 0.0),), radius=1000.0, angles=(0.0, 90.0))
|
||||
(run,) = net(arc)["runs"]
|
||||
assert abs(total(run) - 500.0 * 3.141592653589793) < 1e-6 # 2*pi*1000/4
|
||||
ends = [c for point in run.ends for c in point]
|
||||
assert ends == pytest.approx([0.0, 1000.0, 1000.0, 0.0], abs=1e-9)
|
||||
|
||||
|
||||
# -- fittings ----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_an_inline_valve_joins_the_two_lines_and_adds_no_length():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1100.0, 0.0), (2000.0, 0.0)),
|
||||
insert("V", "GATE_VALVE", (1000.0, -50.0, 1100.0, 50.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
assert abs(total(run) - 1900.0) < EPS # 1000 + 900; the valve body is not pipe
|
||||
assert run.ends == ((0.0, 0.0), (2000.0, 0.0))
|
||||
assert result["stats"]["fittings"] == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("block", "reducer"),
|
||||
[
|
||||
("REDUCER_51_38", True),
|
||||
("ПЕРЕХОД_51", True),
|
||||
("REDÜKSİYON-51", True),
|
||||
("verloop 51x38", True),
|
||||
("REDUZIERSTÜCK_51", True),
|
||||
("GATE_VALVE", False),
|
||||
(None, False),
|
||||
],
|
||||
)
|
||||
def test_reducer_words_in_four_languages(block, reducer):
|
||||
assert is_reducer(block) is reducer
|
||||
|
||||
|
||||
def test_reducer_words_agree_with_the_vocabulary():
|
||||
# one reducer vocabulary: every block equipment_kind calls a reducer is one here
|
||||
for block in ("REDUCER_51_38", "ПЕРЕХОД_51", "REDÜKSİYON-51", "VERLOOPSTUK", "REDUZIERSTÜCK"):
|
||||
assert equipment_kind(block)["kind"] == "reducer"
|
||||
assert is_reducer(block), block
|
||||
|
||||
|
||||
def test_an_insert_holding_more_than_four_free_ends_is_not_a_fitting():
|
||||
# a title-block sized INSERT over four separate lines: eight free ends inside
|
||||
result = net(
|
||||
*[line(str(n), (0.0, 100.0 * n), (500.0, 100.0 * n)) for n in range(4)],
|
||||
insert("TB", "TITLE", (-10.0, -10.0, 510.0, 310.0)),
|
||||
)
|
||||
assert len(result["runs"]) == 4
|
||||
assert result["stats"]["fittings"] == 0
|
||||
assert result["stats"]["fittings_skipped"] == ["TB"]
|
||||
|
||||
|
||||
# -- diameters ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_a_label_is_direct_and_carries_by_continuity_across_a_tee():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1000.0, 0.0), (2000.0, 0.0)),
|
||||
line("3", (1000.0, 0.0), (1000.0, 800.0)),
|
||||
text("D", "Ø51", (400.0, 5.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
by_handle = {edge["handles"]: edge for edge in run.edges}
|
||||
assert by_handle[("1",)]["diameter_source"] == "direct"
|
||||
assert by_handle[("1",)]["label"] == "D"
|
||||
for handle in (("2",), ("3",)):
|
||||
assert by_handle[handle]["diameter"] == "Ø51"
|
||||
assert by_handle[handle]["diameter_source"] == "continuity"
|
||||
assert result["unassigned"] == []
|
||||
|
||||
|
||||
def test_a_label_holds_for_every_piece_of_its_drawn_entity():
|
||||
poly = rec("P", "LWPOLYLINE", PIPE, ((0.0, 0.0), (1000.0, 0.0), (1000.0, 1000.0)))
|
||||
(run,) = net(poly, text("D", "DN20", (400.0, 5.0)))["runs"]
|
||||
assert [edge["diameter_source"] for edge in run.edges] == ["direct", "direct"]
|
||||
assert {edge["diameter"] for edge in run.edges} == {"DN20"}
|
||||
|
||||
|
||||
def test_continuity_stops_at_a_reducer():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1060.0, 0.0), (2000.0, 0.0)),
|
||||
insert("R", "REDUCER_51_38", (1000.0, -30.0, 1060.0, 30.0)),
|
||||
text("D", "Ø51", (400.0, 5.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
by_handle = {edge["handles"]: edge for edge in run.edges}
|
||||
assert by_handle[("1",)]["diameter"] == "Ø51"
|
||||
assert by_handle[("2",)]["diameter"] is None
|
||||
assert by_handle[("2",)]["diameter_source"] == "unassigned"
|
||||
assert result["unassigned"][0]["reason"].startswith("no diameter label")
|
||||
assert result["stats"]["reducers"] == 1
|
||||
|
||||
|
||||
def test_continuity_stops_at_a_reducer_the_line_is_broken_at():
|
||||
# BREAK-at-point: two LINEs share a vertex inside the reducer's box, no free end there
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1030.0, 0.0)),
|
||||
line("2", (1030.0, 0.0), (2000.0, 0.0)),
|
||||
insert("R", "REDUCER_51_38", (1000.0, -30.0, 1060.0, 30.0)),
|
||||
text("D", "Ø51", (400.0, 5.0)),
|
||||
)
|
||||
by_handle = {edge["handles"]: edge for edge in result["runs"][0].edges}
|
||||
assert by_handle[("1",)]["diameter"] == "Ø51"
|
||||
assert by_handle[("2",)]["diameter_source"] == "unassigned"
|
||||
assert (result["stats"]["fittings"], result["stats"]["reducers"]) == (1, 1)
|
||||
|
||||
|
||||
def test_a_polyline_drawn_through_a_reducer_is_cut_there():
|
||||
poly = rec("P", "LWPOLYLINE", PIPE, ((0.0, 0.0), (2000.0, 0.0)))
|
||||
result = net(
|
||||
poly,
|
||||
line("BR", (1500.0, 0.0), (1500.0, 800.0)),
|
||||
insert("R", "REDUCER_51_38", (1000.0, -30.0, 1060.0, 30.0)),
|
||||
text("D", "Ø51", (400.0, 5.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
# cut at the foot of the box centre (1030, 0): 1030 + 470 + 500 + 800 branch
|
||||
assert abs(total(run) - 2800.0) < EPS
|
||||
by_span = {(edge["a"][0], edge["b"][0], edge["handles"]): edge for edge in run.edges}
|
||||
labelled = by_span[(0.0, 1030.0, ("P",))]
|
||||
assert (labelled["diameter"], labelled["diameter_source"]) == ("Ø51", "direct")
|
||||
# the label holds for the polyline only up to the reducer; nothing crosses it
|
||||
others = [edge for edge in run.edges if edge is not labelled]
|
||||
assert len(others) == 3
|
||||
assert {edge["diameter_source"] for edge in others} == {"unassigned"}
|
||||
assert result["stats"]["reducers"] == 1
|
||||
|
||||
|
||||
def test_a_pipe_running_beside_a_reducer_is_not_cut():
|
||||
# a parallel pipe grazing the reducer's box edge does not pass through its middle
|
||||
result = net(
|
||||
line("1", (0.0, 30.0), (2000.0, 30.0)),
|
||||
insert("R", "REDUCER_51_38", (1000.0, -30.0, 1060.0, 30.0)),
|
||||
text("D", "Ø51", (400.0, 35.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
assert len(run.edges) == 1
|
||||
assert run.edges[0]["diameter_source"] == "direct"
|
||||
assert result["stats"]["reducers"] == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("order", ["ABC", "BAC", "CAB", "ACB", "BCA", "CBA"])
|
||||
def test_a_three_way_valve_guesses_nothing_whatever_the_record_order(order):
|
||||
# C leaves the valve 1000 from A (Ø51) and 1000 from B (DN20): equally near
|
||||
records = {
|
||||
"A": line("A", (0.0, 0.0), (1000.0, 0.0)),
|
||||
"B": line("B", (1100.0, 0.0), (2100.0, 0.0)),
|
||||
"C": line("C", (1050.0, 50.0), (1050.0, 1050.0)),
|
||||
}
|
||||
result = net(
|
||||
*[records[name] for name in order],
|
||||
insert("V", "3WAY_VALVE", (1000.0, -50.0, 1100.0, 50.0)),
|
||||
text("TA", "Ø51", (400.0, 5.0)),
|
||||
text("TB", "DN20", (1500.0, 5.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
c = next(edge for edge in run.edges if edge["handles"] == ("C",))
|
||||
assert (c["diameter"], c["diameter_source"]) == (None, "unassigned")
|
||||
assert [u["reason"] for u in result["unassigned"]] == ["two different diameters equally near"]
|
||||
assert run.ends == ((0.0, 0.0), (1050.0, 1050.0), (2100.0, 0.0))
|
||||
|
||||
|
||||
def test_a_label_nearer_to_a_pipe_on_an_unread_layer_is_not_taken():
|
||||
# the label stands 5 below the WATER line and 22.5 above PRODUCT (search 25)
|
||||
drawing = snap(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0), PIPE),
|
||||
line("W", (0.0, 30.0), (1000.0, 30.0), "WATER"),
|
||||
text("D", "DN20", (400.0, 20.0)),
|
||||
)
|
||||
assert "WATER" not in default_layers(drawing)
|
||||
result = build_network(drawing)
|
||||
assert result["runs"][0].edges[0]["diameter_source"] == "unassigned"
|
||||
assert result["stats"]["labels_skipped"] == [
|
||||
{
|
||||
"handle": "D",
|
||||
"text": "DN20",
|
||||
"reason": "nearer to a line on layer WATER, which is not read",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_an_equipment_outline_does_not_take_a_pipe_label():
|
||||
# closed outlines are shapes, not pipes: a label beside the pipe end stays the pipe's
|
||||
result = net(
|
||||
tank("A", "T4100", (-500.0, -500.0, 0.0, 500.0)),
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
text("D", "Ø51", (2.0, 5.0)),
|
||||
)
|
||||
edge = result["runs"][0].edges[0]
|
||||
assert (edge["diameter"], edge["diameter_source"]) == ("Ø51", "direct")
|
||||
|
||||
|
||||
def test_continuity_crosses_a_valve():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1100.0, 0.0), (2000.0, 0.0)),
|
||||
insert("V", "GATE_VALVE", (1000.0, -50.0, 1100.0, 50.0)),
|
||||
text("D", "Ø51", (400.0, 5.0)),
|
||||
)
|
||||
edge = next(e for e in result["runs"][0].edges if e["handles"] == ("2",))
|
||||
assert (edge["diameter"], edge["diameter_source"]) == ("Ø51", "continuity")
|
||||
|
||||
|
||||
def test_a_conflicting_label_stops_continuity_and_the_nearer_one_wins():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1000.0, 0.0), (2000.0, 0.0)),
|
||||
line("3", (2000.0, 0.0), (3000.0, 0.0)),
|
||||
text("A", "Ø51", (400.0, 5.0)),
|
||||
text("B", "DN20", (1400.0, 5.0)),
|
||||
)
|
||||
by_handle = {edge["handles"]: edge for edge in result["runs"][0].edges}
|
||||
assert by_handle[("1",)]["diameter"] == "Ø51"
|
||||
assert by_handle[("2",)]["diameter"] == "DN20"
|
||||
# piece 3 is next to piece 2 (DN20), one piece further from piece 1 (Ø51)
|
||||
assert (by_handle[("3",)]["diameter"], by_handle[("3",)]["diameter_source"]) == (
|
||||
"DN20",
|
||||
"continuity",
|
||||
)
|
||||
|
||||
|
||||
def test_two_diameters_equally_near_leave_the_piece_unassigned():
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)),
|
||||
line("2", (1000.0, 0.0), (2000.0, 0.0)),
|
||||
line("3", (2000.0, 0.0), (3000.0, 0.0)),
|
||||
text("A", "Ø51", (400.0, 5.0)),
|
||||
text("B", "DN20", (2400.0, 5.0)),
|
||||
)
|
||||
middle = next(e for e in result["runs"][0].edges if e["handles"] == ("2",))
|
||||
assert middle["diameter_source"] == "unassigned"
|
||||
assert result["unassigned"][0]["reason"] == "two different diameters equally near"
|
||||
|
||||
|
||||
def test_labels_are_kept_exactly_as_drawn():
|
||||
result = net(line("1", (0.0, 0.0), (1000.0, 0.0)), text("D", "SMS51", (400.0, 5.0)))
|
||||
assert result["runs"][0].edges[0]["diameter"] == "SMS51"
|
||||
|
||||
|
||||
def test_a_label_beyond_the_search_distance_is_not_read():
|
||||
# median text height 5 -> label_search 25; the label's centre stands 200 away
|
||||
result = net(line("1", (0.0, 0.0), (1000.0, 0.0)), text("D", "Ø51", (400.0, 200.0)))
|
||||
assert result["stats"]["label_search"] == 25.0
|
||||
assert result["runs"][0].edges[0]["diameter_source"] == "unassigned"
|
||||
|
||||
|
||||
# -- services ----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_a_return_word_overrides_the_layer_for_the_whole_run():
|
||||
# the layer says supply; only the word says this run returns
|
||||
layer = "CIP_SUPPLY"
|
||||
base = classify_layer(layer)["service"]
|
||||
assert base in ("cip_supply", "cip_return"), "the vocabulary must file CIP_SUPPLY as CIP"
|
||||
records = [
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0), layer),
|
||||
line("2", (1000.0, 0.0), (1000.0, 1000.0), layer),
|
||||
text("W", "ВОЗВРАТ", (400.0, 5.0)),
|
||||
]
|
||||
result = build_network(snap(*records), layers=[layer])
|
||||
(run,) = result["runs"]
|
||||
assert run.service == "cip_return"
|
||||
assert {edge["service"] for edge in run.edges} == {"cip_return"}
|
||||
|
||||
|
||||
def test_a_cip_layer_naming_no_direction_takes_it_from_the_word():
|
||||
# classify_layer files a directionless CIP layer as `unknown` on purpose:
|
||||
# the layer is not trusted with the direction, so the word must decide it
|
||||
layer = "CIP_LINES"
|
||||
found = classify_layer(layer)
|
||||
assert (found["service"], found["keyword"]) == ("unknown", "cip")
|
||||
records = [
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0), layer),
|
||||
line("2", (1000.0, 0.0), (1000.0, 1000.0), layer),
|
||||
text("W", "ВОЗВРАТ", (400.0, 5.0)),
|
||||
]
|
||||
result = build_network(snap(*records), layers=[layer])
|
||||
(run,) = result["runs"]
|
||||
assert run.service == "cip_return"
|
||||
assert result["stats"]["service_conflicts"] == []
|
||||
|
||||
|
||||
def test_a_directionless_cip_layer_without_a_word_stays_unknown():
|
||||
(run,) = build_network(
|
||||
snap(line("1", (0.0, 0.0), (1000.0, 0.0), "CIP_LINES")), layers=["CIP_LINES"]
|
||||
)["runs"]
|
||||
assert run.service == "unknown"
|
||||
|
||||
|
||||
def test_without_a_word_the_run_keeps_its_layer_service():
|
||||
(run,) = net(line("1", (0.0, 0.0), (1000.0, 0.0)))["runs"]
|
||||
assert run.service == classify_layer(PIPE)["service"]
|
||||
|
||||
|
||||
# -- equipment ---------------------------------------------------------------
|
||||
|
||||
|
||||
def test_run_ends_attach_to_the_tags_whose_geometry_holds_them():
|
||||
result = net(
|
||||
tank("A", "T4100", (-500.0, -500.0, 0.0, 500.0)),
|
||||
tank("B", "T4101", (2000.0, -500.0, 2500.0, 500.0)),
|
||||
line("1", (0.0, 0.0), (2000.0, 0.0)),
|
||||
)
|
||||
(run,) = result["runs"]
|
||||
assert run.tags == ("T4100", "T4101")
|
||||
assert result["attachments"]["R1"] == [
|
||||
{"at": (0.0, 0.0), "tag": "T4100"},
|
||||
{"at": (2000.0, 0.0), "tag": "T4101"},
|
||||
]
|
||||
|
||||
|
||||
def test_a_tagged_insert_is_equipment_not_a_fitting():
|
||||
pump = rec(
|
||||
"PU",
|
||||
"INSERT",
|
||||
"EQUIPMENT",
|
||||
((1050.0, 0.0),),
|
||||
block="*U12",
|
||||
attribs=(("TAG", "M82"),),
|
||||
bbox=(1000.0, -50.0, 1100.0, 50.0),
|
||||
)
|
||||
result = net(
|
||||
line("1", (0.0, 0.0), (1000.0, 0.0)), line("2", (1100.0, 0.0), (2000.0, 0.0)), pump
|
||||
)
|
||||
assert len(result["runs"]) == 2
|
||||
assert all("M82" in run.tags for run in result["runs"])
|
||||
assert result["stats"]["fittings"] == 0
|
||||
|
||||
|
||||
def test_a_room_outline_around_two_tags_is_not_equipment():
|
||||
room = rec(
|
||||
"ROOM",
|
||||
"LWPOLYLINE",
|
||||
"ROOMS",
|
||||
((-5000.0, -5000.0), (9000.0, -5000.0), (9000.0, 5000.0), (-5000.0, 5000.0)),
|
||||
closed=True,
|
||||
bbox=(-5000.0, -5000.0, 9000.0, 5000.0),
|
||||
)
|
||||
outline = rec(
|
||||
"A",
|
||||
"LWPOLYLINE",
|
||||
"EQUIPMENT",
|
||||
((-500.0, -500.0), (0.0, -500.0), (0.0, 500.0), (-500.0, 500.0)),
|
||||
closed=True,
|
||||
bbox=(-500.0, -500.0, 0.0, 500.0),
|
||||
)
|
||||
# T4100 is written 30 above its tank, inside the room: the room holds T4100
|
||||
# and T4101, so it is not equipment, and the tank is the nearest shape
|
||||
# within 2 x label_search = 50
|
||||
result = net(
|
||||
room,
|
||||
outline,
|
||||
text("AT", "T4100", (-400.0, 530.0)),
|
||||
tank("B", "T4101", (2000.0, -500.0, 2500.0, 500.0)),
|
||||
line("1", (0.0, 0.0), (2000.0, 0.0)),
|
||||
)
|
||||
assert result["equipment"]["T4100"][0]["handle"] == "A"
|
||||
assert result["runs"][0].tags == ("T4100", "T4101")
|
||||
|
||||
|
||||
def test_a_shape_far_larger_than_the_others_is_still_found():
|
||||
# the grid keeps a box covering many cells aside; it must still be found
|
||||
small = [
|
||||
rec(
|
||||
f"C{n}",
|
||||
"CIRCLE",
|
||||
"EQUIPMENT",
|
||||
((5000.0 + 300.0 * n, 5000.0),),
|
||||
radius=50.0,
|
||||
bbox=(4950.0 + 300.0 * n, 4950.0, 5050.0 + 300.0 * n, 5050.0),
|
||||
)
|
||||
for n in range(5)
|
||||
]
|
||||
big = rec(
|
||||
"BIG",
|
||||
"LWPOLYLINE",
|
||||
"EQUIPMENT",
|
||||
((-500.0, -500.0), (-500.0 + 40000.0, -500.0), (39500.0, 39500.0), (-500.0, 39500.0)),
|
||||
closed=True,
|
||||
bbox=(-500.0, -500.0, 39500.0, 39500.0),
|
||||
)
|
||||
result = net(
|
||||
*small, big, text("AT", "T4100", (-400.0, 0.0)), line("1", (0.0, 0.0), (0.0, -2000.0))
|
||||
)
|
||||
assert result["equipment"]["T4100"][0]["handle"] == "BIG"
|
||||
assert result["runs"][0].tags == ("T4100",)
|
||||
|
||||
|
||||
def test_a_free_end_attaches_to_nothing():
|
||||
result = net(
|
||||
tank("A", "T4100", (-500.0, -500.0, 0.0, 500.0)), line("1", (0.0, 0.0), (2000.0, 0.0))
|
||||
)
|
||||
assert result["attachments"]["R1"][1] == {"at": (2000.0, 0.0), "tag": None}
|
||||
|
||||
|
||||
# -- layer choice and refusals -------------------------------------------------
|
||||
|
||||
|
||||
def test_default_layers_are_the_confidently_classified_network_layers():
|
||||
drawing = snap(line("1", (0.0, 0.0), (1.0, 0.0)), line("2", (0.0, 0.0), (1.0, 0.0), "MISC"))
|
||||
expected = tuple(
|
||||
name
|
||||
for name in ("MISC", PIPE)
|
||||
if classify_layer(name)["discipline"] in ("piping", "electrical")
|
||||
and classify_layer(name)["confidence"] >= 0.9
|
||||
)
|
||||
assert default_layers(drawing) == expected
|
||||
|
||||
|
||||
def test_label_search_defaults_to_five_median_text_heights():
|
||||
drawing = snap(
|
||||
text("a", "x", (0, 0), 2.0), text("b", "y", (0, 0), 4.0), text("c", "z", (0, 0), 9.0)
|
||||
)
|
||||
assert label_search_default(drawing) == 20.0 # 5 x median(2, 4, 9)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("kwargs", "message"),
|
||||
[
|
||||
({"tol": 0.0}, "tol: must be a finite number greater than zero"),
|
||||
({"tol": float("nan")}, "tol: must be a finite number greater than zero"),
|
||||
({"label_search": -1.0}, "label_search: must be a finite number greater than zero"),
|
||||
({"layers": ["NOPE"]}, "layers: NOPE not in the drawing; it has: PRODUCT"),
|
||||
({"layers": []}, "layers: an empty list reads nothing"),
|
||||
],
|
||||
)
|
||||
def test_refusals_name_the_parameter(kwargs, message):
|
||||
drawing = snap(line("1", (0.0, 0.0), (1.0, 0.0)))
|
||||
kwargs = {"layers": [PIPE], **kwargs}
|
||||
with pytest.raises(ValueError, match="^" + message.replace("(", r"\(")):
|
||||
build_network(drawing, **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"))
|
||||
return truth, read_snapshot(truth["pid"])
|
||||
|
||||
|
||||
def run_key(tags, service):
|
||||
return (tuple(sorted(tags)), service)
|
||||
|
||||
|
||||
def test_the_generator_names_each_run_by_its_tags_and_service(plant):
|
||||
truth, _pid = plant
|
||||
keys = [run_key(run["tags"], run["service"]) for run in truth["pipe_runs"]]
|
||||
assert len(keys) == len(set(keys))
|
||||
|
||||
|
||||
def test_every_synthetic_run_has_its_service_diameter_and_tags(plant):
|
||||
truth, pid = plant
|
||||
found: dict = {}
|
||||
for run in build_network(pid)["runs"]:
|
||||
found.setdefault(run_key(run.tags, run.service), []).append(run)
|
||||
for expected in truth["pipe_runs"]:
|
||||
key = run_key(expected["tags"], expected["service"])
|
||||
runs = found.get(key, [])
|
||||
assert len(runs) == 1, f"{key}: {len(runs)} runs carry these tags and this service"
|
||||
if expected["status"] == "C":
|
||||
continue
|
||||
edges = runs[0].edges
|
||||
assert expected["diameter"] in {edge["diameter"] for edge in edges}, key
|
||||
sources = {edge["diameter_source"] for edge in edges}
|
||||
if expected["status"] == "A":
|
||||
assert sources == {"direct"}, key
|
||||
else:
|
||||
assert sources & {"continuity", "unassigned"}, key
|
||||
|
||||
|
||||
def test_the_synthetic_service_layers_are_read_by_default(plant):
|
||||
truth, pid = plant
|
||||
layers = set(build_network(pid)["stats"]["layers"])
|
||||
service_layers = {
|
||||
layer
|
||||
for layer, service in truth["layer_services"].items()
|
||||
if service not in ("electrical", "unknown")
|
||||
}
|
||||
assert service_layers <= layers
|
||||
@@ -0,0 +1,253 @@
|
||||
"""The takeoff workbook: four sheets, three languages, CSV always, XLSX with openpyxl.
|
||||
|
||||
The result written is `pipe_rows` over a hand-built pair: a P&ID with two
|
||||
runs (T4100-T4101 labelled Ø51, T4101-T4102 unlabelled) and a layout whose
|
||||
INSUNITS says inches over millimetre geometry, measured on the layout (28 m
|
||||
and 12 m).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import engineering.understand.report as report
|
||||
from engineering.understand.network import build_network
|
||||
from engineering.understand.report import LANGS, SHEETS, write_workbook
|
||||
from engineering.understand.snapshot import EntityRecord, Snapshot
|
||||
from engineering.understand.takeoff import pipe_rows
|
||||
|
||||
PIPE = "PRODUCT"
|
||||
WALLS = "WALLS"
|
||||
SCHEMATIC = {
|
||||
"verdict": "schematic",
|
||||
"within_10pct": 0.25,
|
||||
"pair_count": 12,
|
||||
"median": 1.3,
|
||||
"iqr": 0.8,
|
||||
"factor": None,
|
||||
"matched": [],
|
||||
"ambiguous": [],
|
||||
}
|
||||
|
||||
|
||||
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):
|
||||
return [
|
||||
rec(
|
||||
handle,
|
||||
"LWPOLYLINE",
|
||||
"EQUIPMENT",
|
||||
((x0, y0), (x0 + 500.0, y0), (x0 + 500.0, y0 + 500.0), (x0, y0 + 500.0)),
|
||||
closed=True,
|
||||
bbox=(x0, y0, x0 + 500.0, y0 + 500.0),
|
||||
),
|
||||
text(handle + "T", tag, (x0 + 150.0, y0 + 225.0), 50.0),
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def result():
|
||||
pid = snap(
|
||||
[
|
||||
*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),
|
||||
],
|
||||
4,
|
||||
"pid.dxf",
|
||||
)
|
||||
layout = snap(
|
||||
[
|
||||
rec(
|
||||
"W",
|
||||
"LWPOLYLINE",
|
||||
WALLS,
|
||||
((0, 0), (40000, 0), (40000, 20000), (0, 20000)),
|
||||
closed=True,
|
||||
bbox=(0.0, 0.0, 40000.0, 20000.0),
|
||||
),
|
||||
rec("X", "LINE", WALLS, ((20000, 0), (20000, 20000))),
|
||||
text("R1", "ROOM 101", (5000.0, 10000.0), 250.0),
|
||||
text("R2", "ROOM 102", (30000.0, 10000.0), 250.0),
|
||||
text("T0", "T4100", (2000.0, 3000.0), 250.0),
|
||||
text("T1", "T4101", (30000.0, 3000.0), 250.0),
|
||||
text("T2", "T4102", (30000.0, 15000.0), 250.0),
|
||||
],
|
||||
1,
|
||||
"layout.dxf",
|
||||
)
|
||||
return pipe_rows(build_network(pid, layers=[PIPE]), pid, layout, scale=SCHEMATIC)
|
||||
|
||||
|
||||
METRAJ_HEADERS = {
|
||||
"tr": [
|
||||
"Mahal",
|
||||
"Hizmet",
|
||||
"Çap",
|
||||
"Doğrudan (m)",
|
||||
"Süreklilik (m)",
|
||||
"Atanmamış (m)",
|
||||
"Net (m)",
|
||||
"Pay oranı",
|
||||
"Pay (m)",
|
||||
"Paylı toplam (m)",
|
||||
"Şematik (m)",
|
||||
"Durum",
|
||||
"Hatlar",
|
||||
],
|
||||
"en": [
|
||||
"Room",
|
||||
"Service",
|
||||
"Diameter",
|
||||
"Direct (m)",
|
||||
"Continuity (m)",
|
||||
"Unassigned (m)",
|
||||
"Net (m)",
|
||||
"Allowance",
|
||||
"Allowance (m)",
|
||||
"Total incl. allowance (m)",
|
||||
"Schematic (m)",
|
||||
"Status",
|
||||
"Runs",
|
||||
],
|
||||
"ru": [
|
||||
"Помещение",
|
||||
"Среда",
|
||||
"Диаметр",
|
||||
"Прямо (м)",
|
||||
"По непрерывности (м)",
|
||||
"Не назначено (м)",
|
||||
"Нетто (м)",
|
||||
"Запас",
|
||||
"Запас (м)",
|
||||
"Итого с запасом (м)",
|
||||
"По схеме (м)",
|
||||
"Статус",
|
||||
"Линии",
|
||||
],
|
||||
}
|
||||
TOTAL = {"tr": "TOPLAM", "en": "TOTAL", "ru": "ИТОГО"}
|
||||
|
||||
|
||||
def read_csv(path):
|
||||
with open(path, encoding="utf-8-sig", newline="") as handle:
|
||||
return list(csv.reader(handle))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lang", LANGS)
|
||||
def test_csv_sheets_carry_the_headers_of_each_language(result, tmp_path, lang):
|
||||
files = write_workbook(result, kind="pipe", lang=lang, path=str(tmp_path / "takeoff.xlsx"))
|
||||
names = [Path(p).name for p in files["csv"]]
|
||||
assert names == [
|
||||
"takeoff_metraj.csv",
|
||||
"takeoff_ozet.csv",
|
||||
"takeoff_kontrol.csv",
|
||||
"takeoff_metodoloji.csv",
|
||||
]
|
||||
metraj = read_csv(files["csv"][0])
|
||||
assert metraj[0] == METRAJ_HEADERS[lang]
|
||||
total = metraj[-1]
|
||||
assert total[0] == TOTAL[lang]
|
||||
assert float(total[6]) == pytest.approx(40.0) # 28 m + 12 m net
|
||||
assert float(total[9]) == pytest.approx(48.0) # with the 20 % allowance
|
||||
|
||||
|
||||
def test_the_workbook_has_the_four_sheets_and_numbers_stay_numbers(result, tmp_path):
|
||||
openpyxl = pytest.importorskip("openpyxl")
|
||||
files = write_workbook(result, kind="pipe", lang="tr", path=str(tmp_path / "takeoff.xlsx"))
|
||||
assert files["refused"] is None
|
||||
book = openpyxl.load_workbook(files["xlsx"])
|
||||
assert book.sheetnames == list(SHEETS) == ["Metraj", "Özet", "Kontrol", "Metodoloji"]
|
||||
rows = list(book["Metraj"].iter_rows(values_only=True))
|
||||
assert list(rows[0]) == METRAJ_HEADERS["tr"]
|
||||
assert rows[-1][0] == "TOPLAM"
|
||||
assert rows[-1][6] == pytest.approx(40.0)
|
||||
assert isinstance(rows[1][6], (int, float)) # a number cell, not the text '18.0'
|
||||
assert rows[1][6] == pytest.approx(18.0) # room 101's share of the Ø51 run
|
||||
|
||||
|
||||
def test_the_method_sheet_is_generated_from_the_result(result, tmp_path):
|
||||
files = write_workbook(result, kind="pipe", lang="en", path=str(tmp_path / "t.xlsx"))
|
||||
method = dict((row[0], row[1]) for row in read_csv(files["csv"][3])[1:] if len(row) == 2)
|
||||
assert method["Length source"] == "layout"
|
||||
assert method["Scale check: verdict"] == "schematic"
|
||||
assert method["Scale check: pairs"] == "12"
|
||||
assert method["Scale verified"] == "yes"
|
||||
assert method["Units: layout"] == "INSUNITS in → mm"
|
||||
assert "x first, then y" in method["Rule: route"]
|
||||
warnings = [row[1] for row in read_csv(files["csv"][3]) if row[0] == "Warning"]
|
||||
assert any("INSUNITS declares in" in w for w in warnings)
|
||||
|
||||
|
||||
def test_kontrol_lists_the_unassigned_diameter(result, tmp_path):
|
||||
files = write_workbook(result, kind="pipe", lang="en", path=str(tmp_path / "t.xlsx"))
|
||||
kontrol = read_csv(files["csv"][2])
|
||||
assert kontrol[0] == ["Run", "Kind", "Service", "Tags", "Schematic (m)", "Reason"]
|
||||
(row,) = kontrol[1:]
|
||||
assert row[1] == "Diameter unassigned" and row[3] == "T4101, T4102"
|
||||
assert row[5] == "No diameter label"
|
||||
|
||||
|
||||
def test_without_openpyxl_the_csv_is_written_and_xlsx_is_refused(result, tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(report, "xlsx_available", lambda: False)
|
||||
files = write_workbook(result, kind="pipe", lang="tr", path=str(tmp_path / "takeoff.xlsx"))
|
||||
assert files["xlsx"] is None
|
||||
assert not (tmp_path / "takeoff.xlsx").exists()
|
||||
assert all(Path(p).exists() for p in files["csv"])
|
||||
assert files["refused"]["capability"] == "xlsx_write"
|
||||
assert files["refused"]["ok"] is False
|
||||
assert '".[office]"' in files["refused"]["error"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("kwargs", "message"),
|
||||
[
|
||||
({"kind": "steel"}, "kind: 'steel' unknown; choose from pipe"),
|
||||
({"lang": "de"}, "lang: 'de' unknown; choose from tr, en, ru"),
|
||||
({"path": "takeoff.ods"}, "path: the workbook is written as .xlsx"),
|
||||
],
|
||||
)
|
||||
def test_refusals_come_before_any_file(result, tmp_path, kwargs, message):
|
||||
arguments = {"kind": "pipe", "lang": "en", "path": str(tmp_path / "t.xlsx"), **kwargs}
|
||||
if "path" in kwargs:
|
||||
arguments["path"] = str(tmp_path / kwargs["path"])
|
||||
with pytest.raises(ValueError, match="^" + message.replace(".", r"\.")):
|
||||
write_workbook(result, **arguments)
|
||||
assert list(tmp_path.iterdir()) == []
|
||||
@@ -0,0 +1,48 @@
|
||||
"""Orthogonal lengths: Manhattan distance, the rectilinear MST, the x-then-y route.
|
||||
|
||||
Every expected value is worked by hand in the comment beside it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from engineering.understand.route import manhattan, rmst, route_legs
|
||||
|
||||
EPS = 1e-9
|
||||
|
||||
|
||||
def test_manhattan_is_the_sum_of_the_axis_distances():
|
||||
assert manhattan((0.0, 0.0), (3.0, -4.0)) == 7.0 # 3 + 4, not the diagonal 5
|
||||
|
||||
|
||||
def test_fewer_than_two_points_have_no_tree():
|
||||
assert rmst([]) == (0.0, ())
|
||||
assert rmst([(5.0, 5.0)]) == (0.0, ())
|
||||
|
||||
|
||||
def test_a_two_ended_run_is_dx_plus_dy():
|
||||
length, edges = rmst([(1000.0, 3000.0), (29000.0, 15000.0)])
|
||||
assert abs(length - 40000.0) < EPS # 28000 + 12000
|
||||
assert edges == ((0, 1),)
|
||||
|
||||
|
||||
def test_a_square_of_four_tags_is_three_sides():
|
||||
# Prim from point 0: (0,1) at 10 (tie with 3 goes to the lower index),
|
||||
# then (1,2) at 10, then (0,3) at 10 -> 30, never the 40 of the full ring
|
||||
length, edges = rmst([(0.0, 0.0), (10.0, 0.0), (10.0, 10.0), (0.0, 10.0)])
|
||||
assert abs(length - 30.0) < EPS
|
||||
assert edges == ((0, 1), (1, 2), (0, 3))
|
||||
|
||||
|
||||
def test_three_tags_in_a_row_chain_instead_of_a_star():
|
||||
length, edges = rmst([(0.0, 0.0), (100.0, 0.0), (200.0, 0.0)])
|
||||
assert abs(length - 200.0) < EPS # 100 + 100, not 100 + 200
|
||||
assert edges == ((0, 1), (1, 2))
|
||||
|
||||
|
||||
def test_the_route_runs_along_x_first_then_y():
|
||||
assert route_legs((0.0, 0.0), (30.0, 40.0)) == (
|
||||
((0.0, 0.0), (30.0, 0.0)),
|
||||
((30.0, 0.0), (30.0, 40.0)),
|
||||
)
|
||||
assert route_legs((0.0, 0.0), (0.0, 40.0)) == (((0.0, 0.0), (0.0, 40.0)),)
|
||||
assert route_legs((1.0, 1.0), (1.0, 1.0)) == ()
|
||||
@@ -0,0 +1,423 @@
|
||||
"""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"]
|
||||
@@ -0,0 +1,142 @@
|
||||
"""pipe_takeoff and cable_takeoff over the MCP wire, on the synthetic plant pair - never on a
|
||||
client drawing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastmcp import Client
|
||||
|
||||
import server
|
||||
from tests.fixtures.plant_pair import build_plant_pair
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
CSV_NAMES = [
|
||||
"takeoff_metraj.csv",
|
||||
"takeoff_ozet.csv",
|
||||
"takeoff_kontrol.csv",
|
||||
"takeoff_metodoloji.csv",
|
||||
]
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(monkeypatch):
|
||||
"""In-memory client on the headless engine - 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"
|
||||
yield connected
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def plant(tmp_path):
|
||||
return build_plant_pair(tmp_path)
|
||||
|
||||
|
||||
async def test_pipe_takeoff_measures_on_the_layout_and_writes_the_workbook(client, plant, tmp_path):
|
||||
output = tmp_path / "takeoff.xlsx"
|
||||
arguments = {
|
||||
"pid_path": plant["pid"],
|
||||
"layout_path": plant["layout"],
|
||||
"output": str(output),
|
||||
"lang": "tr",
|
||||
}
|
||||
result = (await client.call_tool("pipe_takeoff", arguments)).structured_content
|
||||
assert result["length_source"] == "layout"
|
||||
assert result["scale"]["verdict"] == plant["scale_verdict"]
|
||||
files = result["files"]
|
||||
assert [Path(p).name for p in files["csv"]] == CSV_NAMES
|
||||
assert all(Path(p).exists() for p in files["csv"])
|
||||
if files["xlsx"] is None:
|
||||
assert files["refused"]["capability"] == "xlsx_write"
|
||||
else:
|
||||
assert Path(files["xlsx"]) == output
|
||||
assert output.exists()
|
||||
|
||||
|
||||
async def test_pid_lengths_on_the_schematic_pid_are_refused_without_force(client, plant):
|
||||
arguments = {"pid_path": plant["pid"], "layout_path": plant["layout"], "length_source": "pid"}
|
||||
result = await client.call_tool("pipe_takeoff", arguments, raise_on_error=False)
|
||||
assert result.is_error is True
|
||||
text = result.content[0].text
|
||||
assert "the scale check calls this P&ID schematic" in text
|
||||
assert "force=True" in text
|
||||
|
||||
|
||||
async def test_force_measures_on_the_pid_and_marks_it_unverified(client, plant):
|
||||
arguments = {
|
||||
"pid_path": plant["pid"],
|
||||
"layout_path": plant["layout"],
|
||||
"length_source": "pid",
|
||||
"force": True,
|
||||
}
|
||||
result = (await client.call_tool("pipe_takeoff", arguments)).structured_content
|
||||
assert (result["length_source"], result["scale_verified"]) == ("pid", False)
|
||||
assert result["files"] is None # no output named, nothing written
|
||||
|
||||
|
||||
async def test_an_unknown_language_is_refused_before_anything_is_read(client, tmp_path):
|
||||
arguments = {"pid_path": str(tmp_path / "never_read.dxf"), "lang": "de"}
|
||||
result = await client.call_tool("pipe_takeoff", arguments, raise_on_error=False)
|
||||
assert result.is_error is True
|
||||
assert "lang: 'de' unknown; choose from tr, en, ru" in result.content[0].text
|
||||
|
||||
|
||||
async def test_the_drawings_are_never_modified(client, plant, tmp_path):
|
||||
before = {key: Path(plant[key]).read_bytes() for key in ("pid", "layout")}
|
||||
arguments = {
|
||||
"pid_path": plant["pid"],
|
||||
"layout_path": plant["layout"],
|
||||
"output": str(tmp_path / "t.xlsx"),
|
||||
}
|
||||
await client.call_tool("pipe_takeoff", arguments)
|
||||
assert {key: Path(plant[key]).read_bytes() for key in ("pid", "layout")} == before
|
||||
|
||||
|
||||
async def test_the_tool_says_it_writes_a_file_and_destroys_nothing(client):
|
||||
tools = {tool.name: tool for tool in await client.list_tools()}
|
||||
annotations = tools["pipe_takeoff"].annotations
|
||||
assert annotations.readOnlyHint is False
|
||||
assert annotations.destructiveHint is False
|
||||
|
||||
|
||||
async def test_cable_takeoff_gives_every_synthetic_load_its_rounded_cable(client, plant, tmp_path):
|
||||
arguments = {
|
||||
"pid_path": plant["pid"],
|
||||
"layout_path": plant["layout"],
|
||||
"output": str(tmp_path / "cables.xlsx"),
|
||||
"lang": "ru",
|
||||
}
|
||||
result = (await client.call_tool("cable_takeoff", arguments)).structured_content
|
||||
rows = {row["tag"]: row for row in result["rows"]}
|
||||
for expected in plant["cable_rows"]:
|
||||
assert rows[expected["tag"]]["cable_m"] == expected["roundup_m"], expected["tag"]
|
||||
assert Path(result["files"]["csv"][0]).name == "cables_metraj.csv"
|
||||
|
||||
|
||||
async def test_cable_takeoff_without_a_layout_leaves_every_length_empty(client, plant):
|
||||
result = (
|
||||
await client.call_tool("cable_takeoff", {"pid_path": plant["pid"]})
|
||||
).structured_content
|
||||
assert result["rows"], "the synthetic P&ID carries electrical loads"
|
||||
assert all(row["cable_m"] is None for row in result["rows"])
|
||||
no_layout = {item["tag"] for item in result["open_items"] if item["item"] == "no_layout"}
|
||||
assert no_layout == {row["tag"] for row in result["rows"]}
|
||||
|
||||
|
||||
async def test_a_malformed_section_rule_is_refused_by_path(client, plant):
|
||||
arguments = {"pid_path": plant["pid"], "section_rules": [{"max_kw": 5}]}
|
||||
result = await client.call_tool("cable_takeoff", arguments, raise_on_error=False)
|
||||
assert result.is_error is True
|
||||
assert "section_rules[0].section: expected a non-empty text" in result.content[0].text
|
||||
|
||||
|
||||
async def test_the_cable_tool_says_it_writes_a_file_and_destroys_nothing(client):
|
||||
tools = {tool.name: tool for tool in await client.list_tools()}
|
||||
annotations = tools["cable_takeoff"].annotations
|
||||
assert annotations.readOnlyHint is False
|
||||
assert annotations.destructiveHint is False
|
||||
Reference in New Issue
Block a user