fix(task-16): hints only name ports pid_line_draw accepts; signal_unknown is judged as a signal line

Two review findings on the P&ID critique focuses.

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, so a free end near
a sheet frame was hinted to connect to the frame — a call `pid_line_draw`
refuses, since the frame carries no ACADMCP_PID payload. Every dangling row
is now re-hinted over connectable ports only (catalogue / XDATA nodes,
declared ports), so every hint is executable.

`signal_unknown` — the class the reader assigns to a signal-layer line whose
exact class it cannot tell — was skipped by the incompatibility check
because it is not in the drawable class table. Its kind is still signal, so
a signal-layer line ending on a process port is now an error as spec §11.1
requires; only an unclassified line is skipped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd
This commit is contained in:
Umut CAN
2026-09-16 12:01:37 +03:00
co-authored by Claude Opus 5
parent 3c4fbf6c42
commit 6aded6bbf0
2 changed files with 110 additions and 12 deletions
+56 -12
View File
@@ -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"
),
},
)
+54
View File
@@ -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"