fix(understand): what the field test found — P&ID lengths need no equipment, caret notation, numbered rooms
Run on a real P&ID pair (kept outside the repository), the pipe takeoff wrote
an empty table. Three causes, each fixed with a test that failed first:
- Measured on the P&ID, a run still needed equipment tags at both ends,
which only routing on the layout needs. Few pipes on a field P&ID end on
a tagged outline. On the P&ID each piece now goes to the room it lies in:
cut at the room outlines when the P&ID has room faces, else to the
nearest room label. Diameters are kept piece by piece. On the layout a
takeoff that could not route every run now says how many, why, the drawn
length left to the control rows, and that length_source='pid' with
force=True measures on the P&ID instead.
- A TEXT stores its line breaks in caret notation ('6^JROOM'), which
plain() left as it was, so a room label's number read as '6^J...'.
plain() now decodes ^J, ^M, ^I and '^ ', leaving MTEXT stacks
('\S1^2;') alone.
- Labels that only mention a room were read as rooms: a control cabinet
naming the room it serves, or a note pointing into a room. An unnumbered
label that names equipment is no room. Without room faces, when the
drawing numbers its rooms, only numbered labels take pieces.
On the field pair the takeoff now fills its table, and every room key is a
real room number. The comparison with the hand-made workbook stays with
the user's files. README collected tests: 4981. The suite passes (4980, 1
xfailed).
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
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
307ccc799f
commit
ac32700ba8
@@ -27,7 +27,7 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
|
||||
|
||||
</div>
|
||||
|
||||
> **v1.5 release snapshot:** 247 tools · 8 resources · 5 prompt templates · 4960 collected tests.
|
||||
> **v1.5 release snapshot:** 247 tools · 8 resources · 5 prompt templates · 4981 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.
|
||||
|
||||
|
||||
@@ -35,7 +35,7 @@ from engineering.understand.clusters import find_clusters
|
||||
from engineering.understand.labels import parse_tag, wiring_target
|
||||
from engineering.understand.snapshot import EntityRecord, Snapshot, length_of
|
||||
from engineering.understand.units import infer_units, robust_extents, wall_segments
|
||||
from engineering.understand.vocab import classify_layer, room_label
|
||||
from engineering.understand.vocab import classify_layer, equipment_kind, room_label
|
||||
|
||||
__all__ = [
|
||||
"describe",
|
||||
@@ -186,6 +186,10 @@ def find_rooms(records: Sequence[EntityRecord], *, tol: float, membership: dict)
|
||||
if rec.type in _TEXT_TYPES and rec.text:
|
||||
parsed = room_label(rec.text)
|
||||
at = _anchor(rec)
|
||||
# an unnumbered label that names equipment ('control cabinet - filling
|
||||
# room') labels the equipment; a numbered 'pump room 3' is a room
|
||||
if parsed and parsed["number"] is None and equipment_kind(rec.text):
|
||||
continue
|
||||
if parsed and at is not None:
|
||||
labels.append((rec, parsed, at))
|
||||
segments, bulged = wall_segments(records)
|
||||
|
||||
@@ -65,6 +65,10 @@ _TOGGLE_CODES = frozenset("LlOoKkN")
|
||||
|
||||
_UNICODE_ESCAPE = re.compile(r"\\[Uu]\+([0-9A-Fa-f]{4})")
|
||||
_PERCENT_CODE = re.compile(r"%%(\d{3}|[cdpCDP]|[uoUOkK])")
|
||||
#: Caret notation, how a TEXT or ATTRIB stores a control character: ^J a line
|
||||
#: break, ^M a carriage return (read as a break), ^I a tab, and "^ " a caret.
|
||||
_CARET = re.compile(r"\^([JMI ])")
|
||||
_CARET_CHARS = {"J": "\n", "M": "\n", "I": "\t", " ": "^"}
|
||||
|
||||
#: Every spelling of the diameter sign met in drawings: Ø (U+00D8), ø (U+00F8),
|
||||
#: ∅ (U+2205, AutoCAD's listed diameter symbol), ⌀ (U+2300,
|
||||
@@ -138,8 +142,9 @@ def plain(text: str | None) -> str:
|
||||
|
||||
``\\U+XXXX`` becomes its character; ``%%c`` / ``%%d`` / ``%%p`` become Ø / ° /
|
||||
±, ``%%nnn`` the character with that code, ``%%%`` a percent sign, and the
|
||||
``%%u`` / ``%%o`` / ``%%k`` toggles disappear. In MTEXT, ``\\P`` is a line
|
||||
break, ``\\~`` a space, ``\\S a^b;`` / ``\\S a/b;`` the stacked ``a/b``,
|
||||
``%%u`` / ``%%o`` / ``%%k`` toggles disappear, and the caret notation of a
|
||||
TEXT (``^J`` a line break, ``^I`` a tab, ``^ `` a caret) is decoded. In
|
||||
MTEXT, ``\\P`` is a line break, ``\\~`` a space, ``\\S a^b;`` / ``\\S a/b;`` the stacked ``a/b``,
|
||||
``\\\\`` ``\\{`` ``\\}`` the literal characters; every argument code
|
||||
(``\\f...;``, ``\\H...;``, ``\\C...;`` ...) and every toggle is dropped, and so
|
||||
are the grouping braces.
|
||||
@@ -147,6 +152,7 @@ def plain(text: str | None) -> str:
|
||||
if not text:
|
||||
return ""
|
||||
source = _UNICODE_ESCAPE.sub(lambda m: chr(int(m.group(1), 16)), str(text))
|
||||
source = _CARET.sub(lambda m: _CARET_CHARS[m.group(1)], source)
|
||||
source = source.replace("%%%", "\x00")
|
||||
|
||||
def percent(match: re.Match) -> str:
|
||||
|
||||
@@ -226,9 +226,13 @@ def room_regions(snap: Snapshot, cluster=None, *, prep: dict | None = None) -> d
|
||||
def key(row: dict) -> str:
|
||||
return str(row["number"] or row["name"] or row["text"])
|
||||
|
||||
# without a face a piece goes to the nearest label: when the drawing numbers
|
||||
# its rooms, an unnumbered mention ('to room storage') is a note, not a room
|
||||
numbered = [row for row in found["rooms"] if row["number"]]
|
||||
candidates = numbered or found["rooms"]
|
||||
return {
|
||||
"faces": [(key(row), face) for row, face in found["labelled"]],
|
||||
"labels": sorted((key(row), (row["at"][0], row["at"][1])) for row in found["rooms"]),
|
||||
"labels": sorted((key(row), (row["at"][0], row["at"][1])) for row in candidates),
|
||||
}
|
||||
|
||||
|
||||
@@ -322,6 +326,58 @@ def _choose_source(requested: str, layout, scale: dict, force: bool) -> tuple[st
|
||||
return "layout", f"auto_{verdict}"
|
||||
|
||||
|
||||
def _row(rows: dict, room: str, service: str, diameter: str) -> dict:
|
||||
return rows.setdefault(
|
||||
(room, service, diameter),
|
||||
{
|
||||
"room": room,
|
||||
"service": service,
|
||||
"diameter": diameter,
|
||||
"direct_m": 0.0,
|
||||
"continuity_m": 0.0,
|
||||
"unassigned_m": 0.0,
|
||||
"net_m": 0.0,
|
||||
"schematic_m": 0.0,
|
||||
"status": "A",
|
||||
"runs": [],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def _pid_cells(edges, regions: dict, m_per_unit: float) -> dict:
|
||||
"""{(room, diameter): {net, direct, continuity, unassigned}} in metres, piece by piece.
|
||||
|
||||
With room faces a piece is cut at the outlines (`apportion` on its chord,
|
||||
scaled to the piece's true length, so a bend keeps its arc length); without
|
||||
faces the whole piece goes to the room label nearest its midpoint.
|
||||
"""
|
||||
cells: dict[tuple[str, str], dict] = {}
|
||||
for edge in edges:
|
||||
a, b = edge["a"], edge["b"]
|
||||
metres = edge["length"] * m_per_unit
|
||||
pieces = apportion([(a, b)], regions) if regions["faces"] else {}
|
||||
chord = sum(pieces.values())
|
||||
if chord > 0.0:
|
||||
shares = {room: value / chord for room, value in pieces.items()}
|
||||
else:
|
||||
shares = {room_at(regions, ((a[0] + b[0]) / 2.0, (a[1] + b[1]) / 2.0)): 1.0}
|
||||
diameter = edge["diameter"] or UNASSIGNED
|
||||
for room, share in shares.items():
|
||||
cell = cells.setdefault(
|
||||
(room, diameter), {"net": 0.0, "direct": 0.0, "continuity": 0.0, "unassigned": 0.0}
|
||||
)
|
||||
cell["net"] += metres * share
|
||||
cell[edge["diameter_source"]] += metres * share
|
||||
return cells
|
||||
|
||||
|
||||
def _reasons(flagged: list[dict]) -> str:
|
||||
counts: dict[str, int] = {}
|
||||
for run in flagged:
|
||||
counts[run["reason"]] = counts.get(run["reason"], 0) + 1
|
||||
return ", ".join(f"{reason} {n}" for reason, n in sorted(counts.items()))
|
||||
|
||||
|
||||
def pipe_rows(
|
||||
network,
|
||||
pid: Snapshot,
|
||||
@@ -343,12 +399,16 @@ def pipe_rows(
|
||||
`length_source='pid'` on a P&ID the scale check calls schematic unless
|
||||
`force=True` (the refusal quotes the statistics).
|
||||
|
||||
Status per run: **A** every end on a tag placed on the layout and every
|
||||
piece's diameter read directly; **B** a diameter carried by continuity or
|
||||
unassigned; **C** not routable - an end on no equipment, fewer than two
|
||||
tags, a tag missing from (or written twice on) the layout. A C run goes to
|
||||
`control` with its schematic length only, is counted in the totals and is
|
||||
flagged. A run holding several diameters shares its length among them in
|
||||
Status per run: **A** every piece's diameter read directly; **B** a
|
||||
diameter carried by continuity or unassigned; **C**, on the layout only,
|
||||
not routable - an end on no equipment, fewer than two tags, a tag missing
|
||||
from (or written twice on) the layout. A C run goes to `control` with its
|
||||
schematic length only, is counted in the totals and is flagged, and a
|
||||
warning says how much could not be routed. Measured on the P&ID a run needs
|
||||
no equipment at its ends - the drawn geometry is the length - and each
|
||||
piece goes to the P&ID room it lies in (cut at the room outlines when the
|
||||
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.
|
||||
"""
|
||||
options = check_pipe_options(
|
||||
@@ -401,16 +461,19 @@ def pipe_rows(
|
||||
part["drawn"] += edge["length"]
|
||||
part[edge["diameter_source"]] += edge["length"]
|
||||
why = None
|
||||
if any(end["tag"] is None for end in ends) or not ends:
|
||||
why = "end_not_on_equipment"
|
||||
elif len(run.tags) < 2:
|
||||
why = "single_tag"
|
||||
elif source == "layout":
|
||||
missing = [tag for tag in run.tags if tag not in positions]
|
||||
if missing:
|
||||
why = (
|
||||
"tag_ambiguous" if any(t in ambiguous for t in missing) else "tag_not_on_layout"
|
||||
)
|
||||
if source == "layout":
|
||||
if any(end["tag"] is None for end in ends) or not ends:
|
||||
why = "end_not_on_equipment"
|
||||
elif len(run.tags) < 2:
|
||||
why = "single_tag"
|
||||
else:
|
||||
missing = [tag for tag in run.tags if tag not in positions]
|
||||
if missing:
|
||||
why = (
|
||||
"tag_ambiguous"
|
||||
if any(t in ambiguous for t in missing)
|
||||
else "tag_not_on_layout"
|
||||
)
|
||||
detail = {
|
||||
"id": run.id,
|
||||
"service": run.service,
|
||||
@@ -468,15 +531,16 @@ def pipe_rows(
|
||||
detail["vertical_m"] = vertical * len(tree)
|
||||
length_m = detail["layout_m"] + detail["vertical_m"]
|
||||
else:
|
||||
if layout is not None and scale.get("verdict") == "to_scale":
|
||||
length_m = schematic_m / scale["factor"]
|
||||
else:
|
||||
length_m = schematic_m
|
||||
centre = (
|
||||
sum(end["at"][0] for end in ends) / len(ends),
|
||||
sum(end["at"][1] for end in ends) / len(ends),
|
||||
factor = (
|
||||
scale["factor"]
|
||||
if layout is not None and scale.get("verdict") == "to_scale"
|
||||
else 1.0
|
||||
)
|
||||
rooms = {room_at(regions, centre): length_m}
|
||||
length_m = schematic_m / factor
|
||||
cells = _pid_cells(run.edges, regions, m_pid / factor)
|
||||
rooms = {}
|
||||
for (room, _diameter), cell in cells.items():
|
||||
rooms[room] = rooms.get(room, 0.0) + cell["net"]
|
||||
detail["length_m"] = length_m
|
||||
detail["rooms"] = rooms
|
||||
status = (
|
||||
@@ -485,25 +549,25 @@ def pipe_rows(
|
||||
else "B"
|
||||
)
|
||||
detail["status"] = status
|
||||
if source == "pid":
|
||||
# measured piece by piece: each cell is one room and one diameter
|
||||
for (room, diameter), cell in cells.items():
|
||||
row = _row(rows, room, run.service, diameter)
|
||||
row["net_m"] += cell["net"]
|
||||
row["direct_m"] += cell["direct"]
|
||||
row["continuity_m"] += cell["continuity"]
|
||||
row["unassigned_m"] += cell["unassigned"]
|
||||
row["schematic_m"] += cell["net"] * factor
|
||||
if cell["continuity"] or cell["unassigned"]:
|
||||
row["status"] = "B"
|
||||
if run.id not in row["runs"]:
|
||||
row["runs"].append(run.id)
|
||||
continue
|
||||
for diameter, part in parts.items():
|
||||
share = part["drawn"] / drawn if drawn else 0.0
|
||||
for room, room_m in rooms.items():
|
||||
x = room_m * share
|
||||
row = rows.setdefault(
|
||||
(room, run.service, diameter),
|
||||
{
|
||||
"room": room,
|
||||
"service": run.service,
|
||||
"diameter": diameter,
|
||||
"direct_m": 0.0,
|
||||
"continuity_m": 0.0,
|
||||
"unassigned_m": 0.0,
|
||||
"net_m": 0.0,
|
||||
"schematic_m": 0.0,
|
||||
"status": "A",
|
||||
"runs": [],
|
||||
},
|
||||
)
|
||||
row = _row(rows, room, run.service, diameter)
|
||||
row["net_m"] += x
|
||||
row["direct_m"] += x * part["direct"] / part["drawn"]
|
||||
row["continuity_m"] += x * part["continuity"] / part["drawn"]
|
||||
@@ -549,6 +613,14 @@ def pipe_rows(
|
||||
f"({network['stats']['tol']:g} drawing units) collapsed to a point and are not "
|
||||
"counted"
|
||||
)
|
||||
flagged = [r for r in runs_out if r["status"] == "C"]
|
||||
if source == "layout" and flagged:
|
||||
warnings.append(
|
||||
f"{len(flagged)} of {len(runs_out)} runs could not be routed on the layout "
|
||||
f"({_reasons(flagged)}); their {unroutable:.1f} m drawn on the P&ID are in the "
|
||||
"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."
|
||||
)
|
||||
if not scale_verified:
|
||||
warnings.append(
|
||||
"scale_verified: false - lengths are drawn lengths on a P&ID that no scale check "
|
||||
|
||||
@@ -171,6 +171,18 @@ def test_a_room_label_inside_the_wall_faces_is_measured():
|
||||
assert report["room_faces"]["unlabelled"] == 1
|
||||
|
||||
|
||||
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
|
||||
# equipment ('PUMP ROOM 3') is still a room.
|
||||
records = [
|
||||
text("C", "CONTROL CABINET FILLING ROOM", 1000.0, 1000.0),
|
||||
text("P", "PUMP ROOM 3", 5000.0, 1000.0),
|
||||
]
|
||||
report = describe(snapshot(records))
|
||||
assert [(room["number"], room["name"]) for room in report["rooms"]] == [("3", "PUMP")]
|
||||
|
||||
|
||||
def test_an_attribute_and_a_text_of_one_tag_are_one_occurrence_and_a_copy_is_two():
|
||||
assert parse_tag("T4100") == "T4100"
|
||||
records = [
|
||||
|
||||
@@ -31,6 +31,16 @@ def test_plain_decodes_the_percent_codes():
|
||||
assert plain("%%065") == "A"
|
||||
|
||||
|
||||
def test_plain_decodes_the_caret_notation_of_a_text_entity():
|
||||
# A TEXT or ATTRIB stores its control characters in caret notation: ^J a line
|
||||
# break, ^I a tab, "^ " a caret. Left undecoded, "6^JROOM" is one word and a
|
||||
# room label's number reads as "6^JROOM".
|
||||
assert plain("6^JROOM") == "6\nROOM"
|
||||
assert plain("A^IB") == "A\tB"
|
||||
assert plain("x^ y") == "x^y"
|
||||
assert plain("\\S1^2;") == "1/2" # an MTEXT stack keeps its separator
|
||||
|
||||
|
||||
def test_plain_decodes_a_unicode_escape():
|
||||
assert plain("\\U+2205 51") == "\u2205 51"
|
||||
assert plain("\\u+00D8") == "\u00d8"
|
||||
|
||||
@@ -240,6 +240,71 @@ def test_a_run_with_a_free_end_is_c(pair):
|
||||
assert reasons == {"end_not_on_equipment"}
|
||||
|
||||
|
||||
def test_a_layout_takeoff_says_how_much_it_could_not_route(pair):
|
||||
# the free-ended run is not routed: the warning names the count, the drawn
|
||||
# length it leaves to the control rows and the P&ID alternative
|
||||
pid, layout, _network = pair
|
||||
records = pid_records() + [rec("L3", "LINE", PIPE, ((0.0, 600.0), (0.0, 1600.0)))]
|
||||
drawing = snap(records, 4, "pid.dxf")
|
||||
result = pipe_rows(
|
||||
build_network(drawing, layers=[PIPE]), drawing, layout, length_source="layout"
|
||||
)
|
||||
(warning,) = [w for w in result["warnings"] if "could not be routed" in w]
|
||||
assert warning.startswith("1 of 3 runs could not be routed on the layout")
|
||||
assert "1.0 m drawn on the P&ID" in warning # 1000 mm
|
||||
assert "length_source='pid'" in warning
|
||||
|
||||
|
||||
def test_pid_lengths_need_no_equipment_at_the_ends():
|
||||
# A product line that ends in the open - no tag, no equipment - still has a
|
||||
# drawn length: measured on the P&ID it is a row, and each piece goes to the
|
||||
# P&ID room it lies in (the wall at x = 4000 cuts the second line).
|
||||
walls = [
|
||||
rec(
|
||||
"W",
|
||||
"LWPOLYLINE",
|
||||
WALLS,
|
||||
((0, 0), (8000, 0), (8000, 4000), (0, 4000)),
|
||||
closed=True,
|
||||
bbox=(0.0, 0.0, 8000.0, 4000.0),
|
||||
),
|
||||
rec("X", "LINE", WALLS, ((4000, 0), (4000, 4000))),
|
||||
text("R1", "ROOM 101", (1000.0, 3000.0), 250.0),
|
||||
text("R2", "ROOM 102", (5000.0, 3000.0), 250.0),
|
||||
]
|
||||
pipes = [
|
||||
rec("p1", "LINE", PIPE, ((1000.0, 1000.0), (3000.0, 1000.0))), # 2000, room 101
|
||||
rec("p2", "LINE", PIPE, ((3000.0, 1000.0), (6000.0, 1000.0))), # 1000 + 2000
|
||||
text("d", "Ø51", (2000.0, 1100.0), 50.0),
|
||||
]
|
||||
drawing = snap(walls + pipes, 4, "pid.dxf")
|
||||
result = pipe_rows(build_network(drawing, layers=[PIPE]), drawing, None, services=["product"])
|
||||
rows = {(row["room"], row["diameter"]): row for row in result["rows"]}
|
||||
assert rows[(room_key("ROOM 101"), "Ø51")]["net_m"] == pytest.approx(3.0, abs=EPS)
|
||||
assert rows[(room_key("ROOM 102"), "Ø51")]["net_m"] == pytest.approx(2.0, abs=EPS)
|
||||
assert result["totals"]["unroutable_m"] == 0.0
|
||||
(run,) = result["runs"]
|
||||
assert run["tags"] == [] and run["status"] in ("A", "B")
|
||||
|
||||
|
||||
def test_without_room_faces_a_piece_goes_to_the_nearest_numbered_room():
|
||||
# No walls: pieces go to the nearest room label. 'TO ROOM STORAGE' is a note
|
||||
# pointing at a room, not a room: when the drawing numbers its rooms, only
|
||||
# the numbered labels are candidates, so the line at x 8500-9500 is room 102.
|
||||
records = [
|
||||
text("R1", "ROOM 101", (0.0, 3000.0), 250.0),
|
||||
text("R2", "ROOM 102", (12000.0, 3000.0), 250.0),
|
||||
text("N", "TO ROOM STORAGE", (9000.0, 1200.0), 250.0),
|
||||
rec("p", "LINE", PIPE, ((8500.0, 1000.0), (9500.0, 1000.0))),
|
||||
text("d", "Ø51", (9000.0, 1100.0), 50.0),
|
||||
]
|
||||
drawing = snap(records, 4, "pid.dxf")
|
||||
result = pipe_rows(build_network(drawing, layers=[PIPE]), drawing, None, services=["product"])
|
||||
assert [(row["room"], row["net_m"]) for row in result["rows"]] == [
|
||||
(room_key("ROOM 102"), 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")
|
||||
|
||||
Reference in New Issue
Block a user