diff --git a/README.md b/README.md index 73c44ca..e6f0480 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 · 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. diff --git a/engineering/understand/describe.py b/engineering/understand/describe.py index f4d359b..be1b408 100644 --- a/engineering/understand/describe.py +++ b/engineering/understand/describe.py @@ -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) diff --git a/engineering/understand/labels.py b/engineering/understand/labels.py index bc03faa..3704a87 100644 --- a/engineering/understand/labels.py +++ b/engineering/understand/labels.py @@ -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: diff --git a/engineering/understand/takeoff.py b/engineering/understand/takeoff.py index 3854a91..e3e3b2f 100644 --- a/engineering/understand/takeoff.py +++ b/engineering/understand/takeoff.py @@ -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 " diff --git a/tests/test_understand_describe.py b/tests/test_understand_describe.py index 82e5542..90f76b1 100644 --- a/tests/test_understand_describe.py +++ b/tests/test_understand_describe.py @@ -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 = [ diff --git a/tests/test_understand_labels.py b/tests/test_understand_labels.py index b159feb..1faeed8 100644 --- a/tests/test_understand_labels.py +++ b/tests/test_understand_labels.py @@ -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" diff --git a/tests/test_understand_takeoff.py b/tests/test_understand_takeoff.py index 2f5105c..14fea34 100644 --- a/tests/test_understand_takeoff.py +++ b/tests/test_understand_takeoff.py @@ -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")