diff --git a/README.md b/README.md
index b33ebea..303ea23 100644
--- a/README.md
+++ b/README.md
@@ -27,8 +27,8 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
-> **v1.5 release snapshot:** 242 tools · 8 resources · 5 prompt templates · 4516 collected tests.
-> 242 is the **registered** count; a default install advertises 237 over `tools/list`,
+> **v1.5 release snapshot:** 243 tools · 8 resources · 5 prompt templates · 4516 collected tests.
+> 243 is the **registered** count; a default install advertises 238 over `tools/list`,
> because `ENABLE_3D` is unset. `system_about` is the runtime authority.
## Why this exists
@@ -122,7 +122,7 @@ Claude Desktop, Cursor, or any stdio MCP host. For HTTP: `autocad-mcp --transpor
| Quality loop | `drawing_preflight` → `drawing_plan` → `drawing_critique` → `drawing_refine` → `drawing_finalize` (0–100 score) |
| Delivery | `drawing_deliver`: DXF/PDF/PNG + SHA-256 manifest + reopen-parity checks |
-242 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates.
+243 tools in 26 groups; TOOL_PACKS=core hides the nine pid_* tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (TOOL_PACKS=core,mech keeps the mechanical ones, TOOL_PACKS=core,arch the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (autocad://drawing/info, layers, blocks, entities/stats, entities/{layer_name}, system/status, pid/symbols, standards/isa51) and 5 prompt templates.
**Two rules worth knowing.** Every coordinate in and out of a tool is WCS on both engines — the one exception is TEXT `rotation`, which stays in the entity frame because a mirrored TEXT is mirror-imaged and no scalar angle expresses that. And never read vertices back and shoelace them: that loses **28.2%** of the area on a semicircular edge, silently. `analysis_measure_entity(handle)` reads the real geometry and states its own accuracy.
diff --git a/discovery/aliases.py b/discovery/aliases.py
index 700b8ef..ec5bd36 100644
--- a/discovery/aliases.py
+++ b/discovery/aliases.py
@@ -2487,6 +2487,23 @@ TOOL_ALIASES: dict[str, ToolAliases] = {
"ведомость трубопроводов",
),
),
+ "cable_takeoff": ToolAliases(
+ acad=(),
+ synonyms=(
+ "cable takeoff",
+ "cable schedule",
+ "cable lengths",
+ "cable list",
+ "how much cable",
+ "motor cable lengths",
+ "kablo metrajı",
+ "kablo metraji",
+ "kablo listesi",
+ "kablo cetveli",
+ "кабельный журнал",
+ "ведомость кабелей",
+ ),
+ ),
}
diff --git a/docs/tool-inventory.json b/docs/tool-inventory.json
index 7890aff..10b5410 100644
--- a/docs/tool-inventory.json
+++ b/docs/tool-inventory.json
@@ -3,13 +3,13 @@
"schema_version": 1,
"package_version": "1.5.1",
"totals": {
- "tools": 242,
+ "tools": 243,
"resources": 8,
"prompts": 5,
"groups": 26
},
"groups": {
- "analysis": 13,
+ "analysis": 14,
"architecture": 12,
"batch": 3,
"blocks": 9,
@@ -543,6 +543,21 @@
"destructiveHint": false
}
},
+ {
+ "name": "cable_takeoff",
+ "group": "analysis",
+ "cost": "safe",
+ "summary": "Cable takeoff: one row per load, Manhattan length to its panel, rounded up to the metre.",
+ "tags": [
+ "analysis",
+ "plant"
+ ],
+ "annotations": {
+ "title": "Plant: Cable Takeoff",
+ "readOnlyHint": false,
+ "destructiveHint": false
+ }
+ },
{
"name": "cad_batch",
"group": "batch",
diff --git a/engineering/understand/draw.py b/engineering/understand/draw.py
index 2882ec9..c1f3dbd 100644
--- a/engineering/understand/draw.py
+++ b/engineering/understand/draw.py
@@ -15,9 +15,15 @@ import math
from engineering.understand.network import build_network
from engineering.understand.report import check_lang, write_workbook
from engineering.understand.snapshot import take_snapshot
-from engineering.understand.takeoff import check_pipe_options, pipe_rows
+from engineering.understand.takeoff import (
+ cable_rows,
+ check_allowance,
+ check_pipe_options,
+ check_section_rules,
+ pipe_rows,
+)
-__all__ = ["run_pipe_takeoff"]
+__all__ = ["run_cable_takeoff", "run_pipe_takeoff"]
def _check_tol(tol) -> float:
@@ -84,3 +90,30 @@ async def run_pipe_takeoff(
else None
)
return result
+
+
+async def run_cable_takeoff(
+ backend,
+ *,
+ pid_path: str | None = None,
+ layout_path: str | None = None,
+ output: str | None = None,
+ lang: str = "en",
+ allowance: float = 0.20,
+ section_rules=None,
+) -> dict:
+ """`cable_rows` over a P&ID and an optional layout, plus the workbook when `output` is named."""
+ check_lang(lang)
+ allowance = check_allowance(allowance)
+ rules = check_section_rules(section_rules)
+ pid = await take_snapshot(backend, pid_path)
+ layout = await take_snapshot(backend, layout_path) if layout_path else None
+ result = await asyncio.to_thread(
+ cable_rows, pid, layout, allowance=allowance, section_rules=list(rules) or None
+ )
+ result["files"] = (
+ await asyncio.to_thread(write_workbook, result, kind="cable", lang=lang, path=output)
+ if output
+ else None
+ )
+ return result
diff --git a/engineering/understand/report.py b/engineering/understand/report.py
index b96c220..59d210b 100644
--- a/engineering/understand/report.py
+++ b/engineering/understand/report.py
@@ -158,6 +158,47 @@ LABELS: dict[str, tuple[str, str, str]] = {
),
"services": ("Kapsanan hizmetler", "Services taken off", "Учтённые среды"),
"warning": ("Uyarı", "Warning", "Предупреждение"),
+ # cable takeoff
+ "tag": ("Etiket", "Tag", "Тег"),
+ "kw": ("Güç (kW)", "Power (kW)", "Мощность (кВт)"),
+ "phases": ("Faz", "Phases", "Фазы"),
+ "voltage": ("Gerilim (V)", "Voltage (V)", "Напряжение (В)"),
+ "neutral": ("Nötr", "Neutral", "Нейтраль"),
+ "vfd": ("Sürücü (VFD)", "VFD", "ЧРП"),
+ "panel": ("Pano", "Panel", "Шкаф"),
+ "length_m": ("Mesafe (m)", "Distance (m)", "Расстояние (м)"),
+ "cable_m": ("Kablo (m)", "Cable (m)", "Кабель (м)"),
+ "section": ("Kesit", "Section", "Сечение"),
+ "package": ("Paket", "Package", "Комплект"),
+ "loads": ("Yük sayısı", "Loads", "Нагрузок"),
+ "item": ("Madde", "Item", "Пункт"),
+ "detail": ("Ayrıntı", "Detail", "Подробности"),
+ "by_panel": ("Panoya göre", "By panel", "По шкафу"),
+ "missing_power": ("Güç yazılmamış", "No power stated", "Мощность не указана"),
+ "missing_panel": ("Pano belirtilmemiş", "No panel named", "Шкаф не указан"),
+ "no_layout": ("Yerleşim yok", "No layout", "Нет планировки"),
+ "panel_not_on_layout": (
+ "Pano yerleşimde yok",
+ "Panel not on the layout",
+ "Шкафа нет на планировке",
+ ),
+ "no_section_rule": ("Kesit kuralı yok", "No section rule covers it", "Нет правила сечения"),
+ "conflicting_power": (
+ "Çelişen güç yazıları",
+ "Conflicting power texts",
+ "Противоречивая мощность",
+ ),
+ "rule_cable_route": ("Kural: kablo güzergâhı", "Rule: cable route", "Правило: трасса кабеля"),
+ "rule_rounding": ("Kural: yuvarlama", "Rule: rounding", "Правило: округление"),
+ "rule_power": ("Kural: güç", "Rule: power", "Правило: мощность"),
+ "rule_panel": ("Kural: pano", "Rule: panel", "Правило: шкаф"),
+ "rule_package": ("Kural: pano paketleri", "Rule: panel packages", "Правило: комплектные шкафы"),
+ "rule_section": ("Kural: kesit", "Rule: section", "Правило: сечение"),
+ "section_rules": (
+ "Kesit kuralları (çağıranın)",
+ "Section rules (the caller's)",
+ "Правила сечений (вызывающего)",
+ ),
}
#: The method sheet's sentences, as (tr, en, ru).
@@ -244,6 +285,55 @@ TEXTS: dict[str, tuple[str, str, str]] = {
"Линия с несколькими диаметрами делит длину между ними пропорционально их длине на "
"схеме P&ID.",
),
+ "rule_cable_route": (
+ "Uzunluk = yerleşimde yükün etiketinden panosunun etiketine Manhattan mesafesi.",
+ "Length = the Manhattan distance on the layout from the load's tag to its panel's tag.",
+ "Длина = манхэттенское расстояние на планировке от тега нагрузки до тега её шкафа.",
+ ),
+ "rule_rounding": (
+ "Kablo = YUKARIYUVARLA(uzunluk × (1 + pay)) tam metreye; uzunluk önce 1e-6 m'ye "
+ "yuvarlanır.",
+ "Cable = ROUNDUP(length × (1 + allowance)) to the whole metre; the length is rounded to "
+ "1e-6 m first.",
+ "Кабель = ОКРВВЕРХ(длина × (1 + запас)) до целого метра; длина сначала округляется до "
+ "1e-6 м.",
+ ),
+ "rule_power": (
+ "Güç, P&ID'de her etiketin yakınındaki elektrik yazılarından okunur; gücü yazılmamış "
+ "yükün hücresi boş kalır ve açık madde olur - güç asla uydurulmaz.",
+ "Power is read from the P&ID's electrical texts near each tag; a load with no stated "
+ "power keeps an empty cell and an open item - power is never invented.",
+ "Мощность берётся из электрических надписей P&ID рядом с каждым тегом; у нагрузки без "
+ "указанной мощности ячейка пуста и заводится открытый вопрос — мощность никогда не "
+ "выдумывается.",
+ ),
+ "rule_panel": (
+ "Pano, yerleşimdeki bağlantı notlarından ('Wiring to CP1' gibi), yoksa P&ID'den alınır; "
+ "adlar normalleştirilir (CP 1 → CP-1).",
+ "The panel comes from the layout's wiring callouts ('Wiring to CP1' and its variants), "
+ "else from the P&ID; names are normalised (CP 1 → CP-1).",
+ "Шкаф берётся из выносок подключения на планировке ('Wiring to CP1' и варианты), иначе "
+ "из P&ID; имена нормализуются (CP 1 → CP-1).",
+ ),
+ "rule_package": (
+ "Pano paketleri: kendisi güç belirten bir panoya bağlı yük o paketin parçasıdır; mahal ve "
+ "pano toplamları paketin gücünü sayar, alt yüklerini saymaz - hiçbir güç iki kez "
+ "sayılmaz.",
+ "Panel packages: a load wired to a panel that itself states a power is part of that "
+ "package; room and panel totals count the package's power and not its child loads, so "
+ "no power is counted twice.",
+ "Комплектные шкафы: нагрузка, подключённая к шкафу с собственной указанной мощностью, "
+ "входит в его комплект; итоги по помещениям и шкафам учитывают мощность комплекта, а не "
+ "его дочерних нагрузок, — ничего не учитывается дважды.",
+ ),
+ "rule_section": (
+ "Kablo kesitleri yalnız çağıranın section_rules tablosundan gelir; tablo yoksa sütun "
+ "boş kalır. Hiçbir standart tablo uygulanmaz.",
+ "Cable sections come only from the caller's section_rules; without them the column "
+ "stays blank. No standard table is applied.",
+ "Сечения кабелей — только из таблицы section_rules вызывающего; без неё столбец пуст. "
+ "Никакая стандартная таблица не применяется.",
+ ),
}
@@ -441,7 +531,106 @@ def _pipe_method(result: dict, lang: str) -> list[list]:
return [[_t(topic, lang), value] for topic, value in rows]
-_BUILDERS = {"pipe": _pipe_sheets}
+def _cable_sheets(result: dict, lang: str) -> dict[str, tuple[list, list]]:
+ columns = (
+ "tag",
+ "room",
+ "kw",
+ "phases",
+ "voltage",
+ "neutral",
+ "vfd",
+ "panel",
+ "length_m",
+ "allowance",
+ "cable_m",
+ "section",
+ "package",
+ )
+ rows = [
+ [
+ row["tag"],
+ row["room"],
+ row["kw"],
+ row["phases"],
+ row["voltage"],
+ _yes(row["neutral"], lang),
+ _yes(row["vfd"], lang),
+ row["panel"],
+ _num(row["length_m"]),
+ row["allowance"],
+ row["cable_m"],
+ row["section"],
+ row["package"],
+ ]
+ for row in result["rows"]
+ ]
+ totals = result["totals"]
+ rows.append(
+ [
+ _t("total", lang),
+ None,
+ _num(totals["known_kw"]),
+ None,
+ None,
+ None,
+ None,
+ None,
+ None,
+ None,
+ totals["cable_m"],
+ None,
+ None,
+ ]
+ )
+ summary = []
+ for group in ("by_room", "by_panel"):
+ for item in result["summary"][group]:
+ summary.append(
+ [_t(group, lang), item["key"], item["loads"], _num(item["kw"]), item["cable_m"]]
+ )
+ summary.append(
+ [_t("total", lang), None, totals["loads"], _num(totals["known_kw"]), totals["cable_m"]]
+ )
+ control = [
+ [item["tag"], _t(item["item"], lang), item["detail"]] for item in result["open_items"]
+ ]
+ return {
+ "Metraj": ([_t(c, lang) for c in columns], rows),
+ "Özet": ([_t(c, lang) for c in ("group", "key", "loads", "kw", "cable_m")], summary),
+ "Kontrol": ([_t(c, lang) for c in ("tag", "item", "detail")], control),
+ "Metodoloji": ([_t("topic", lang), _t("value", lang)], _cable_method(result, lang)),
+ }
+
+
+def _cable_method(result: dict, lang: str) -> list[list]:
+ method = result["method"]
+ rules = method["section_rules"]
+ rows = [
+ ["source_pid", method["sources"]["pid"]],
+ ["source_layout", method["sources"]["layout"] or "—"],
+ ["units_pid", _unit_text(result["units"]["pid"])],
+ ["units_layout", _unit_text(result["units"]["layout"])],
+ ["rule_cable_route", _say("rule_cable_route", lang)],
+ ["rule_allowance", method["allowance"]],
+ ["rule_rounding", _say("rule_rounding", lang)],
+ ["rule_power", _say("rule_power", lang)],
+ ["rule_panel", _say("rule_panel", lang)],
+ ["rule_package", _say("rule_package", lang)],
+ ["rule_section", _say("rule_section", lang)],
+ ["section_rules", "; ".join(_rule_text(rule) for rule in rules) or "—"],
+ ["label_search", round(float(method["search"]["pid"]), 6)],
+ ]
+ rows += [["warning", warning] for warning in result["warnings"]]
+ return [[_t(topic, lang), value] for topic, value in rows]
+
+
+def _rule_text(rule: dict) -> str:
+ phases = "" if rule["phases"] is None else f" / {rule['phases']} ph"
+ return f"<= {rule['max_kw']:g} kW{phases}: {rule['section']}"
+
+
+_BUILDERS = {"pipe": _pipe_sheets, "cable": _cable_sheets}
def write_workbook(result: dict, *, kind: str, lang: str, path: str) -> dict:
diff --git a/engineering/understand/takeoff.py b/engineering/understand/takeoff.py
index bf340f0..5d35b57 100644
--- a/engineering/understand/takeoff.py
+++ b/engineering/understand/takeoff.py
@@ -33,23 +33,34 @@ import statistics
from collections.abc import Iterable
from engineering.arch.faces import DEFAULT_TOL, Face, face_containing, planar_faces
-from engineering.understand.network import MODEL, TEXT_TYPES, tag_occurrences
-from engineering.understand.route import rmst, route_legs
+from engineering.understand.labels import normalize_panel, parse_electrical, wiring_target
+from engineering.understand.network import (
+ MODEL,
+ TEXT_TYPES,
+ label_search_default,
+ tag_occurrences,
+)
+from engineering.understand.route import manhattan, rmst, route_legs
from engineering.understand.snapshot import Pt, Snapshot
from engineering.understand.vocab import SERVICES, classify_layer, room_label
__all__ = [
+ "CABLE_SEARCH_SHARE",
"DEFAULT_SERVICES",
"LENGTH_SOURCES",
"UNIT_METRES",
"apportion",
+ "cable_rows",
"check_allowance",
"check_pipe_options",
+ "check_section_rules",
"layout_tags",
"pipe_rows",
"room_at",
"room_regions",
+ "roundup_m",
"scale_stand_in",
+ "section_for",
"unit_of",
]
@@ -832,3 +843,300 @@ def pipe_rows(
"rooms": _rooms_mode(regions),
},
}
+
+
+# -- cable rows ----------------------------------------------------------------
+
+#: A load's power block stands under its symbol rather than on a line, so its
+#: electrical texts and its wiring callout are searched twice as far as a
+#: diameter label: 10 x the median text height.
+CABLE_SEARCH_SHARE = 2.0
+SECTION_RULE_KEYS = ("max_kw", "section", "phases")
+
+
+def check_section_rules(section_rules) -> tuple[dict, ...]:
+ """The caller's own cable-section rule table, validated and sorted by max_kw.
+
+ Each rule is ``{"max_kw": > 0, "section": text, "phases": 1 | 3 (optional)}``;
+ a load takes the first rule whose max_kw it does not exceed and whose
+ phases, when the rule names them, equal the load's. No standard table exists
+ in this repository: without rules no section is proposed.
+ """
+ if section_rules is None:
+ return ()
+ if not isinstance(section_rules, (list, tuple)):
+ raise ValueError("section_rules: expected a list of {max_kw, section[, phases]}")
+ rules = []
+ for i, rule in enumerate(section_rules):
+ where = f"section_rules[{i}]"
+ if not isinstance(rule, dict):
+ raise ValueError(f"{where}: expected {{max_kw, section[, phases]}}, got {rule!r}")
+ unknown = sorted(set(rule) - set(SECTION_RULE_KEYS))
+ if unknown:
+ raise ValueError(f"{where}: unknown key(s) {', '.join(map(str, unknown))}")
+ max_kw = _number(rule.get("max_kw"), f"{where}.max_kw")
+ if max_kw <= 0.0:
+ raise ValueError(f"{where}.max_kw: must be greater than zero, got {max_kw:g}")
+ section = rule.get("section")
+ if not isinstance(section, str) or not section.strip():
+ raise ValueError(
+ f"{where}.section: expected a non-empty text such as '5x2.5', got {section!r}"
+ )
+ phases = rule.get("phases")
+ if phases is not None and phases not in (1, 3):
+ raise ValueError(f"{where}.phases: 1 or 3, got {phases!r}")
+ rules.append({"max_kw": max_kw, "section": section.strip(), "phases": phases})
+ return tuple(sorted(rules, key=lambda r: (r["max_kw"], r["phases"] or 0)))
+
+
+def section_for(kw: float | None, phases: int | None, rules) -> str | None:
+ """The first rule the load fits, or None - never a guessed section."""
+ if kw is None:
+ return None
+ for rule in rules:
+ if kw <= rule["max_kw"] and (rule["phases"] is None or rule["phases"] == phases):
+ return rule["section"]
+ return None
+
+
+def roundup_m(length_m: float, allowance: float) -> int:
+ """ROUNDUP(length x (1 + allowance)) to the metre, on the value rounded to 1e-6 m first.
+
+ The rounding keeps binary noise from buying a metre of cable: 50 m with a
+ 10 % allowance is 55.00000000000001 in floating point, and 55 m on the sheet.
+ """
+ return math.ceil(round(length_m * (1.0 + allowance), 6))
+
+
+def _nearest_tag(at: Pt, places: dict[str, list[Pt]], search: float) -> str | None:
+ found = [
+ (math.dist(at, p), tag)
+ for tag, points in places.items()
+ for p in points
+ if math.dist(at, p) <= search
+ ]
+ return min(found)[1] if found else None
+
+
+def _inside_box(p: Pt, box) -> bool:
+ return box is None or (box[0] <= p[0] <= box[2] and box[1] <= p[1] <= box[3])
+
+
+def cable_rows(
+ pid: Snapshot, layout: Snapshot | None, *, allowance=0.20, section_rules=None
+) -> dict:
+ """One row per electrical load, the cable method made repeatable (spec §9).
+
+ Loads are the P&ID tags with electrical texts near them (`parse_electrical`:
+ kW, phases, voltage, + N, VFD) and the tags a wiring callout names. The
+ panel comes from the layout's callouts (`wiring_target`: 'Wiring to CP1'
+ -> CP-1), else the P&ID's. The length is the Manhattan distance on the
+ layout from the load's tag to its panel's tag, and the cable is
+ `roundup_m(length, allowance)`. **Power is never invented**: a load with no
+ stated kW keeps an empty cell and an open item. With no layout, lengths
+ stay empty and every row gets an open item - the P&ID is not a length
+ source for cables. A load wired to a panel that itself states a power is
+ part of that package: room and panel totals count the package's power and
+ not its children's. Sections come only from `section_rules`.
+
+ Refused by name before anything is read: an allowance outside 0-1 and a
+ malformed section rule.
+ """
+ allowance = check_allowance(allowance)
+ rules = check_section_rules(section_rules)
+ pid_places = {tag: [p["at"] for p in places] for tag, places in tag_occurrences(pid).items()}
+ pid_search = CABLE_SEARCH_SHARE * label_search_default(pid)
+
+ loads: dict[str, list] = {}
+ orphans: list[str] = []
+ pid_callouts: dict[str, str] = {}
+ for rec in pid.records:
+ if rec.space != MODEL or rec.type not in TEXT_TYPES or not rec.text or not rec.points:
+ continue
+ at = rec.points[0]
+ target = wiring_target(rec.text)
+ if target is not None:
+ tag = _nearest_tag(at, pid_places, pid_search)
+ if tag is not None:
+ pid_callouts.setdefault(tag, target)
+ continue
+ found = parse_electrical(rec.text)
+ if found["kw"] is None and not (found["phases"] or found["voltage"] or found["vfd"]):
+ continue
+ tag = _nearest_tag(at, pid_places, pid_search)
+ if tag is None:
+ orphans.append(rec.handle)
+ continue
+ distance = min(math.dist(at, p) for p in pid_places[tag])
+ loads.setdefault(tag, []).append((distance, rec.handle, found))
+
+ positions: dict[str, Pt] = {}
+ ambiguous: set[str] = set()
+ panels: dict[str, tuple[str, str]] = {}
+ layout_unit = unit_of(layout) if layout is not None else None
+ cluster = None
+ layout_search = None
+ if layout is not None:
+ placed = layout_tags(layout, set(loads) | set(pid_callouts))
+ positions, ambiguous, cluster = (
+ placed["positions"],
+ set(placed["ambiguous"]),
+ placed["cluster"],
+ )
+ layout_search = CABLE_SEARCH_SHARE * label_search_default(layout)
+ layout_places = {tag: [p] for tag, p in positions.items()}
+ for rec in layout.records:
+ if rec.space != MODEL or rec.type not in TEXT_TYPES or not rec.text or not rec.points:
+ continue
+ target = wiring_target(rec.text)
+ if target is None or not _inside_box(rec.points[0], cluster):
+ continue
+ tag = _nearest_tag(rec.points[0], layout_places, layout_search)
+ if tag is not None:
+ panels.setdefault(tag, (target, "layout"))
+ for tag, target in pid_callouts.items():
+ panels.setdefault(tag, (target, "pid"))
+ panel_at = {}
+ for tag, p in positions.items():
+ name = normalize_panel(tag)
+ if name is not None:
+ panel_at[name] = p
+ regions = room_regions(layout, cluster) if layout is not None else room_regions(pid)
+
+ rows, items = [], []
+ for tag in sorted(set(loads) | set(panels)):
+ texts = sorted(loads.get(tag, []), key=lambda item: item[:2])
+ powers = [(d, f["kw"]) for d, _h, f in texts if f["kw"] is not None]
+ kw = powers[0][1] if powers else None
+ if len({value for _d, value in powers}) > 1:
+ items.append(
+ {
+ "tag": tag,
+ "item": "conflicting_power",
+ "detail": ", ".join(f"{value:g} kW" for _d, value in powers),
+ }
+ )
+ phases = next((f["phases"] for _d, _h, f in texts if f["phases"]), None)
+ voltage = next((f["voltage"] for _d, _h, f in texts if f["voltage"]), None)
+ panel, panel_source = panels.get(tag, (None, None))
+ row = {
+ "tag": tag,
+ "room": "",
+ "at": None,
+ "kw": kw,
+ "phases": phases,
+ "voltage": voltage,
+ "neutral": any(f["neutral"] for _d, _h, f in texts),
+ "vfd": any(f["vfd"] for _d, _h, f in texts),
+ "panel": panel,
+ "panel_source": panel_source,
+ "length_m": None,
+ "allowance": allowance,
+ "cable_m": None,
+ "section": section_for(kw, phases, rules),
+ "package": None,
+ }
+ if kw is None:
+ items.append(
+ {
+ "tag": tag,
+ "item": "missing_power",
+ "detail": "no power is stated near the tag on the P&ID",
+ }
+ )
+ if panel is None:
+ items.append(
+ {
+ "tag": tag,
+ "item": "missing_panel",
+ "detail": "no wiring callout names a panel for this tag",
+ }
+ )
+ if rules and kw is not None and row["section"] is None:
+ items.append(
+ {"tag": tag, "item": "no_section_rule", "detail": f"no rule covers {kw:g} kW"}
+ )
+ if layout is None:
+ items.append(
+ {
+ "tag": tag,
+ "item": "no_layout",
+ "detail": "no layout given: the P&ID is not a length source for cables",
+ }
+ )
+ if pid_places.get(tag):
+ row["room"] = room_at(regions, pid_places[tag][0])
+ else:
+ here = positions.get(tag)
+ if here is None:
+ items.append(
+ {
+ "tag": tag,
+ "item": "tag_ambiguous" if tag in ambiguous else "tag_not_on_layout",
+ "detail": "the load's tag cannot be placed on the layout",
+ }
+ )
+ else:
+ row["at"] = here
+ row["room"] = room_at(regions, here)
+ if panel is not None and panel not in panel_at:
+ items.append(
+ {
+ "tag": tag,
+ "item": "panel_not_on_layout",
+ "detail": f"{panel} is not written once on the layout",
+ }
+ )
+ if here is not None and panel in panel_at:
+ row["length_m"] = manhattan(here, panel_at[panel]) * layout_unit["m_per_unit"]
+ row["cable_m"] = roundup_m(row["length_m"], allowance)
+ rows.append(row)
+
+ stated = {
+ (normalize_panel(r["tag"]) or r["tag"]): r["tag"] for r in rows if r["kw"] is not None
+ }
+ for row in rows:
+ owner = stated.get(row["panel"]) if row["panel"] else None
+ if owner is not None and owner != row["tag"]:
+ row["package"] = owner
+
+ def counted(row: dict) -> float:
+ return row["kw"] if row["kw"] is not None and row["package"] is None else 0.0
+
+ summary = {}
+ for name, field in (("by_room", "room"), ("by_panel", "panel")):
+ summary[name] = [
+ {
+ "key": key,
+ "loads": sum(1 for r in rows if (r[field] or "") == key),
+ "kw": sum(counted(r) for r in rows if (r[field] or "") == key),
+ "cable_m": sum(r["cable_m"] or 0 for r in rows if (r[field] or "") == key),
+ }
+ for key in sorted({row[field] or "" for row in rows})
+ ]
+ units = {"pid": unit_of(pid), "layout": layout_unit}
+ warnings = [unit["warning"] for unit in units.values() if unit and unit["warning"]]
+ if orphans:
+ warnings.append(f"electrical text(s) near no tag, not used: {', '.join(orphans)}")
+ return {
+ "kind": "cable",
+ "rows": rows,
+ "open_items": items,
+ "summary": summary,
+ "totals": {
+ "loads": len(rows),
+ "known_kw": sum(counted(r) for r in rows),
+ "cable_m": sum(r["cable_m"] or 0 for r in rows),
+ "open_items": len(items),
+ "packages": sum(1 for r in rows if r["package"]),
+ },
+ "units": units,
+ "warnings": warnings,
+ "method": {
+ "sources": {"pid": pid.source, "layout": layout.source if layout else None},
+ "allowance": allowance,
+ "search": {"pid": pid_search, "layout": layout_search},
+ "section_rules": [dict(rule) for rule in rules],
+ "rooms": _rooms_mode(regions),
+ },
+ }
diff --git a/server.py b/server.py
index 96cff3c..4a9771d 100644
--- a/server.py
+++ b/server.py
@@ -10555,7 +10555,7 @@ async def arch_plan_from_spec(
# ---------------------------------------------------------------------------
-# ── SECTION 27: Plant Takeoffs (1 tool) ─────────────────────────────────────
+# ── SECTION 27: Plant Takeoffs (2 tools) ────────────────────────────────────
# ---------------------------------------------------------------------------
@@ -10714,6 +10714,111 @@ async def pipe_takeoff(
raise ToolError(str(exc)) from exc
+@cad_tool(
+ summary="Cable takeoff: one row per load, Manhattan length to its panel, rounded up to the metre.",
+ cost="safe",
+)
+@mcp.tool(
+ annotations={
+ "title": "Plant: Cable Takeoff",
+ "readOnlyHint": False,
+ "destructiveHint": False,
+ },
+ tags={"plant", "analysis"},
+)
+async def cable_takeoff(
+ pid_path: Annotated[
+ str | None,
+ Field(
+ default=None,
+ description="The P&ID carrying the electrical texts (kW, phases, voltage, VFD). "
+ "Omit to read the current drawing.",
+ ),
+ ] = None,
+ layout_path: Annotated[
+ str | None,
+ Field(
+ default=None,
+ description="The layout with the tags, the panels and the 'Wiring to CPn' "
+ "callouts. Omitted, every length stays empty with an open item.",
+ ),
+ ] = None,
+ output: Annotated[
+ str | None,
+ Field(
+ default=None,
+ description="Workbook to write (.xlsx); the four sheets are also written as CSV "
+ "beside it. Omit to return the rows only.",
+ ),
+ ] = None,
+ lang: Annotated[
+ str, Field(default="en", description="Headers and method text: tr | en | ru.")
+ ] = "en",
+ allowance: Annotated[
+ float,
+ Field(
+ default=0.20,
+ ge=0.0,
+ le=1.0,
+ description="Allowance before rounding up to the metre (0.20 = 20 %).",
+ ),
+ ] = 0.20,
+ section_rules: Annotated[
+ list[dict] | None,
+ Field(
+ default=None,
+ description="Your own section table: [{max_kw, section, phases (1|3, optional)}], "
+ "e.g. [{'max_kw': 5.5, 'section': '5x2.5'}]. Omitted, no section is proposed.",
+ ),
+ ] = None,
+ ctx: Context = None,
+) -> dict:
+ """A cable takeoff: one row per electrical load, the shop's cable method made repeatable.
+
+ Power, phases, voltage, neutral and VFD come from the P&ID's electrical
+ texts near each tag (Russian, Turkish and English spellings). **Power is
+ never invented**: a load with no stated kW keeps an empty cell and an open
+ item. The panel comes from the layout's wiring callouts ('Wiring to CP1'
+ and its variants, normalised to CP-1), else from the P&ID. The length is the
+ Manhattan distance on the layout from the load's tag to its panel's tag,
+ and the cable is ROUNDUP(length x (1 + allowance)) to the metre. Room and
+ panel totals count known power once: a load wired to a panel that states
+ its own power is part of that package, and the package's power is counted
+ instead of its children's - the rule is written on the method sheet.
+ Sections only from `section_rules`; no standard table is applied. Open
+ items: missing power, missing panel, a tag or panel not on the layout, no
+ layout at all, a load no section rule covers.
+
+ Read-only on both drawings. With `output` it writes one workbook - Metraj,
+ Özet, Kontrol, Metodoloji - as XLSX and, always, as CSV beside it.
+
+ Refused before a drawing is read: `lang` outside tr / en / ru, an allowance
+ outside 0-1 and a malformed section rule (named by its path,
+ `section_rules[0].section: ...`). A missing, unreadable or oversize drawing
+ is refused by the reader, which names MAX_DXF_BYTES. Without openpyxl the
+ XLSX is not written and `files.refused` carries `capability: "xlsx_write"`
+ naming the `office` extra; the CSV files are written all the same.
+ """
+ from engineering.understand.draw import run_cable_takeoff
+
+ pid = str(validate_path(pid_path)) if pid_path else None
+ layout = str(validate_path(layout_path)) if layout_path else None
+ target = str(validate_path(output, allow_write=True)) if output else None
+ await ctx.info(f"cable_takeoff: {pid or 'current drawing'} / {layout or 'no layout'}")
+ try:
+ return await run_cable_takeoff(
+ _backend(ctx),
+ pid_path=pid,
+ layout_path=layout,
+ output=target,
+ lang=lang,
+ allowance=allowance,
+ section_rules=section_rules,
+ )
+ except (ValueError, OSError) as exc:
+ raise ToolError(str(exc)) from exc
+
+
# ---------------------------------------------------------------------------
# ── RESOURCES ───────────────────────────────────────────────────────────────
# ---------------------------------------------------------------------------
diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py
index fc4d189..e54ff07 100644
--- a/tests/test_tool_registry.py
+++ b/tests/test_tool_registry.py
@@ -495,7 +495,7 @@ async def test_tool_groups_is_byte_identical_to_the_source_declared_grouping():
async def test_tool_group_sizes_are_unchanged():
- """Frozen snapshot of the surface (242 tools, 26 groups).
+ """Frozen snapshot of the surface (243 tools, 26 groups).
Taken before @cad_tool landed at 131 tools; `batch` moved 2 -> 3 when
v1.5.0's `cad_batch` joined `entity_batch_create`/`entity_batch_modify`, and
@@ -598,10 +598,13 @@ async def test_tool_group_sizes_are_unchanged():
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.
"""
sizes = {label: len(names) for label, names in (await server._tool_groups()).items()}
assert sizes == {
- "analysis": 13,
+ "analysis": 14,
"architecture": 12,
"batch": 3,
"blocks": 9,
@@ -628,4 +631,4 @@ async def test_tool_group_sizes_are_unchanged():
"validation": 1,
"view": 4,
}
- assert sum(sizes.values()) == 242
+ assert sum(sizes.values()) == 243
diff --git a/tests/test_understand_cable.py b/tests/test_understand_cable.py
new file mode 100644
index 0000000..ef9f56a
--- /dev/null
+++ b/tests/test_understand_cable.py
@@ -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)
diff --git a/tests/test_understand_tools.py b/tests/test_understand_tools.py
index cf96a3e..db15d8e 100644
--- a/tests/test_understand_tools.py
+++ b/tests/test_understand_tools.py
@@ -1,4 +1,5 @@
-"""pipe_takeoff over the MCP wire, on the synthetic plant pair - never on a client drawing."""
+"""pipe_takeoff and cable_takeoff over the MCP wire, on the synthetic plant pair - never on a
+client drawing."""
from __future__ import annotations
@@ -101,3 +102,41 @@ async def test_the_tool_says_it_writes_a_file_and_destroys_nothing(client):
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