From 346eac4ac6ca248770d3cbab0cb413c455a01610 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Wed, 16 Sep 2026 11:02:11 +0300 Subject: [PATCH] fix(task-15): instrument connected_to follows junction ends and never names its own wire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit instrument_index only resolved {"node"} ends of a signal line, so a transmitter tapping the middle of a pipe (a {"junction"} end) reported connected_to="" and an instrument on untagged equipment fell back to the signal line's own number — which Task 13 always stamps, so the handle fallback was unreachable and every untagged mounting was labelled with its wire. Spec 9.5's "line numbers" clause is exactly the junction case. _far_ends now names the far node's tag (its handle when untagged) and, at a junction, the line number of every line the junction joins (that line's handle when unnumbered). Two tests exercise both mountings. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- engineering/pid/deliverables.py | 32 ++++++++++++++++++++--- tests/test_pid_deliverables.py | 46 +++++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 4 deletions(-) diff --git a/engineering/pid/deliverables.py b/engineering/pid/deliverables.py index a1446b8..a739b79 100644 --- a/engineering/pid/deliverables.py +++ b/engineering/pid/deliverables.py @@ -54,9 +54,36 @@ def _bubble_part(node: dict, index: int) -> str | None: return "_".join(parts[3:]).lower() if len(parts) > 3 else None +def _far_ends(node_id: str, edge: dict, nodes: dict, edges: dict, junctions: dict) -> list[str]: + """What the other end of an instrument's signal line reaches (spec 9.5). + + A node end names that node's tag, or its handle when it has none. A + junction end (a tap into the middle of a pipe) names the line number of + every line the junction joins, or that line's handle when it is + unnumbered. The signal line's own number is never an answer: Task 13 + stamps one on every line, so falling back to it would label every + untagged mounting with the wire instead of the thing it is mounted on. + """ + names: list[str] = [] + for ref in (edge["from"], edge["to"]): + if not ref: + continue # dangling end — nothing to name + if "node" in ref: + if ref["node"] != node_id: + other = nodes[ref["node"]] + names.append(other.get("tag") or other["id"]) + elif "junction" in ref: + for other_id in junctions[ref["junction"]]["edges"]: + if other_id != edge["id"]: + other = edges.get(other_id) or {} + names.append(other.get("line_number") or other_id) + return names + + def instrument_index(graph: dict) -> list[dict]: nodes = _node_map(graph) edges = {e["id"]: e for e in graph["edges"]} + junctions = {j["id"]: j for j in graph.get("junctions", [])} rows = [] for node in graph["nodes"]: if node["kind"] != "instrument": @@ -68,10 +95,7 @@ def instrument_index(graph: dict) -> list[dict]: for edge_id in port["edges"]: edge = edges[edge_id] signal_lines += 1 - for ref in (edge["from"], edge["to"]): - if ref and "node" in ref and ref["node"] != node["id"]: - other = nodes[ref["node"]] - connected.append(other.get("tag") or edge.get("line_number") or other["id"]) + connected.extend(_far_ends(node["id"], edge, nodes, edges, junctions)) rows.append( { "tag": node.get("tag"), diff --git a/tests/test_pid_deliverables.py b/tests/test_pid_deliverables.py index 7f5a28b..bb28a58 100644 --- a/tests/test_pid_deliverables.py +++ b/tests/test_pid_deliverables.py @@ -166,3 +166,49 @@ async def test_csv_is_written_inside_allowed_paths(backend, tmp_path, monkeypatc with pytest.raises(ToolError, match="not inside any allowed directory"): await deliverable(backend, "line_list", csv_path=str(outside)) assert not outside.exists(), "a refused path is never written" + + +async def test_instrument_tapping_a_pipe_reports_the_pipe_line_number(backend): + # A transmitter whose signal line lands on the middle of a process line + # (not on a port) — the far end is a junction, and spec 9.5 says the row + # names the line number of what the junction joins. + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-101") + vessel = await place_symbol(backend, "vertical_vessel", 300, 120, tag="V-201") + pipe = await draw_line( + backend, + {"handle": pump["handle"], "port": "discharge"}, + {"handle": vessel["handle"], "port": "N3"}, + size="100", + service="P", + spec="CS1", + ) + pt = await place_symbol(backend, "instrument", 193.5, 168, tag="PT-101", location="field") + await draw_line( + backend, {"handle": pt["handle"]}, {"x": 193.5, "y": 128}, line_class="pneumatic" + ) + graph = await build_graph(backend) + signal = next(e for e in graph["edges"] if e["line_class"] == "pneumatic") + assert "junction" in (signal["to"] or {}), "the fixture must produce a junction end" + rows = instrument_index(graph) + row = next(r for r in rows if r["tag"] == "PT-101") + assert row["connected_to"] == pipe["line_number"] == "100-P-1-CS1" + assert row["signal_lines"] == 1 + + +async def test_instrument_on_untagged_equipment_reports_its_handle(backend): + # The vessel has no tag, so the only honest identity is its handle — never + # the signal line's own number (which Task 13 always stamps, so a fallback + # to it would silently mislabel every untagged mounting). + vessel = await place_symbol(backend, "vertical_vessel", 300, 120) + lt = await place_symbol(backend, "instrument", 300, 200, tag="LT-201", location="field") + signal = await draw_line( + backend, + {"handle": lt["handle"]}, + {"handle": vessel["handle"], "port": "N1"}, + line_class="electric", + ) + graph = await build_graph(backend) + row = next(r for r in instrument_index(graph) if r["tag"] == "LT-201") + assert row["connected_to"] == vessel["handle"] + assert row["connected_to"] != signal["line_number"] + assert row["signal_lines"] == 1