diff --git a/engineering/pid/graph.py b/engineering/pid/graph.py index 1e52a28..6e570a2 100644 --- a/engineering/pid/graph.py +++ b/engineering/pid/graph.py @@ -12,7 +12,7 @@ import re from dataclasses import dataclass, field from typing import TYPE_CHECKING -from .lines import LAYER_TO_CLASS, PID_LINE_LAYERS +from .lines import LAYER_TO_CLASS, LINE_CLASSES, PID_LINE_LAYERS from .symbols import Port, transform_port from .tags import parse_tag from .xdata import read_payload @@ -55,6 +55,15 @@ CONFIDENCE = {"catalog": 1.0, "xdata": 0.95, "heuristic": 0.6, "inferred": 0.3} _PAGE = 1000 +def _is_signal_class(line_class: str | None) -> bool: + """True for a class that is a signal in the ISA-5.1 sense (`signal_unknown` + is the layer-classified one that LINE_CLASSES does not list).""" + if line_class == "signal_unknown": + return True + cls = LINE_CLASSES.get(line_class) if line_class else None + return bool(cls and cls.kind == "signal") + + @dataclass class _Node: id: str @@ -684,7 +693,15 @@ async def build_graph( if found: node.tag, node.tag_source = found, "text" for edge in edges.values(): - if edge.line_number: + if edge.line_number or edge.payload or _is_signal_class(edge.line_class): + # The label search is a fallback for a line that carries no + # record of its own. An engine-authored line (payload) is that + # record — its `number: null` is the answer, not an invitation to + # adopt the nearest pipe's label. A signal line never has a pipe + # line number (ISA-5.1), and LINE_NUMBER_RE is the pipe grammar, + # so any match near it belongs to a neighbouring pipe: the + # ordinary transmitter-to-controller run 12 mm above a header + # would otherwise be listed under the header's number. continue a, b = max(zip(edge.vertices, edge.vertices[1:], strict=False), key=lambda s: math.dist(*s)) found = _text_near( diff --git a/tests/test_pid_deliverables.py b/tests/test_pid_deliverables.py index bab8b75..a6c721f 100644 --- a/tests/test_pid_deliverables.py +++ b/tests/test_pid_deliverables.py @@ -137,6 +137,30 @@ async def test_line_list_rows(backend): assert hand_drawn["to_tag"] is None and hand_drawn["number_source"] is None +async def test_line_list_keeps_a_signal_line_unnumbered_beside_a_labelled_pipe(backend): + """The line list prints what the reader publishes, and the reader must not + hand a 12 mm-parallel signal run the header's label (its own XDATA says + ``number: null``).""" + proc = await draw_line( + backend, {"x": 20, "y": 100}, {"x": 200, "y": 100}, size="100", service="P" + ) + ft = await place_symbol( + backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary" + ) + fic = await place_symbol( + backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary" + ) + sig = await draw_line(backend, {"handle": ft["handle"]}, {"handle": fic["handle"]}, "electric") + rows = {r["handle"]: r for r in line_list(await build_graph(backend))} + assert (rows[proc["handle"]]["line_number"], rows[proc["handle"]]["number_source"]) == ( + "100-P-1", + "xdata", + ) + signal = rows[sig["handle"]] + assert (signal["line_number"], signal["number_source"]) == (None, None) + assert (signal["from_tag"], signal["to_tag"]) == ("FT-1", "FIC-1") + + async def test_equipment_list_rows(backend): graph = await _pid(backend) rows = equipment_list(graph) diff --git a/tests/test_pid_graph.py b/tests/test_pid_graph.py index 160160d..1786c5d 100644 --- a/tests/test_pid_graph.py +++ b/tests/test_pid_graph.py @@ -289,6 +289,71 @@ async def test_line_number_and_tags_from_nearby_text(backend): assert edge["line_number"] == '6"-P-1234-CS1' and edge["number_source"] == "label" +async def test_a_signal_line_next_to_a_pipe_label_never_adopts_the_pipes_number(backend): + """The label search is a fallback for a line with no record of its own. + + A transmitter-to-controller run 12 mm above a labelled header (the + ordinary spacing on a dense sheet) is well inside ``label_search``; its + own payload says ``number: null`` and a signal line has no pipe line + number, so the reader must report None — not the header's number under + ``number_source="label"``. + """ + proc = await draw_line( + backend, {"x": 20, "y": 100}, {"x": 200, "y": 100}, size="100", service="P" + ) + assert proc["line_number"] == "100-P-1" + ft = await place_symbol( + backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary" + ) + fic = await place_symbol( + backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary" + ) + sig = await draw_line(backend, {"handle": ft["handle"]}, {"handle": fic["handle"]}, "electric") + assert sig["line_number"] is None + graph = await build_graph(backend) + by_handle = {e["id"]: e for e in graph["edges"]} + pipe, signal = by_handle[proc["handle"]], by_handle[sig["handle"]] + assert (pipe["line_number"], pipe["number_source"]) == ("100-P-1", "xdata") + assert (signal["line_number"], signal["number_source"]) == (None, None) + assert signal["from"]["node"] == ft["handle"] and signal["to"]["node"] == fic["handle"] + + +async def test_a_verbatim_number_on_a_signal_line_is_still_read_back(backend): + ft = await place_symbol( + backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary" + ) + fic = await place_symbol( + backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary" + ) + sig = await draw_line( + backend, + {"handle": ft["handle"]}, + {"handle": fic["handle"]}, + "electric", + line_number="SIG-7", + ) + graph = await build_graph(backend) + edge = next(e for e in graph["edges"] if e["id"] == sig["handle"]) + assert (edge["line_number"], edge["number_source"]) == ("SIG-7", "xdata") + + +async def test_a_foreign_signal_line_next_to_a_pipe_label_stays_unnumbered(backend): + """A layer-classified signal line (no payload) is a signal all the same: + LINE_NUMBER_RE is the pipe grammar, so a match near it is a neighbour's. + The pipe on the same sheet still takes its label.""" + await backend.entity_create_line(0, 0, 100, 0, layer="PROCESS-PIPING-MAIN") + await backend.entity_create_text('6"-P-1234-CS1', 40, 2, 2.5, layer="PROCESS-LINE-TEXT") + await backend.entity_create_line(0, 6, 100, 6, layer="ELECTRICAL-LINE") + await backend.entity_create_line(0, 10, 100, 10, layer="INSTRUMENT-LINE-SIGNAL") + graph = await build_graph(backend) + by_class = {e["line_class"]: e for e in graph["edges"]} + assert set(by_class) == {"process_major", "electric", "signal_unknown"} + pipe = by_class["process_major"] + assert (pipe["line_number"], pipe["number_source"]) == ('6"-P-1234-CS1', "label") + for cls in ("electric", "signal_unknown"): + assert (by_class[cls]["line_number"], by_class[cls]["number_source"]) == (None, None), cls + + async def test_scope_all_walks_every_layout_and_restores_the_current_one(backend): await place_symbol(backend, "gate", 0, 0, tag="HV-1") await backend.layout_create("SHEET-2")