fix(task-16): P&ID focuses score evidence, never a 0.3 touch guess or an inferred port kind
Two ways the six focuses turned a reader guess into a gate-failing error: - The "no P&ID content" guard was `not graph["nodes"]`, but build_graph (include_foreign=True) files ANY INSERT as an `unknown_block` at 0.3 the moment a plain LINE's end lands on its box. A bolt with a note leader on a purely mechanical sheet became a P&ID node, the leader a foreign edge, and its free end a `pid_dangling_line` error — so drawing_finalize (focus=None) failed a sheet the seven mechanical focuses pass. Now `has_pid_content` needs a recognised node (catalog / xdata / heuristic); an `unknown_block` is never tagged or scored, and `_dangling` only reports an edge that is a P&ID line by evidence (a class from XDATA or a P&ID layer, or an end on a recognised node) — on a real P&ID the same leader is not dangling either. - `_incompatible` read `port["kind"]` off inferred ports, which the graph hard-codes to "process" for every foreign block. A signal line correctly drawn to a CTO control valve's actuator was an error whose hint said to redraw the control signal as a process line. Inferred ports are skipped; only a declared (catalogue / XDATA) port kind is judged. Tests: both repros as regressions (unit + the server drawing_finalize gate), a foreign GATE_VALVE whose dangling line is still reported (the heuristic path spec §9.3 pins is not silenced), a note leader on a real P&ID, and the spec's own negative — all six return [] on the README gear sheet. 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
co-authored by
Claude Opus 5
Claude Opus 5
parent
b4c1423c8e
commit
715c8efca4
@@ -2,7 +2,18 @@
|
||||
|
||||
Every check is pure over the graph dict; ``run_critique`` builds the graph once
|
||||
into the per-run ``shared`` context so six focuses cost one read. A drawing
|
||||
with no P&ID nodes yields no issues — mechanical sheets are never scored here.
|
||||
with no recognised P&ID symbol yields no issues — mechanical sheets are never
|
||||
scored here.
|
||||
|
||||
The checks score evidence, never a guess. ``build_graph(include_foreign=True)``
|
||||
files *any* INSERT as an ``unknown_block`` at confidence 0.3 the moment a plain
|
||||
LINE's end lands on its bounding box (spec §9.2 step 2) — a bolt with a note
|
||||
leader qualifies — and gives every foreign block's inferred port a placeholder
|
||||
``kind: "process"``. Neither is a measurement (spec §9.4: read the confidence
|
||||
before trusting a foreign graph), so no focus rests on them: an
|
||||
``unknown_block`` is not P&ID content, is never tagged or scored, and an edge
|
||||
anchored on nothing better is never called dangling; an inferred port is never
|
||||
called incompatible.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -29,9 +40,44 @@ PID_FOCUSES = (
|
||||
_KEY = "pid_graph"
|
||||
|
||||
|
||||
def _recognised(node: dict) -> bool:
|
||||
"""A node the drawing vouches for: catalogue, XDATA or heuristic (a keyword
|
||||
block name / tag attribute). ``unknown_block`` is a 0.3 touch guess."""
|
||||
return node["kind"] != "unknown_block"
|
||||
|
||||
|
||||
def has_pid_content(graph: dict) -> bool:
|
||||
"""True when at least one recognised P&ID symbol is on the sheet."""
|
||||
return any(_recognised(n) for n in graph["nodes"])
|
||||
|
||||
|
||||
def _scored_nodes(graph: dict) -> list[dict]:
|
||||
return [n for n in graph["nodes"] if _recognised(n)]
|
||||
|
||||
|
||||
def _evidenced_edges(graph: dict) -> set[str]:
|
||||
"""Edge ids that are P&ID lines by evidence: a line class from XDATA or a
|
||||
P&ID line layer, or an end on a recognised node. A foreign line whose only
|
||||
anchor is an ``unknown_block`` box is in the graph on a guess."""
|
||||
nodes = {n["id"]: n for n in graph["nodes"]}
|
||||
out: set[str] = set()
|
||||
for edge in graph["edges"]:
|
||||
if edge.get("class_source"):
|
||||
out.add(edge["id"])
|
||||
continue
|
||||
for ref in (edge["from"], edge["to"]):
|
||||
if ref and "node" in ref and _recognised(nodes[ref["node"]]):
|
||||
out.add(edge["id"])
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _dangling(graph: dict) -> list[Issue]:
|
||||
evidenced = _evidenced_edges(graph)
|
||||
out = []
|
||||
for d in graph["dangling"]:
|
||||
if d["edge"] not in evidenced:
|
||||
continue
|
||||
nearest = d.get("nearest")
|
||||
if nearest:
|
||||
hint = (
|
||||
@@ -56,7 +102,7 @@ def _dangling(graph: dict) -> list[Issue]:
|
||||
|
||||
def _duplicate_tags(graph: dict) -> list[Issue]:
|
||||
seen: dict[str, list[str]] = {}
|
||||
for node in graph["nodes"]:
|
||||
for node in _scored_nodes(graph):
|
||||
# connectors carry free text ("TO P&ID-002"), never a unique tag
|
||||
if node.get("tag") and node["kind"] != "connector":
|
||||
seen.setdefault(node["tag"].upper(), []).append(node["id"])
|
||||
@@ -87,6 +133,12 @@ def _incompatible(graph: dict) -> list[Issue]:
|
||||
if not ref or "node" not in ref:
|
||||
continue
|
||||
port = nodes[ref["node"]]["ports"].get(ref["port"], {})
|
||||
if port.get("inferred"):
|
||||
# A foreign block's port kind is a placeholder, not a
|
||||
# measurement: a signal line to a CTO control valve's actuator
|
||||
# is right, and "redraw as a process line" would be the wrong
|
||||
# fix. Only a declared (catalogue / XDATA) port kind is judged.
|
||||
continue
|
||||
actual = port.get("kind")
|
||||
if actual in ("process", "signal") and actual != cls.kind:
|
||||
out.append(
|
||||
@@ -126,7 +178,7 @@ def _untagged(graph: dict) -> list[Issue]:
|
||||
|
||||
def _illegal(graph: dict) -> list[Issue]:
|
||||
out = []
|
||||
for n in graph["nodes"]:
|
||||
for n in _scored_nodes(graph):
|
||||
if not n.get("tag") or n["kind"] == "connector":
|
||||
continue
|
||||
parsed = parse_tag(n["tag"], kind="instrument" if n["kind"] == "instrument" else "auto")
|
||||
@@ -170,7 +222,7 @@ _CHECKS = {
|
||||
|
||||
|
||||
def issues_for(focus: str, graph: dict) -> list[Issue]:
|
||||
if not graph["nodes"]:
|
||||
if not has_pid_content(graph):
|
||||
return []
|
||||
return _CHECKS[focus](graph)
|
||||
|
||||
|
||||
@@ -4,7 +4,9 @@ from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from engineering.pid.critique import has_pid_content, issues_for
|
||||
from engineering.pid.drawlines import draw_line
|
||||
from engineering.pid.graph import build_graph
|
||||
from engineering.pid.insert import place_symbol
|
||||
from engineering.plan_spec import ALL_CRITIQUE_FOCUSES
|
||||
from engineering.preflight import pid_plan_warnings, preflight_drawing
|
||||
@@ -110,6 +112,53 @@ async def test_unconnected_equipment_is_informational(backend):
|
||||
assert len(issues) == 1 and issues[0].severity == "info"
|
||||
|
||||
|
||||
async def test_incompatible_is_silent_on_a_foreign_blocks_inferred_port(backend):
|
||||
"""A CTO-library control valve has no declared ports: the graph infers one
|
||||
where the line lands and gives it a placeholder ``kind: "process"``. A signal
|
||||
line to its actuator is correct, and the old "redraw as a process line"
|
||||
error was a guess presented as a measurement."""
|
||||
await backend.drawing_apply_iso_layers("pid")
|
||||
await backend.block_define(
|
||||
"CONTROL_VALVE",
|
||||
[
|
||||
{"type": "polyline", "points": [[-6, -3], [6, 3], [6, -3], [-6, 3]], "closed": True},
|
||||
{"type": "line", "x1": 0, "y1": 0, "x2": 0, "y2": 6},
|
||||
{"type": "arc", "cx": 0, "cy": 6, "r": 3, "start_deg": 0, "end_deg": 180},
|
||||
],
|
||||
attdefs=[{"tag": "TAG", "x": 0, "y": 12, "height": 2.5}],
|
||||
)
|
||||
cv = await backend.block_insert(
|
||||
"CONTROL_VALVE", 160, 100, attributes={"TAG": "FCV-101"}, layer="PROCESS-EQUIPMENT"
|
||||
)
|
||||
fic = await place_symbol(
|
||||
backend, "instrument", 160, 140, tag="FIC-101", type="dcs", location="primary"
|
||||
)
|
||||
await draw_line(backend, {"handle": fic["handle"]}, {"x": 160, "y": 109}, line_class="electric")
|
||||
graph = await build_graph(backend, include_foreign=True)
|
||||
valve = next(n for n in graph["nodes"] if n["id"] == cv.handle)
|
||||
assert valve["source"] == "heuristic" and valve["ports"]["p1"]["inferred"] is True
|
||||
issues = await backend.drawing_critique(None)
|
||||
assert [i.focus for i in issues if i.focus.startswith("pid_")] == []
|
||||
|
||||
|
||||
async def test_a_dangling_line_off_a_foreign_valve_is_still_reported(backend):
|
||||
"""The evidence rule must not silence the foreign path spec §9.3 pins: a
|
||||
heuristic node (keyword name + TAG attribute) is recognised, so a plain
|
||||
line leaving its box for nowhere is a real dangling end."""
|
||||
await backend.block_define(
|
||||
"GATE_VALVE",
|
||||
[
|
||||
{"type": "polyline", "points": [[-4, -2], [-4, 2], [0, 0]], "closed": True},
|
||||
{"type": "polyline", "points": [[4, -2], [4, 2], [0, 0]], "closed": True},
|
||||
],
|
||||
attdefs=[{"tag": "TAG", "x": 0, "y": 4, "height": 2.5}],
|
||||
)
|
||||
await backend.block_insert("GATE_VALVE", 50, 50, attributes={"TAG": "HV-9"})
|
||||
line = await backend.entity_create_line(54, 50, 90, 50, layer="0")
|
||||
issues = await backend.drawing_critique(["pid_dangling_line"])
|
||||
assert [(i.severity, i.handles) for i in issues] == [("error", [line.handle])]
|
||||
|
||||
|
||||
async def test_mechanical_drawing_is_untouched_by_pid_focuses(backend):
|
||||
from engineering.layers import ensure_engineering_layers
|
||||
|
||||
@@ -119,6 +168,83 @@ async def test_mechanical_drawing_is_untouched_by_pid_focuses(backend):
|
||||
assert await backend.drawing_critique(list(PID_FOCUSES)) == []
|
||||
|
||||
|
||||
async def _bolt_with_note_leader(backend):
|
||||
"""A mechanical INSERT with a note leader from its edge — the shape that
|
||||
files the bolt as an ``unknown_block`` at 0.3 and the leader as a foreign
|
||||
edge with a free far end."""
|
||||
await backend.block_define(
|
||||
"BOLT_M10",
|
||||
[
|
||||
{"type": "circle", "cx": 0, "cy": 0, "r": 5},
|
||||
{"type": "polyline", "points": [[-5, -5], [5, -5], [5, 5], [-5, 5]], "closed": True},
|
||||
],
|
||||
)
|
||||
bolt = await backend.block_insert("BOLT_M10", 100, 100, layer="GEOMETRY")
|
||||
leader = await backend.entity_create_line(105, 100, 140, 120, layer="GEOMETRY")
|
||||
await backend.entity_create_text("SEE NOTE 3", 141, 120, 3.5, layer="TEXT")
|
||||
return bolt.handle, leader.handle
|
||||
|
||||
|
||||
async def test_a_mechanical_insert_with_a_note_leader_is_not_pid_content(backend):
|
||||
"""Spec §11.1: a drawing without P&ID content is never scored. The graph
|
||||
still files the bolt as an ``unknown_block`` (that is the reader's job);
|
||||
the critique must not mistake that 0.3 guess for a P&ID symbol and fail
|
||||
the finalize gate on the leader's free end."""
|
||||
await backend.drawing_apply_iso_layers("mech")
|
||||
bolt, leader = await _bolt_with_note_leader(backend)
|
||||
graph = await build_graph(backend, include_foreign=True)
|
||||
assert [(n["id"], n["kind"], n["confidence"]) for n in graph["nodes"]] == [
|
||||
(bolt, "unknown_block", 0.3)
|
||||
]
|
||||
assert [d["edge"] for d in graph["dangling"]] == [leader]
|
||||
assert has_pid_content(graph) is False
|
||||
for focus in PID_FOCUSES:
|
||||
assert issues_for(focus, graph) == [], focus
|
||||
issues = await backend.drawing_critique(None)
|
||||
assert [i.focus for i in issues if i.focus.startswith("pid_")] == []
|
||||
|
||||
|
||||
async def test_finalize_gate_passes_a_mechanical_sheet_with_a_note_leader(backend, tmp_path):
|
||||
"""The gate runs ``focus=None``, so the six focuses sit inside every
|
||||
``drawing_finalize`` — the mechanical sheet must pass it as it did before
|
||||
they joined."""
|
||||
import server
|
||||
|
||||
class _Ctx:
|
||||
def __init__(self, backend):
|
||||
self.lifespan_context = {"backend": backend}
|
||||
|
||||
async def warning(self, message):
|
||||
pass
|
||||
|
||||
await backend.drawing_apply_iso_layers("mech")
|
||||
await _bolt_with_note_leader(backend)
|
||||
payload = await server.drawing_finalize(save_path=str(tmp_path / "bolt.dxf"), ctx=_Ctx(backend))
|
||||
assert payload["critique"] == []
|
||||
|
||||
|
||||
async def test_a_note_leader_off_an_unknown_block_on_a_pid_is_not_dangling(backend):
|
||||
"""On a real P&ID the same leader is in the graph only because it touched
|
||||
a 0.3 box; the one genuine finding (the free process line) stays."""
|
||||
await _clean(backend)
|
||||
await _bolt_with_note_leader(backend)
|
||||
issues = await backend.drawing_critique(["pid_dangling_line"])
|
||||
assert len(issues) == 1 and issues[0].detail["x"] == pytest.approx(220.0)
|
||||
|
||||
|
||||
async def test_readme_gear_sheet_is_untouched_by_pid_focuses(backend):
|
||||
"""The spec's own negative: all six return [] on the README gear sheet."""
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
path = Path(__file__).resolve().parents[1] / "scripts" / "render_readme_showcase.py"
|
||||
spec = importlib.util.spec_from_file_location("render_readme_showcase", path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
await module.build(backend)
|
||||
assert await backend.drawing_critique(list(PID_FOCUSES)) == []
|
||||
|
||||
|
||||
async def test_refine_reports_pid_issues_as_manual_and_moves_nothing(backend):
|
||||
# The refine loop is `engineering.refiner.refine_drawing` (what the
|
||||
# `drawing_refine` tool calls) — the backend has no `drawing_refine` method.
|
||||
|
||||
Reference in New Issue
Block a user