diff --git a/CHANGELOG.md b/CHANGELOG.md
index 12eb2a7..6141997 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -401,6 +401,28 @@ member. Spec: `docs/superpowers/specs/2026-09-24-v1.6-understand-design.md`.
naming the room it serves, or a note pointing into a room. An unnumbered
label naming equipment is no room. Without room faces, when the drawing
numbers its rooms, only numbered labels take pieces.
+ - The bare Russian "to" before a panel (`к CP1`, `к щиту CP-2`) is a
+ wiring callout.
+- **Track H, what the field test asked for** (each with a synthetic test):
+ - **Rooms drawn as frames.** A room label no wall face holds takes the
+ smallest closed, straight-sided polyline around it that holds no other
+ room number and is at least ten label heights a side (`label+frame`,
+ confidence 0.8). This is how a P&ID draws its rooms, so P&ID lengths are
+ cut at the room frames.
+ - **The nearest panel, flagged.** `cable_takeoff(panel_rule="nearest")`,
+ the default: a load that no callout places, whose tag is on the layout,
+ takes the nearest panel there. Only names with a panel prefix
+ (`is_panel_name`: CP, JB, DB, MCC ...) or names a callout gives are
+ candidates. The row is flagged `panel_inferred`, and an inferred panel
+ never makes a package. `panel_rule="stated"` leaves such a load open. The
+ benchmark and the correctness check read stated panels only.
+ - **Possible packages and detail copies.** `pipe_takeoff` reports
+ `detail_copies`: compact groups of tags drawn twice (`repeated`), and
+ compact groups of one equipment's own parts, such as M150B, M150C and
+ M150D of M150 (`family`). A pasteurizer drawn in detail, whose internal
+ pipes its vendor supplies, shows up this way. Nothing is removed until
+ `exclude_regions=["D2"]` names it. The excluded runs go to the control
+ rows as `excluded_region` and are totalled in `excluded_m`.
### Changed
diff --git a/README.md b/README.md
index 2489600..86a1ef5 100644
--- a/README.md
+++ b/README.md
@@ -27,7 +27,7 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
-> **v1.5 release snapshot:** 247 tools · 8 resources · 5 prompt templates · 4982 collected tests.
+> **v1.5 release snapshot:** 247 tools · 8 resources · 5 prompt templates · 4987 collected tests.
> 247 is the **registered** count; a default install advertises 242 over `tools/list`,
> because `ENABLE_3D` is unset. `system_about` is the runtime authority.
diff --git a/benchmarks/adapters/autocad_mcp_pro.py b/benchmarks/adapters/autocad_mcp_pro.py
index 0f39e6f..9fc6789 100644
--- a/benchmarks/adapters/autocad_mcp_pro.py
+++ b/benchmarks/adapters/autocad_mcp_pro.py
@@ -837,7 +837,9 @@ class AutoCADMCPProAdapter(BenchmarkAdapter):
if reasons:
pipe_mismatch.append({"run": run["id"], "reason": "; ".join(reasons)})
- cable = {row["tag"]: row for row in cable_rows(pid, layout)["rows"]}
+ # panel_rule="stated": the gate reads what the drawing says; a nearest-panel
+ # guess is no evidence
+ cable = {row["tag"]: row for row in cable_rows(pid, layout, panel_rule="stated")["rows"]}
cable_mismatch = []
for want in truth["cable_rows"]:
got = cable.get(want["tag"])
diff --git a/benchmarks/correctness_suite.py b/benchmarks/correctness_suite.py
index 0038ac8..a3ae684 100644
--- a/benchmarks/correctness_suite.py
+++ b/benchmarks/correctness_suite.py
@@ -915,7 +915,9 @@ async def cable_takeoff_roundup():
truth = _plant_pair()
rows = {
row["tag"]: row
- for row in cable_rows(read_snapshot(truth["pid"]), read_snapshot(truth["layout"]))["rows"]
+ for row in cable_rows(
+ read_snapshot(truth["pid"]), read_snapshot(truth["layout"]), panel_rule="stated"
+ )["rows"]
}
expected = {"M11": ("CP-1", 9), "T102": ("CP-1", 17), "M12": ("CP-2", 19), "T202": ("CP-2", 10)}
for tag, (panel, metres) in expected.items():
diff --git a/engineering/understand/describe.py b/engineering/understand/describe.py
index be1b408..c20f008 100644
--- a/engineering/understand/describe.py
+++ b/engineering/understand/describe.py
@@ -52,6 +52,11 @@ ATTRIBUTE_TAG_CONFIDENCE = 0.95
TEXT_TAG_CONFIDENCE = 0.8
ROOM_FACE_CONFIDENCE = 0.9
ROOM_LABEL_CONFIDENCE = 0.6
+#: A room read from a closed frame around its label: weaker than wall faces.
+ROOM_FRAME_CONFIDENCE = 0.8
+#: A frame is a room only when each side is at least this many label text
+#: heights long: a box drawn tight around the label is no room.
+FRAME_MIN_TEXT_HEIGHTS = 10.0
TITLE_NAME_CONFIDENCE = 0.9
TITLE_ATTRIBUTE_CONFIDENCE = 0.6
#: The face finder pairs every wall segment with every other; above this many
@@ -178,8 +183,12 @@ def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict)
A label is any text `vocab.room_label` recognises. Wall layers are the
ones `vocab.classify_layer` files under ``architecture``; their straight
segments go through `engineering.arch.faces.planar_faces` (arcs are left
- out and counted). ``labelled`` pairs each room row with its `Face` for a
- caller that needs the geometry; ``rooms`` is JSON-ready.
+ out and counted). A label no wall face holds takes the smallest closed,
+ straight-sided polyline around it that holds no other room label and is at
+ least ``FRAME_MIN_TEXT_HEIGHTS`` label heights a side - how a P&ID draws
+ its rooms - at ``ROOM_FRAME_CONFIDENCE``. ``labelled`` pairs each room row
+ with its `Face` for a caller that needs the geometry; ``rooms`` is
+ JSON-ready.
"""
labels = []
for rec in records:
@@ -205,8 +214,12 @@ def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict)
rooms: list[dict] = []
labelled: list[tuple[dict, Face]] = []
used: set[int] = set()
+ frames = _frames(records)
for rec, parsed, at in labels:
face = face_containing(faces, at) if faces else None
+ from_frame = None
+ if face is None and frames:
+ from_frame = _frame_around(rec, parsed, at, frames, labels)
row = {
"number": parsed.get("number"),
"name": parsed.get("name"),
@@ -221,14 +234,16 @@ def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict)
}
if face is not None:
used.add(id(face))
+ chosen = face or from_frame
+ if chosen is not None:
row["face"] = {
- "area": face.area,
- "polygon": [[p[0], p[1]] for p in face.loop],
- "centroid": [face.centroid[0], face.centroid[1]],
+ "area": chosen.area,
+ "polygon": [[p[0], p[1]] for p in chosen.loop],
+ "centroid": [chosen.centroid[0], chosen.centroid[1]],
}
- row["source"] = "label+face"
- row["confidence"] = ROOM_FACE_CONFIDENCE
- labelled.append((row, face))
+ row["source"] = "label+face" if face is not None else "label+frame"
+ row["confidence"] = ROOM_FACE_CONFIDENCE if face is not None else ROOM_FRAME_CONFIDENCE
+ labelled.append((row, chosen))
rooms.append(row)
rooms.sort(key=lambda r: (r["cluster"] or "", r["number"] or "", r["name"] or "", r["at"]))
return {
@@ -242,6 +257,74 @@ def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict)
}
+def _frames(records: Sequence[EntityRecord]) -> list[tuple[EntityRecord, Face]]:
+ """Closed, straight-sided polylines as faces: the frames a room may be drawn as."""
+ out = []
+ for rec in records:
+ if rec.type not in ("LWPOLYLINE", "POLYLINE") or not rec.closed or len(rec.points) < 3:
+ continue
+ if any(rec.bulges):
+ continue
+ face = _polygon_face(rec.points)
+ if face is not None:
+ out.append((rec, face))
+ return out
+
+
+def _polygon_face(points) -> Face | None:
+ pts = [(float(x), float(y)) for x, y in points]
+ signed = 0.0
+ cx = cy = 0.0
+ for (x0, y0), (x1, y1) in zip(pts, pts[1:] + pts[:1], strict=True):
+ cross = x0 * y1 - x1 * y0
+ signed += cross
+ cx += (x0 + x1) * cross
+ cy += (y0 + y1) * cross
+ signed /= 2.0
+ if abs(signed) <= 0.0:
+ return None
+ centroid = (cx / (6.0 * signed), cy / (6.0 * signed))
+ if signed < 0.0:
+ pts.reverse()
+ start = min(range(len(pts)), key=lambda i: (pts[i][1], pts[i][0]))
+ loop = tuple(pts[start:] + pts[:start])
+ return Face(loop=loop, area=abs(signed), centroid=centroid)
+
+
+def _frame_around(rec, parsed, at, frames, labels) -> Face | None:
+ """The smallest frame around ``at`` that is a room: no label of another room
+ number in it (an unnumbered mention - 'hot room' - does not count), each side
+ at least FRAME_MIN_TEXT_HEIGHTS label heights."""
+ least = FRAME_MIN_TEXT_HEIGHTS * float(rec.height or 0.0)
+ best = None
+ for frame_rec, face in frames:
+ box = frame_rec.bbox or (
+ min(p[0] for p in face.loop),
+ min(p[1] for p in face.loop),
+ max(p[0] for p in face.loop),
+ max(p[1] for p in face.loop),
+ )
+ if not (box[0] <= at[0] <= box[2] and box[1] <= at[1] <= box[3]):
+ continue
+ if min(box[2] - box[0], box[3] - box[1]) < least:
+ continue
+ if face_containing([face], at) is None:
+ continue
+ others = [
+ other_at
+ for other, other_parsed, other_at in labels
+ if other is not rec
+ and other_parsed["number"]
+ and other_parsed["number"] != parsed["number"]
+ and face_containing([face], other_at) is not None
+ ]
+ if others:
+ continue
+ if best is None or face.area < best.area:
+ best = face
+ return best
+
+
def drawing_units(snap: Snapshot, *, prep: dict | None = None, rooms: dict | None = None) -> dict:
"""`units.infer_units` the one way every reader calls it: on the records kept
after the outliers, with the labelled room faces as evidence.
diff --git a/engineering/understand/draw.py b/engineering/understand/draw.py
index 0ba308e..a01bcc3 100644
--- a/engineering/understand/draw.py
+++ b/engineering/understand/draw.py
@@ -16,6 +16,7 @@ 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 (
+ PANEL_RULES,
cable_rows,
check_allowance,
check_pipe_options,
@@ -51,6 +52,7 @@ async def run_pipe_takeoff(
force: bool = False,
layers=None,
tol: float | None = None,
+ exclude_regions=None,
) -> dict:
"""`pipe_rows` over a P&ID and an optional layout, plus the workbook when `output` is named.
@@ -85,6 +87,7 @@ async def run_pipe_takeoff(
vertical_allowance=options["vertical_allowance"],
length_source=options["length_source"],
force=bool(force),
+ exclude_regions=tuple(exclude_regions or ()),
)
result = await asyncio.to_thread(compute)
@@ -105,15 +108,25 @@ async def run_cable_takeoff(
lang: str = "en",
allowance: float = 0.20,
section_rules=None,
+ panel_rule: str = "nearest",
) -> 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)
+ if panel_rule not in PANEL_RULES:
+ raise ValueError(
+ f"panel_rule: {panel_rule!r} unknown; choose from {', '.join(PANEL_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
+ cable_rows,
+ pid,
+ layout,
+ allowance=allowance,
+ section_rules=list(rules) or None,
+ panel_rule=panel_rule,
)
result["files"] = (
await asyncio.to_thread(write_workbook, result, kind="cable", lang=lang, path=output)
diff --git a/engineering/understand/labels.py b/engineering/understand/labels.py
index 06f4289..b94dbe4 100644
--- a/engineering/understand/labels.py
+++ b/engineering/understand/labels.py
@@ -32,6 +32,7 @@ import re
__all__ = [
"CYRILLIC_LOOKALIKES",
"fold_lookalikes",
+ "is_panel_name",
"normalize_panel",
"parse_diameters",
"parse_electrical",
@@ -299,6 +300,21 @@ def parse_electrical(text: str | None) -> dict:
}
+#: Letter prefixes that name what a load is wired to: control panel, junction
+#: box, distribution board, motor control centre, local / motor control panel,
+#: power panel, switchboard. A convention list, not a standard.
+PANEL_PREFIXES = ("CP", "JB", "DB", "MCC", "LCP", "MCP", "PP", "SB", "SWB")
+
+
+def is_panel_name(name: str | None) -> bool:
+ """A normalised name whose letters say it is a panel: ``CP-1``, ``JB-2``,
+ ``CP-M82``, ``MCC-3`` - not ``M-87``, which `normalize_panel` would also
+ shape, because any letters-and-digits token has a panel's form."""
+ if not name:
+ return False
+ return str(name).split("-", 1)[0] in PANEL_PREFIXES
+
+
def wiring_target(text: str | None) -> str | None:
"""The panel a wiring callout points to, normalised.
diff --git a/engineering/understand/report.py b/engineering/understand/report.py
index 59d210b..c1881dd 100644
--- a/engineering/understand/report.py
+++ b/engineering/understand/report.py
@@ -80,6 +80,12 @@ LABELS: dict[str, tuple[str, str, str]] = {
# Kontrol kinds and reasons
"unassigned_diameter": ("Çap atanmamış", "Diameter unassigned", "Диаметр не назначен"),
"unroutable": ("Yönlendirilemez", "Unroutable", "Трассировка невозможна"),
+ "excluded_region": (
+ "Detay kopyası (hariç)",
+ "Detail copy (left out)",
+ "Копия детали (исключена)",
+ ),
+ "detail_copies": ("Olası detay kopyaları", "Possible detail copies", "Возможные копии деталей"),
"end_not_on_equipment": (
"Bir uç ekipmanda değil",
"An end is on no equipment",
@@ -192,6 +198,7 @@ LABELS: dict[str, tuple[str, str, str]] = {
"rule_rounding": ("Kural: yuvarlama", "Rule: rounding", "Правило: округление"),
"rule_power": ("Kural: güç", "Rule: power", "Правило: мощность"),
"rule_panel": ("Kural: pano", "Rule: panel", "Правило: шкаф"),
+ "panel_inferred": ("Pano tahmini (en yakın)", "Panel inferred (nearest)", "Шкаф по близости"),
"rule_package": ("Kural: pano paketleri", "Rule: panel packages", "Правило: комплектные шкафы"),
"rule_section": ("Kural: kesit", "Rule: section", "Правило: сечение"),
"section_rules": (
@@ -315,6 +322,17 @@ TEXTS: dict[str, tuple[str, str, str]] = {
"Шкаф берётся из выносок подключения на планировке ('Wiring to CP1' и варианты), иначе "
"из P&ID; имена нормализуются (CP 1 → CP-1).",
),
+ "rule_panel_nearest": (
+ "Pano, yerleşimdeki bağlantı notlarından, yoksa P&ID'den alınır; hiçbiri yoksa ve yükün "
+ "etiketi yerleşimde ise yerleşimdeki en yakın pano seçilir ve satır Kontrol sayfasında "
+ "'pano tahmini' olarak işaretlenir. Adlar normalleştirilir (CP 1 → CP-1).",
+ "The panel comes from the layout's wiring callouts, else from the P&ID; when neither "
+ "names one and the load's tag is on the layout, the nearest panel there is taken and the "
+ "row is flagged 'panel inferred' on the control sheet. Names are normalised (CP 1 → CP-1).",
+ "Шкаф берётся из выносок подключения на планировке, иначе из 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 "
@@ -526,6 +544,15 @@ def _pipe_method(result: dict, lang: str) -> list[list]:
["tol", method["tol"]],
["label_search", round(float(method["label_search"]), 6)],
["overlap", _num(method["overlap_length_m"])],
+ [
+ "detail_copies",
+ "; ".join(
+ f"{region['id']} ({', '.join(region['tags'])})"
+ + (" - " + _t("excluded_region", lang) if region["excluded"] else "")
+ for region in result.get("detail_copies", [])
+ )
+ or "—",
+ ],
]
rows += [["warning", warning] for warning in result["warnings"]]
return [[_t(topic, lang), value] for topic, value in rows]
@@ -615,7 +642,13 @@ def _cable_method(result: dict, lang: str) -> list[list]:
["rule_allowance", method["allowance"]],
["rule_rounding", _say("rule_rounding", lang)],
["rule_power", _say("rule_power", lang)],
- ["rule_panel", _say("rule_panel", lang)],
+ [
+ "rule_panel",
+ _say(
+ "rule_panel_nearest" if method.get("panel_rule") == "nearest" else "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 "—"],
diff --git a/engineering/understand/takeoff.py b/engineering/understand/takeoff.py
index e3e3b2f..0ee9b25 100644
--- a/engineering/understand/takeoff.py
+++ b/engineering/understand/takeoff.py
@@ -27,11 +27,17 @@ drawing.
from __future__ import annotations
import math
+import re
from collections.abc import Iterable
from engineering.arch.faces import face_containing
from engineering.understand.describe import FACE_TOL_FRACTION, drawing_units, find_rooms, prepare
-from engineering.understand.labels import normalize_panel, parse_electrical, wiring_target
+from engineering.understand.labels import (
+ is_panel_name,
+ normalize_panel,
+ parse_electrical,
+ wiring_target,
+)
from engineering.understand.network import (
MODEL,
TEXT_TYPES,
@@ -371,6 +377,150 @@ def _pid_cells(edges, regions: dict, m_per_unit: float) -> dict:
return cells
+#: How a load no callout places gets its panel: the nearest one on the layout
+#: (flagged ``panel_inferred``), or none (``missing_panel``).
+PANEL_RULES = ("nearest", "stated")
+
+
+def _nearest_panel(here: Pt, panel_at: dict, *, exclude=None) -> tuple[str, float] | None:
+ """(panel, distance in layout units) of the panel nearest ``here``; ties by name."""
+ candidates = [(math.dist(here, p), name) for name, p in panel_at.items() if name != exclude]
+ if not candidates:
+ return None
+ distance, name = min(candidates)
+ return name, distance
+
+
+#: Detail copies: tags drawn twice are grouped when their occurrences lie within
+#: this share of the P&ID's robust diagonal of each other (single linkage) ...
+DETAIL_LINK_SHARE = 0.03
+#: ... and a group is a possible detail copy when it holds at least this many
+#: distinct repeated tags, each of which also stands somewhere outside it. The
+#: same count of an equipment's own parts (M150 -> M150B, M150C, M150D) standing
+#: within DETAIL_FAMILY_SHARE of the diagonal is a possible package or detail.
+DETAIL_MIN_TAGS = 3
+DETAIL_FAMILY_SHARE = 0.15
+_FAMILY_TAG = re.compile(r"^(?P[A-Z]{1,4}-?\d{1,5})(?P[A-Z]{1,2})$")
+
+
+def detail_copy_regions(pid: Snapshot, *, prep: dict | None = None) -> list[dict]:
+ """Where a P&ID may draw pipes twice: ``[{"id", "box", "tags", "kind", "base"}]``.
+
+ ``repeated``: tags written more than once (readings of one tag closer than
+ half the link are one) grouped by where they stand; a compact group of at
+ least ``DETAIL_MIN_TAGS`` such tags, every one also written outside it, is a
+ possible detail copy. ``family``: at least ``DETAIL_MIN_TAGS`` parts of one
+ equipment (``M150B``, ``M150C`` ... of ``M150``) standing close together, a
+ possible package or detail whose internal pipes a site takeoff may leave to
+ its vendor. Ids run D1, D2 ... from the left. A heuristic that reports and
+ never removes: the caller decides.
+ """
+ prep = prep or prepare(pid)
+ link = DETAIL_LINK_SHARE * prep["diag"]
+ if link <= 0.0:
+ return []
+ occurrences = {
+ tag: [place["at"] for place in places] for tag, places in tag_occurrences(pid).items()
+ }
+ points = [
+ (tag, at)
+ for tag, places in occurrences.items()
+ for at in _distinct(places, link / 2.0)
+ if len(_distinct(places, link / 2.0)) > 1
+ ]
+ parent = list(range(len(points)))
+
+ def find(k: int) -> int:
+ while parent[k] != k:
+ parent[k] = parent[parent[k]]
+ k = parent[k]
+ return k
+
+ grid: dict[tuple[int, int], list[int]] = {}
+ for k, (_tag, at) in enumerate(points):
+ grid.setdefault((math.floor(at[0] / link), math.floor(at[1] / link)), []).append(k)
+ for (cx, cy), members in grid.items():
+ for dx in (-1, 0, 1):
+ for dy in (-1, 0, 1):
+ for j in grid.get((cx + dx, cy + dy), ()):
+ for i in members:
+ if i < j and math.dist(points[i][1], points[j][1]) <= link:
+ parent[find(i)] = find(j)
+ groups: dict[int, list[int]] = {}
+ for k in range(len(points)):
+ groups.setdefault(find(k), []).append(k)
+ total = {}
+ for tag, _at in points:
+ total[tag] = total.get(tag, 0) + 1
+ regions = []
+ for members in groups.values():
+ tags = sorted({points[k][0] for k in members})
+ if len(tags) < DETAIL_MIN_TAGS:
+ continue
+ here = {}
+ for k in members:
+ here[points[k][0]] = here.get(points[k][0], 0) + 1
+ if any(here[tag] >= total[tag] for tag in tags):
+ continue # a tag written only here: this is where it lives, not a copy
+ xs = [points[k][1][0] for k in members]
+ ys = [points[k][1][1] for k in members]
+ regions.append(
+ {
+ "box": [min(xs) - link, min(ys) - link, max(xs) + link, max(ys) + link],
+ "tags": tags,
+ "kind": "repeated",
+ "base": None,
+ }
+ )
+ families: dict[str, dict[str, list]] = {}
+ for tag, places in occurrences.items():
+ match = _FAMILY_TAG.match(tag)
+ if match:
+ families.setdefault(match["base"], {})[tag] = places
+ for base, parts in sorted(families.items()):
+ if len(parts) < DETAIL_MIN_TAGS:
+ continue
+ xs = [p[0] for places in parts.values() for p in places]
+ ys = [p[1] for places in parts.values() for p in places]
+ if math.hypot(max(xs) - min(xs), max(ys) - min(ys)) > DETAIL_FAMILY_SHARE * prep["diag"]:
+ continue
+ regions.append(
+ {
+ "box": [min(xs) - link, min(ys) - link, max(xs) + link, max(ys) + link],
+ "tags": sorted(parts),
+ "kind": "family",
+ "base": base,
+ }
+ )
+ regions.sort(key=lambda region: (region["box"][0], region["box"][1]))
+ return [{"id": f"D{i}", **region} for i, region in enumerate(regions, start=1)]
+
+
+def _distinct(places, radius: float) -> list:
+ """The places of one tag, readings closer than ``radius`` to a kept one dropped."""
+ kept: list = []
+ for at in places:
+ if all(math.dist(at, other) > radius for other in kept):
+ kept.append(at)
+ return kept
+
+
+def _majority_region(edges, boxes: dict) -> str | None:
+ """The region holding more than half a run's drawn length, by piece midpoints."""
+ total = sum(edge["length"] for edge in edges)
+ if not boxes or total <= 0.0:
+ return None
+ share: dict[str, float] = {}
+ for edge in edges:
+ mid = ((edge["a"][0] + edge["b"][0]) / 2.0, (edge["a"][1] + edge["b"][1]) / 2.0)
+ for region_id, box in boxes.items():
+ if box[0] <= mid[0] <= box[2] and box[1] <= mid[1] <= box[3]:
+ share[region_id] = share.get(region_id, 0.0) + edge["length"]
+ break
+ best = max(share.items(), key=lambda item: item[1], default=None)
+ return best[0] if best is not None and best[1] > total / 2.0 else None
+
+
def _reasons(flagged: list[dict]) -> str:
counts: dict[str, int] = {}
for run in flagged:
@@ -389,6 +539,7 @@ def pipe_rows(
vertical_allowance=0.0,
length_source="auto",
force=False,
+ exclude_regions=(),
) -> dict:
"""Takeoff rows by room x service x diameter, with the runs, checks and method behind them.
@@ -410,6 +561,15 @@ def pipe_rows(
P&ID has room faces, else the nearest room label). On the layout a run
holding several diameters shares its routed length among them in
proportion to their drawn length on the P&ID.
+
+ Detail copies: where several tags the P&ID draws twice sit close together
+ (`detail_copy_regions`), a detail of equipment may repeat its pipes. Each
+ such region is reported in ``detail_copies`` and counted like any other
+ drawing; a run with more than half its drawn length in a region named in
+ ``exclude_regions`` is left out, listed in ``control`` as
+ ``excluded_region`` and totalled in ``excluded_m``. An unknown region id is
+ refused with the ids the P&ID has. Which copy is the detail is the
+ engineer's call: the drawing does not say.
"""
options = check_pipe_options(
services=services,
@@ -442,13 +602,45 @@ def pipe_rows(
)
m_pid = pid_unit["m_per_unit"]
m_layout = layout_unit["m_per_unit"] if layout_unit else None
+ copies = detail_copy_regions(pid, prep=prep_pid)
+ known_ids = [region["id"] for region in copies]
+ left_out = [str(region_id) for region_id in (exclude_regions or ())]
+ for region_id in left_out:
+ if region_id not in known_ids:
+ raise ValueError(
+ f"exclude_regions: {region_id!r} unknown; the P&ID has "
+ f"{', '.join(known_ids) or 'no detail copy'}"
+ )
+ boxes = {region["id"]: region["box"] for region in copies}
+ for region in copies:
+ region["pipe_m"] = 0.0
+ region["excluded"] = region["id"] in left_out
rows: dict[tuple[str, str, str], dict] = {}
runs_out, control, excluded = [], [], {}
+ excluded_m = 0.0
for run in network["runs"]:
if run.service not in options["services"]:
excluded[run.service] = excluded.get(run.service, 0) + 1
continue
+ inside = _majority_region(run.edges, boxes)
+ if inside is not None:
+ region = next(r for r in copies if r["id"] == inside)
+ region["pipe_m"] += sum(edge["length"] for edge in run.edges) * m_pid
+ if inside in left_out:
+ drawn_m = sum(edge["length"] for edge in run.edges) * m_pid
+ excluded_m += drawn_m
+ control.append(
+ {
+ "run": run.id,
+ "kind": "excluded_region",
+ "service": run.service,
+ "tags": list(run.tags),
+ "schematic_m": drawn_m,
+ "reason": inside,
+ }
+ )
+ continue
ends = network["attachments"].get(run.id, [])
drawn = sum(edge["length"] for edge in run.edges)
schematic_m = drawn * m_pid
@@ -621,6 +813,17 @@ def pipe_rows(
"control rows, not in the table. length_source='pid' with force=True measures "
"every run on the P&ID instead, and says the scale is not verified."
)
+ counted_copies = [region for region in copies if not region["excluded"]]
+ if counted_copies:
+ listed = "; ".join(
+ f"{region['id']}: {', '.join(region['tags'])} ({region['pipe_m']:.1f} m)"
+ for region in counted_copies
+ )
+ warnings.append(
+ f"{len(counted_copies)} possible detail cop{'y' if len(counted_copies) == 1 else 'ies'} "
+ f"- tags drawn twice, close together - are counted like the rest ({listed}). If one "
+ "repeats pipes drawn elsewhere, pass its id in exclude_regions."
+ )
if not scale_verified:
warnings.append(
"scale_verified: false - lengths are drawn lengths on a P&ID that no scale check "
@@ -640,7 +843,9 @@ def pipe_rows(
"allowance_m": net_total * allowance,
"total_m": net_total * (1.0 + allowance),
"flagged_runs": sum(1 for r in runs_out if r["status"] == "C"),
+ "excluded_m": excluded_m,
},
+ "detail_copies": copies,
"length_source": source,
"scale": scale,
"scale_verified": scale_verified,
@@ -660,6 +865,7 @@ def pipe_rows(
"overlap_length_m": network["stats"]["overlap_length"] * m_pid,
"route": "rectilinear MST of the tag positions; each connection x first, then y",
"rooms": _rooms_mode(regions),
+ "exclude_regions": left_out,
},
}
@@ -742,14 +948,22 @@ def _inside_box(p: Pt, box) -> bool:
def cable_rows(
- pid: Snapshot, layout: Snapshot | None, *, allowance=0.20, section_rules=None
+ pid: Snapshot,
+ layout: Snapshot | None,
+ *,
+ allowance=0.20,
+ section_rules=None,
+ panel_rule="nearest",
) -> 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
+ -> CP-1), else the P&ID's; with ``panel_rule='nearest'`` (the default) a
+ load no callout places, whose tag is on the layout, takes the nearest panel
+ there and is flagged ``panel_inferred`` - ``'stated'`` leaves it open. An
+ inferred panel never makes a load part of a package. 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
@@ -763,6 +977,10 @@ def cable_rows(
"""
allowance = check_allowance(allowance)
rules = check_section_rules(section_rules)
+ if panel_rule not in PANEL_RULES:
+ raise ValueError(
+ f"panel_rule: {panel_rule!r} unknown; choose from {', '.join(PANEL_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)
@@ -825,6 +1043,8 @@ def cable_rows(
room_regions(layout, cluster, prep=prep_layout) if layout is not None else room_regions(pid)
)
+ # the panels a callout names anywhere, besides every tag with a panel prefix
+ named = {target for target, _source in panels.values()}
rows, items = [], []
for tag in sorted(set(loads) | set(panels)):
texts = sorted(loads.get(tag, []), key=lambda item: item[:2])
@@ -841,6 +1061,16 @@ def cable_rows(
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))
+ inferred = None
+ if panel is None and panel_rule == "nearest" and layout is not None:
+ here_now = positions.get(tag)
+ if here_now is not None:
+ candidates = {
+ name: p for name, p in panel_at.items() if name in named or is_panel_name(name)
+ }
+ inferred = _nearest_panel(here_now, candidates, exclude=normalize_panel(tag))
+ if inferred is not None:
+ panel, panel_source = inferred[0], "nearest"
row = {
"tag": tag,
"room": "",
@@ -874,6 +1104,16 @@ def cable_rows(
"detail": "no wiring callout names a panel for this tag",
}
)
+ elif panel_source == "nearest":
+ items.append(
+ {
+ "tag": tag,
+ "item": "panel_inferred",
+ "detail": "no wiring callout names a panel; "
+ f"{panel} is the nearest panel on the layout, "
+ f"{inferred[1] * layout_unit['m_per_unit']:.1f} m away",
+ }
+ )
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"}
@@ -918,7 +1158,9 @@ def cable_rows(
(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
+ # only a panel the drawing names makes a package; a nearest guess does not
+ known = row["panel"] and row["panel_source"] != "nearest"
+ owner = stated.get(row["panel"]) if known else None
if owner is not None and owner != row["tag"]:
row["package"] = owner
@@ -959,6 +1201,7 @@ def cable_rows(
"allowance": allowance,
"search": {"pid": pid_search, "layout": layout_search},
"section_rules": [dict(rule) for rule in rules],
+ "panel_rule": panel_rule,
"rooms": _rooms_mode(regions),
},
}
diff --git a/server.py b/server.py
index cb99e45..de4f771 100644
--- a/server.py
+++ b/server.py
@@ -11149,6 +11149,14 @@ async def pipe_takeoff(
"drawing, 0.001 on a metre one).",
),
] = None,
+ exclude_regions: Annotated[
+ list[str] | None,
+ Field(
+ default=None,
+ description="Ids of reported detail copies (detail_copies: D1, D2 ...) whose runs "
+ "repeat pipes drawn elsewhere and are left out. Omitted, every copy is counted.",
+ ),
+ ] = None,
ctx: Context = None,
) -> dict:
"""A pipe takeoff: topology from the P&ID, lengths from the layout, rows by room x service x diameter.
@@ -11201,6 +11209,7 @@ async def pipe_takeoff(
force=force,
layers=layers,
tol=tol,
+ exclude_regions=exclude_regions,
)
except (ValueError, OSError) as exc:
raise ToolError(str(exc)) from exc
@@ -11263,6 +11272,14 @@ async def cable_takeoff(
"e.g. [{'max_kw': 5.5, 'section': '5x2.5'}]. Omitted, no section is proposed.",
),
] = None,
+ panel_rule: Annotated[
+ str,
+ Field(
+ default="nearest",
+ description="A load no wiring callout places: 'nearest' takes the nearest panel on "
+ "the layout and flags the row panel_inferred; 'stated' leaves it open.",
+ ),
+ ] = "nearest",
ctx: Context = None,
) -> dict:
"""A cable takeoff: one row per electrical load, the shop's cable method made repeatable.
@@ -11271,7 +11288,9 @@ async def cable_takeoff(
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
+ and its variants, normalised to CP-1), else from the P&ID ('к CP1'); a load
+ neither places takes the nearest panel on the layout, flagged
+ `panel_inferred` (`panel_rule='stated'` leaves it open). 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
@@ -11306,6 +11325,7 @@ async def cable_takeoff(
lang=lang,
allowance=allowance,
section_rules=section_rules,
+ panel_rule=panel_rule,
)
except (ValueError, OSError) as exc:
raise ToolError(str(exc)) from exc
diff --git a/tests/test_understand_cable.py b/tests/test_understand_cable.py
index ef9f56a..931cfec 100644
--- a/tests/test_understand_cable.py
+++ b/tests/test_understand_cable.py
@@ -192,6 +192,34 @@ def test_without_a_layout_lengths_stay_empty_with_an_open_item(pair):
assert all(row["panel"] is None for row in result["rows"])
+def test_a_load_no_callout_places_takes_the_nearest_panel_flagged(pair):
+ # M82's 'Wiring to CP1' callout is gone. By default the nearest panel on the
+ # layout is taken - CP-1 at hypot(13000, 2000) = 13.2 m, not CP-M88 at 23.1 m
+ # - and the row says so; with panel_rule='stated' the panel stays open.
+ pid, _layout = pair
+ layout = snap([r for r in layout_records() if r.handle != "wM82"], 4, "layout.dxf")
+ result = cable_rows(pid, layout)
+ m82 = by_tag(result)["M82"]
+ assert (m82["panel"], m82["panel_source"]) == ("CP-1", "nearest")
+ assert m82["length_m"] * 1000.0 == pytest.approx(15000.0, abs=1e-6) # Manhattan
+ assert m82["cable_m"] == 18
+ assert {
+ "tag": "M82",
+ "item": "panel_inferred",
+ "detail": "no wiring callout names a panel; CP-1 is the nearest panel on the layout, "
+ "13.2 m away",
+ } in result["open_items"]
+ assert ("M82", "missing_panel") not in {(i["tag"], i["item"]) for i in result["open_items"]}
+ assert result["method"]["panel_rule"] == "nearest"
+ stated = cable_rows(pid, layout, panel_rule="stated")
+ assert by_tag(stated)["M82"]["panel"] is None
+ assert ("M82", "missing_panel") in {(i["tag"], i["item"]) for i in stated["open_items"]}
+ with pytest.raises(
+ ValueError, match="^panel_rule: 'closest' unknown; choose from nearest, stated"
+ ):
+ cable_rows(pid, layout, panel_rule="closest")
+
+
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")
diff --git a/tests/test_understand_describe.py b/tests/test_understand_describe.py
index 90f76b1..c229d6f 100644
--- a/tests/test_understand_describe.py
+++ b/tests/test_understand_describe.py
@@ -171,6 +171,42 @@ def test_a_room_label_inside_the_wall_faces_is_measured():
assert report["room_faces"]["unlabelled"] == 1
+def frame(handle, x0, y0, x1, y1, layer="0"):
+ return EntityRecord(
+ handle=handle,
+ type="LWPOLYLINE",
+ layer=layer,
+ space="Model",
+ points=((x0, y0), (x1, y0), (x1, y1), (x0, y1)),
+ closed=True,
+ bbox=(x0, y0, x1, y1),
+ )
+
+
+def test_a_room_drawn_as_a_frame_is_measured_by_its_frame():
+ # A P&ID draws its rooms as closed frames on no wall layer. Each numbered
+ # label takes the smallest frame around it that holds no other room label;
+ # the sheet border (both rooms) and a box drawn tight around a label (under
+ # ten text heights a side) are not rooms.
+ records = [
+ frame("B", 0.0, 0.0, 20000.0, 10000.0), # the border: both labels inside
+ frame("F1", 1000.0, 1000.0, 9000.0, 9000.0), # room 1: 8000 x 8000
+ frame("F2", 11000.0, 1000.0, 19000.0, 7000.0), # room 2: 8000 x 6000
+ frame("K", 1900.0, 7900.0, 4100.0, 8400.0), # a box around room 1's label
+ text("R1", "ROOM 1", 2000.0, 8000.0),
+ text("R2", "ROOM 2", 12000.0, 6000.0),
+ # an unnumbered mention inside room 2 ('hot room') does not deny its frame
+ text("H", "HOT ROOM", 15000.0, 3000.0),
+ ]
+ report = describe(snapshot(records))
+ rooms = {room["number"]: room for room in report["rooms"]}
+ assert rooms["1"]["source"] == rooms["2"]["source"] == "label+frame"
+ assert rooms["1"]["face"]["area"] == pytest.approx(64_000_000.0, abs=EPS)
+ assert rooms["2"]["face"]["area"] == pytest.approx(48_000_000.0, abs=EPS)
+ # a frame is weaker evidence than a face of wall layers
+ assert rooms["1"]["confidence"] == rooms["2"]["confidence"] == 0.8
+
+
def test_a_cabinet_label_naming_its_room_is_not_a_room():
# 'CONTROL CABINET / FILLING ROOM' labels the cabinet, not a room: it names
# equipment and carries no room number. A numbered room named after its
diff --git a/tests/test_understand_labels.py b/tests/test_understand_labels.py
index dc1eeca..931ddf1 100644
--- a/tests/test_understand_labels.py
+++ b/tests/test_understand_labels.py
@@ -11,6 +11,7 @@ import pytest
from engineering.understand.labels import (
CYRILLIC_LOOKALIKES,
+ is_panel_name,
normalize_panel,
parse_diameters,
parse_electrical,
@@ -212,6 +213,15 @@ def test_wiring_callouts_in_four_languages():
assert wiring_target("CP 2 panosuna") == "CP-2"
+def test_a_panel_name_is_told_by_its_letters():
+ # normalize_panel reads any letters-and-digits token; only a panel prefix
+ # (control panel, junction box, distribution board, MCC ...) makes a panel
+ assert is_panel_name("CP-1") and is_panel_name("JB-2") and is_panel_name("CP-M82")
+ assert is_panel_name("MCC-3") and is_panel_name("DB-12")
+ assert not is_panel_name("M-87") and not is_panel_name("T-4100")
+ assert not is_panel_name(None) and not is_panel_name("")
+
+
def test_the_russian_to_before_a_panel_is_a_wiring_callout():
# A Russian P&ID closes a load's text with the panel it is wired to: 'к CP1'
# ('to CP1'), or 'к щиту CP-2' ('to switchboard CP-2'). The unit кВт holds
diff --git a/tests/test_understand_takeoff.py b/tests/test_understand_takeoff.py
index 14fea34..3713e42 100644
--- a/tests/test_understand_takeoff.py
+++ b/tests/test_understand_takeoff.py
@@ -305,6 +305,69 @@ def test_without_room_faces_a_piece_goes_to_the_nearest_numbered_room():
]
+def detail_pid():
+ # three tanks spread over the plant, and a compact detail drawing of the same
+ # three tanks at x = 60 m: the detail repeats their pipes
+ return [
+ *tank("A", "T4100", 0.0, 0.0),
+ *tank("B", "T4101", 20000.0, 0.0),
+ *tank("C", "T4102", 20000.0, 20000.0),
+ rec("L1", "LINE", PIPE, ((500.0, 250.0), (20000.0, 250.0))), # 19.5 m
+ rec("L2", "LINE", PIPE, ((20250.0, 500.0), (20250.0, 20000.0))), # 19.5 m
+ *tank("a", "T4100", 60000.0, 0.0),
+ *tank("b", "T4101", 61000.0, 0.0),
+ *tank("c", "T4102", 61000.0, 1000.0),
+ rec("L3", "LINE", PIPE, ((60500.0, 250.0), (61000.0, 250.0))), # 0.5 m
+ rec("L4", "LINE", PIPE, ((61250.0, 500.0), (61250.0, 1000.0))), # 0.5 m
+ text("D1", "Ø51", (10000.0, 260.0), 50.0),
+ ]
+
+
+def test_a_detail_drawn_twice_is_reported_and_left_out_on_request():
+ drawing = snap(detail_pid(), 4, "pid.dxf")
+ network = build_network(drawing, layers=[PIPE])
+ counted = pipe_rows(network, drawing, None, services=["product"])
+ (region,) = counted["detail_copies"]
+ assert (region["id"], region["tags"]) == ("D1", ["T4100", "T4101", "T4102"])
+ assert (region["kind"], region["base"]) == ("repeated", None)
+ assert region["box"][0] > 55000.0 # the compact copy, not the spread-out plant
+ assert counted["totals"]["net_m"] == pytest.approx(40.0, abs=EPS) # 19.5 + 19.5 + 0.5 + 0.5
+ assert any("possible detail cop" in w and "D1" in w for w in counted["warnings"])
+ left_out = pipe_rows(network, drawing, None, services=["product"], exclude_regions=["D1"])
+ assert left_out["totals"]["net_m"] == pytest.approx(39.0, abs=EPS)
+ assert left_out["totals"]["excluded_m"] == pytest.approx(1.0, abs=EPS)
+ excluded = [c for c in left_out["control"] if c["kind"] == "excluded_region"]
+ assert {c["reason"] for c in excluded} == {"D1"} and len(excluded) == 2
+ with pytest.raises(ValueError, match="^exclude_regions: 'D9' unknown; the P&ID has D1"):
+ pipe_rows(network, drawing, None, services=["product"], exclude_regions=["D9"])
+
+
+def test_a_package_of_one_equipments_parts_is_reported_as_a_region():
+ # A skid drawn in detail: P200's own parts P200A / P200B / P200C sit close
+ # together at x = 60 m, far from the plant's tanks. Their tags are written
+ # once each - no copy - but the family is one package whose internal pipes
+ # a site takeoff may leave to its vendor: reported, left out on request.
+ records = [
+ *tank("A", "T4100", 0.0, 0.0),
+ *tank("B", "T4101", 20000.0, 0.0),
+ rec("L1", "LINE", PIPE, ((500.0, 250.0), (20000.0, 250.0))), # 19.5 m
+ *tank("p", "P200A", 60000.0, 0.0),
+ *tank("q", "P200B", 61000.0, 0.0),
+ *tank("r", "P200C", 61000.0, 1000.0),
+ rec("L3", "LINE", PIPE, ((60500.0, 250.0), (61000.0, 250.0))), # 0.5 m
+ rec("L4", "LINE", PIPE, ((61250.0, 500.0), (61250.0, 1000.0))), # 0.5 m
+ ]
+ drawing = snap(records, 4, "pid.dxf")
+ network = build_network(drawing, layers=[PIPE])
+ counted = pipe_rows(network, drawing, None, services=["product"])
+ (region,) = counted["detail_copies"]
+ assert (region["kind"], region["base"]) == ("family", "P200")
+ assert region["tags"] == ["P200A", "P200B", "P200C"]
+ left_out = pipe_rows(network, drawing, None, services=["product"], exclude_regions=["D1"])
+ assert left_out["totals"]["net_m"] == pytest.approx(19.5, abs=EPS)
+ assert left_out["totals"]["excluded_m"] == pytest.approx(1.0, abs=EPS)
+
+
def test_other_services_are_left_out_unless_asked(pair):
pid, layout, network = pair
assert DEFAULT_SERVICES == ("product", "cip_supply", "cip_return")