feat(understand): cable_takeoff - one row per load, rounded to the metre

cable_rows reads each load's power, phases, voltage, neutral and VFD from the
P&ID's electrical texts near its tag, and its panel from the layout's wiring
callouts (else the P&ID's), names normalised CP 1 -> CP-1. The length is the
Manhattan distance on the layout from the tag to its panel, and the cable
ROUNDUP(length x (1 + allowance)) to the metre, on the length rounded to
1e-6 m so binary noise never buys a metre. Power is never invented: a load
with no stated kW keeps an empty cell and an open item. Without a layout the
lengths stay empty with an open item per row. Room and panel totals count a
panel package's power once, never its children's as well; the rule is on the
method sheet. Sections only from the caller's section_rules - no standard
table exists here.

The workbook gains the cable shape (Metraj, Ozet by room and panel, Kontrol
as the open-items list, Metodoloji) in the same three languages;
cable_takeoff joins SECTION 27. Inventory regenerated; README snapshot 243
tools (238 advertised); registry snapshot analysis: 14.

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
2026-09-24 12:28:59 +03:00
co-authored by Claude Opus 5.5
parent 653535bcc1
commit 459d3b9228
10 changed files with 1110 additions and 15 deletions
+35 -2
View File
@@ -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
+190 -1
View File
@@ -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:
+310 -2
View File
@@ -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),
},
}