diff --git a/engineering/pid/critique.py b/engineering/pid/critique.py index 1b2a0b7..6314568 100644 --- a/engineering/pid/critique.py +++ b/engineering/pid/critique.py @@ -75,18 +75,29 @@ def _evidenced_edges(graph: dict) -> set[str]: return out +def _connectable(node: dict) -> bool: + """A node ``pid_line_draw`` will accept as an endpoint: it resolves ports + from the ACADMCP_PID payload, so only catalogue / XDATA symbols qualify. A + heuristic node (keyword name, tag attribute) is recognised for scoring but + refused by the drawing tool, and a hint that names it is a hint that fails.""" + return node.get("source") in ("catalog", "xdata") + + def _nearest_recognised_port(graph: dict, nodes: dict, space, x: float, y: float) -> dict | None: """The same hint the graph gives a free end (spec §9.2 step 5: the nearest - port within ``10·tolerance``), over ports the drawing vouches for. A - bubble's radial port is matched on its circle, not its centre, so it is - left out exactly as the graph's grid leaves it out.""" + port within ``10·tolerance``), restricted to ports the hint's own + ``pid_line_draw`` call can use: declared ports of catalogue / XDATA + symbols in the same space. Inferred ``p1..pn`` ports are never suggested — + the drawing tool refuses their node. A bubble's radial port is matched on + its circle, not its centre, so it is left out exactly as the graph's grid + leaves it out.""" reach = 10.0 * graph["tolerance"] best = None for node in nodes.values(): - if not _recognised(node) or node["space"] != space: + if not _connectable(node) or node["space"] != space: continue for pname, port in node["ports"].items(): - if port.get("radius", 0.0) > 0: + if port.get("radius", 0.0) > 0 or port.get("inferred"): continue d = math.hypot(port["x"] - x, port["y"] - y) if d <= reach and (best is None or d < best["distance"]): @@ -94,6 +105,13 @@ def _nearest_recognised_port(graph: dict, nodes: dict, space, x: float, y: float return best +def _edge_space(graph: dict, edge_id: str) -> str | None: + for edge in graph["edges"]: + if edge["id"] == edge_id: + return edge.get("space") + return None + + def _dangling_ends(graph: dict) -> list[dict]: """Free ends of evidenced edges, in the graph's ``dangling`` row shape. @@ -106,7 +124,20 @@ def _dangling_ends(graph: dict) -> list[dict]: """ evidenced = _evidenced_edges(graph) nodes = {n["id"]: n for n in graph["nodes"]} - rows = [d for d in graph["dangling"] if d["edge"] in evidenced] + # The graph's own ``nearest`` is computed over every port in its grid, + # including the ``p1..pn`` ports it invents on a promoted unknown_block; a + # hint built from that names a node ``pid_line_draw`` refuses. Recompute + # it over connectable ports only, so every hint is executable. + rows = [ + { + **d, + "nearest": _nearest_recognised_port( + graph, nodes, _edge_space(graph, d["edge"]), d["x"], d["y"] + ), + } + for d in graph["dangling"] + if d["edge"] in evidenced + ] for edge in graph["edges"]: if edge["id"] not in evidenced: continue @@ -179,12 +210,25 @@ def _duplicate_tags(graph: dict) -> list[Issue]: ] +def _edge_kind(line_class: str | None) -> str | None: + """process | signal for a graph edge class. ``signal_unknown`` is what the + reader assigns to a line on the signal layer whose exact class it cannot + tell — its kind is still signal, and the spec's rule (a signal line ending + on a process port) applies; only an unclassified line (``None``) is skipped.""" + if not line_class: + return None + if line_class == "signal_unknown": + return "signal" + cls = LINE_CLASSES.get(line_class) + return cls.kind if cls else None + + def _incompatible(graph: dict) -> list[Issue]: nodes = {n["id"]: n for n in graph["nodes"]} out = [] for edge in graph["edges"]: - cls = LINE_CLASSES.get(edge.get("line_class") or "") - if cls is None: + kind = _edge_kind(edge.get("line_class")) + if kind is None: continue for ref in (edge["from"], edge["to"]): if not ref or "node" not in ref: @@ -197,21 +241,21 @@ def _incompatible(graph: dict) -> list[Issue]: # fix. Only a declared (catalogue / XDATA) port kind is judged. continue actual = port.get("kind") - if actual in ("process", "signal") and actual != cls.kind: + if actual in ("process", "signal") and actual != kind: out.append( Issue( "error", "pid_incompatible_connection", - f"{cls.kind} line {edge['id']} ends on {actual} port {ref['port']} " + f"{kind} line {edge['id']} ends on {actual} port {ref['port']} " f"of {ref['node']}.", [edge["id"], ref["node"]], { - "expected": cls.kind, + "expected": kind, "actual": actual, "port": ref["port"], "line_class": edge["line_class"], "hint": ( - f"redraw with a {actual} line class or connect to a {cls.kind} port" + f"redraw with a {actual} line class or connect to a {kind} port" ), }, ) diff --git a/tests/test_pid_critique.py b/tests/test_pid_critique.py index 94e4218..f12dfcb 100644 --- a/tests/test_pid_critique.py +++ b/tests/test_pid_critique.py @@ -375,3 +375,57 @@ def test_preflight_flags_a_pid_intent_on_the_wrong_layer_set(): assert "LAYER_SET_INTENT_MISMATCH" not in [c.code for c in ok.conflicts] assert pid_plan_warnings("Bearing housing", "mech") == [] assert pid_plan_warnings("piping and instrumentation diagram", "mech") + + +async def test_a_free_end_near_an_invented_frame_port_gets_no_executable_hint(backend): + """The graph's own ``nearest`` is computed over every port in its grid, + including the ``p1`` it invents on a promoted frame. A hint that names + that port is a hint ``pid_line_draw`` refuses (the frame carries no + ACADMCP_PID payload). Every dangling row — not only the ends that touch an + unknown block — must be re-hinted over connectable ports.""" + frame, pump = await _frame_and_pump(backend) + await draw_line(backend, {"handle": pump["handle"], "port": "discharge"}, {"x": 420, "y": 106}) + free = await backend.entity_create_line(300, 108, 417, 108, layer="PROCESS-PIPING-MAIN") + + graph = await build_graph(backend, include_foreign=True) + row = next(d for d in graph["dangling"] if d["edge"] == free.handle and d["end"] == "to") + assert row["nearest"] and row["nearest"]["node"] == frame, "the graph itself still hints at p1" + + issues = await backend.drawing_critique(["pid_dangling_line"]) + to_row = next(i for i in issues if i.handles == [free.handle] and i.detail["end"] == "to") + assert to_row.detail["nearest"] is None + assert "no port within reach" in to_row.detail["hint"] + for issue in issues: + nearest = issue.detail.get("nearest") + if nearest: + node = next(n for n in graph["nodes"] if n["id"] == nearest["node"]) + assert node["source"] in ("catalog", "xdata") + assert node["ports"][nearest["port"]]["inferred"] is False + + +async def test_a_signal_layer_line_of_unknown_class_into_a_process_port_is_incompatible(backend): + """A line on INSTRUMENT-LINE-SIGNAL with a continuous linetype reads as + ``signal_unknown``: the reader cannot tell pneumatic from data, but it is + a signal line, and spec §11.1's rule (a signal-class edge ending on a + process port) applies to it. Skipping it would let ``drawing_finalize`` + pass the connection.""" + from engineering.layers import apply_layer_set + + await apply_layer_set(backend, "pid") + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-101") + out = pump["ports"]["discharge"] + line = await backend.entity_create_line( + out["x"], out["y"], out["x"] + 40, out["y"] + 40, layer="INSTRUMENT-LINE-SIGNAL" + ) + await backend.entity_set_properties(line.handle, linetype="Continuous") + graph = await build_graph(backend) + edge = next(e for e in graph["edges"] if e["id"] == line.handle) + assert edge["line_class"] == "signal_unknown" and edge["from"] == { + "node": pump["handle"], + "port": "discharge", + } + + issues = await backend.drawing_critique(["pid_incompatible_connection"]) + assert len(issues) == 1 and issues[0].severity == "error" + assert issues[0].detail["expected"] == "signal" and issues[0].detail["actual"] == "process" + assert issues[0].detail["line_class"] == "signal_unknown"