diff --git a/engineering/critique.py b/engineering/critique.py index 8b5bad0..5705b99 100644 --- a/engineering/critique.py +++ b/engineering/critique.py @@ -408,6 +408,10 @@ _FOCUS_DISPATCH = { "gdt": _check_gdt, } +from .pid.critique import PID_DISPATCH # noqa: E402 (registered after the mechanical focuses) + +_FOCUS_DISPATCH.update(PID_DISPATCH) + async def run_critique( backend: AutoCADBackend, @@ -416,6 +420,9 @@ async def run_critique( """Dispatch each requested focus check and return aggregated issues.""" foci = list(focus) if focus is not None else list(ALL_CRITIQUE_FOCUSES) issues: list[Issue] = [] + # Per-run context: the P&ID focuses build the connectivity graph once + # here and share it, so six focuses cost one read of the drawing. + shared: dict = {} for f in foci: check = _FOCUS_DISPATCH.get(f) if check is None: @@ -427,5 +434,8 @@ async def run_critique( ) ) continue - issues.extend(await check(backend)) + if getattr(check, "needs_shared", False): + issues.extend(await check(backend, shared)) + else: + issues.extend(await check(backend)) return issues diff --git a/engineering/pid/critique.py b/engineering/pid/critique.py new file mode 100644 index 0000000..a928041 --- /dev/null +++ b/engineering/pid/critique.py @@ -0,0 +1,190 @@ +"""Six P&ID focuses over one graph per critique run. + +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. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from engineering.plan_spec import Issue + +from .graph import build_graph +from .lines import LINE_CLASSES +from .tags import parse_tag + +if TYPE_CHECKING: + from backends.base import AutoCADBackend + +PID_FOCUSES = ( + "pid_dangling_line", + "pid_duplicate_tag", + "pid_incompatible_connection", + "pid_untagged_instrument", + "pid_illegal_tag", + "pid_unconnected_equipment", +) +_KEY = "pid_graph" + + +def _dangling(graph: dict) -> list[Issue]: + out = [] + for d in graph["dangling"]: + nearest = d.get("nearest") + if nearest: + hint = ( + f"pid_line_draw(from_={{'handle': ..., 'port': ...}}, " + f"to={{'handle': '{nearest['node']}', 'port': '{nearest['port']}'}}) " + f"— nearest port {nearest['distance']} mm away" + ) + else: + hint = "no port within reach; connect this end with pid_line_draw or delete the line" + out.append( + Issue( + "error", + "pid_dangling_line", + f"Line {d['edge']} {d['end']} end at ({d['x']:.2f}, {d['y']:.2f}) " + "connects to nothing.", + [d["edge"]], + {"end": d["end"], "x": d["x"], "y": d["y"], "nearest": nearest, "hint": hint}, + ) + ) + return out + + +def _duplicate_tags(graph: dict) -> list[Issue]: + seen: dict[str, list[str]] = {} + for node in graph["nodes"]: + # 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"]) + return [ + Issue( + "error", + "pid_duplicate_tag", + f"Tag {tag} is carried by {len(handles)} symbols.", + handles, + { + "tag": tag, + "hint": "block_set_attributes(handle, {'TAG'|'FUNC'/'LOOP': ...}) on all but one", + }, + ) + for tag, handles in seen.items() + if len(handles) > 1 + ] + + +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: + continue + for ref in (edge["from"], edge["to"]): + if not ref or "node" not in ref: + continue + port = nodes[ref["node"]]["ports"].get(ref["port"], {}) + actual = port.get("kind") + if actual in ("process", "signal") and actual != cls.kind: + out.append( + Issue( + "error", + "pid_incompatible_connection", + f"{cls.kind} line {edge['id']} ends on {actual} port {ref['port']} " + f"of {ref['node']}.", + [edge["id"], ref["node"]], + { + "expected": cls.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" + ), + }, + ) + ) + return out + + +def _untagged(graph: dict) -> list[Issue]: + return [ + Issue( + "warning", + "pid_untagged_instrument", + f"Instrument {n['id']} has no tag.", + [n["id"]], + {"hint": "block_set_attributes(handle, {'FUNC': 'FIC', 'LOOP': '101'})"}, + ) + for n in graph["nodes"] + if n["kind"] == "instrument" and not n.get("tag") + ] + + +def _illegal(graph: dict) -> list[Issue]: + out = [] + for n in graph["nodes"]: + if not n.get("tag") or n["kind"] == "connector": + continue + parsed = parse_tag(n["tag"], kind="instrument" if n["kind"] == "instrument" else "auto") + if not parsed["valid"]: + out.append( + Issue( + "warning", + "pid_illegal_tag", + f"Tag {n['tag']} on {n['id']} is not ISA-5.1: {parsed['errors'][0]}", + [n["id"]], + {"errors": parsed["errors"], "hint": "block_set_attributes with a valid tag"}, + ) + ) + return out + + +def _unconnected(graph: dict) -> list[Issue]: + return [ + Issue( + "info", + "pid_unconnected_equipment", + f"{n['kind']} {n.get('tag') or n['id']} has no line on any port.", + [n["id"]], + {"hint": "informational — connect it with pid_line_draw or leave it if intended"}, + ) + for n in graph["nodes"] + if n["kind"] in ("equipment", "valve") + and n["ports"] + and not any(p["edges"] for p in n["ports"].values()) + ] + + +_CHECKS = { + "pid_dangling_line": _dangling, + "pid_duplicate_tag": _duplicate_tags, + "pid_incompatible_connection": _incompatible, + "pid_untagged_instrument": _untagged, + "pid_illegal_tag": _illegal, + "pid_unconnected_equipment": _unconnected, +} + + +def issues_for(focus: str, graph: dict) -> list[Issue]: + if not graph["nodes"]: + return [] + return _CHECKS[focus](graph) + + +def _make(focus: str): + async def check(backend: AutoCADBackend, shared: dict) -> list[Issue]: + graph = shared.get(_KEY) + if graph is None: + graph = shared[_KEY] = await build_graph(backend, include_foreign=True) + return issues_for(focus, graph) + + check.needs_shared = True # type: ignore[attr-defined] + check.__name__ = f"check_{focus}" + return check + + +PID_DISPATCH = {focus: _make(focus) for focus in PID_FOCUSES} diff --git a/engineering/plan_spec.py b/engineering/plan_spec.py index ac72e2a..1b1d7bb 100644 --- a/engineering/plan_spec.py +++ b/engineering/plan_spec.py @@ -21,6 +21,12 @@ CritiqueFocus = Literal[ "duplicate_entities", "construction_left", "gdt", + "pid_dangling_line", + "pid_duplicate_tag", + "pid_incompatible_connection", + "pid_untagged_instrument", + "pid_illegal_tag", + "pid_unconnected_equipment", ] SnapType = Literal["end", "mid", "center", "quad", "int", "perp", "near"] @@ -32,6 +38,12 @@ ALL_CRITIQUE_FOCUSES: tuple[CritiqueFocus, ...] = ( "duplicate_entities", "construction_left", "gdt", + "pid_dangling_line", + "pid_duplicate_tag", + "pid_incompatible_connection", + "pid_untagged_instrument", + "pid_illegal_tag", + "pid_unconnected_equipment", ) diff --git a/engineering/preflight.py b/engineering/preflight.py index 661233b..12850ad 100644 --- a/engineering/preflight.py +++ b/engineering/preflight.py @@ -5,10 +5,27 @@ from __future__ import annotations import hashlib import json import math +import re from dataclasses import dataclass, field from numbers import Real from typing import Any +PID_INTENT_RE = re.compile( + r"\b(p&id|pid|piping and instrumentation|process flow|instrument)\b", re.IGNORECASE +) + + +def intent_reads_as_pid(intent: str) -> bool: + return bool(PID_INTENT_RE.search(str(intent or ""))) + + +def pid_plan_warnings(intent: str, layer_set_id: str) -> list[str]: + if intent_reads_as_pid(intent) and str(layer_set_id).strip().lower() != "pid": + return [ + f"intent reads as a P&ID but the layer set is '{layer_set_id}'; pass layer_set_id='pid'" + ] + return [] + @dataclass(frozen=True) class PreflightQuestion: @@ -201,6 +218,12 @@ def preflight_drawing( for valid, code, field_name, value, message in parameter_rules: if not valid: conflicts.append(PreflightConflict(code, field_name, [value], message)) + for message in pid_plan_warnings(intent, normalized_layer_set): + conflicts.append( + PreflightConflict( + "LAYER_SET_INTENT_MISMATCH", "layer_set_id", [layer_set_id, "pid"], message + ) + ) normalized_spec = { "intent": str(intent).strip(), diff --git a/server.py b/server.py index f944226..5d2bcf8 100644 --- a/server.py +++ b/server.py @@ -5401,7 +5401,10 @@ async def drawing_finalize( ctx: Context = None, ) -> dict: """Premium completion gate: runs BOTH the 8-step validator AND the premium critique focuses - (iso128, layer_color, dim_overlap, untrimmed_corner, duplicate_entities, construction_left), + (iso128, layer_color, dim_overlap, untrimmed_corner, duplicate_entities, construction_left, + gdt, plus the P&ID focuses pid_dangling_line, pid_duplicate_tag, + pid_incompatible_connection, pid_untagged_instrument, pid_illegal_tag, + pid_unconnected_equipment — silent on a sheet with no P&ID symbols), then saves to disk, exports a screenshot, and returns the DWG path. Raises ToolError if any validator 'error' finding is present, or if critique reports an @@ -5629,7 +5632,13 @@ async def drawing_plan( requirements, spec_hash, ) - return _dc(plan) + from engineering.preflight import pid_plan_warnings + + out = _dc(plan) + warnings = pid_plan_warnings(intent, layer_set_id) + if warnings: + out["warnings"] = warnings + return out @cad_tool( @@ -5650,7 +5659,10 @@ async def drawing_critique( Field( default=None, description="Subset of: iso128, layer_color, dim_overlap, untrimmed_corner, " - "duplicate_entities, construction_left. None = run all.", + "duplicate_entities, construction_left, gdt, and the P&ID focuses " + "pid_dangling_line, pid_duplicate_tag, pid_incompatible_connection, " + "pid_untagged_instrument, pid_illegal_tag, pid_unconnected_equipment " + "(silent on a drawing with no P&ID symbols). None = run all.", ), ] = None, ctx: Context = None, diff --git a/tests/test_pid_critique.py b/tests/test_pid_critique.py new file mode 100644 index 0000000..da6c951 --- /dev/null +++ b/tests/test_pid_critique.py @@ -0,0 +1,155 @@ +"""Six P&ID focuses; all silent on a drawing with no P&ID content.""" + +from __future__ import annotations + +import pytest + +from engineering.pid.drawlines import draw_line +from engineering.pid.insert import place_symbol +from engineering.plan_spec import ALL_CRITIQUE_FOCUSES +from engineering.preflight import pid_plan_warnings, preflight_drawing + +pytestmark = pytest.mark.asyncio + +PID_FOCUSES = ( + "pid_dangling_line", + "pid_duplicate_tag", + "pid_incompatible_connection", + "pid_untagged_instrument", + "pid_illegal_tag", + "pid_unconnected_equipment", +) + + +def test_focuses_are_in_the_closed_enum(): + for focus in PID_FOCUSES: + assert focus in ALL_CRITIQUE_FOCUSES + + +async def _clean(backend): + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-101") + cv = await place_symbol(backend, "globe", 160, 106, tag="FCV-101", actuator="diaphragm") + fic = await place_symbol( + backend, "instrument", 160, 140, tag="FIC-101", type="dcs", location="primary" + ) + await draw_line( + backend, + {"handle": pump["handle"], "port": "discharge"}, + {"handle": cv["handle"], "port": "in"}, + ) + await draw_line(backend, {"handle": cv["handle"], "port": "out"}, {"x": 220, "y": 106}) + await draw_line( + backend, + {"handle": fic["handle"]}, + {"handle": cv["handle"], "port": "signal"}, + line_class="electric", + ) + return pump, cv, fic + + +async def test_clean_pid_has_no_pid_issues_except_the_free_end(backend): + await _clean(backend) + issues = await backend.drawing_critique(list(PID_FOCUSES)) + focuses = [i.focus for i in issues] + assert focuses == ["pid_dangling_line"], "the line to a free point is the only finding" + assert issues[0].severity == "error" and "hint" in issues[0].detail + + +async def test_dangling_carries_the_nearest_port_hint(backend): + pump, cv, _ = await _clean(backend) + await backend.entity_move(cv["handle"], 0.0, 1.0) + issues = await backend.drawing_critique(["pid_dangling_line"]) + hinted = [i for i in issues if i.detail.get("nearest")] + assert hinted and hinted[0].detail["nearest"]["node"] == cv["handle"] + assert "pid_line_draw" in hinted[0].detail["hint"] + + +async def test_duplicate_and_illegal_and_untagged(backend): + await place_symbol(backend, "instrument", 0, 0, tag="FIC-101") + await place_symbol(backend, "instrument", 30, 0, tag="FIC-101") + await place_symbol(backend, "instrument", 60, 0, tag="FCI-102") + await place_symbol(backend, "instrument", 90, 0) + await place_symbol(backend, "gate", 120, 0, tag="FIC-101") + issues = await backend.drawing_critique( + ["pid_duplicate_tag", "pid_illegal_tag", "pid_untagged_instrument"] + ) + by_focus = {} + for issue in issues: + by_focus.setdefault(issue.focus, []).append(issue) + assert len(by_focus["pid_duplicate_tag"]) == 1 + assert len(by_focus["pid_duplicate_tag"][0].handles) == 3 + assert by_focus["pid_duplicate_tag"][0].severity == "error" + assert len(by_focus["pid_illegal_tag"]) == 1 + assert by_focus["pid_illegal_tag"][0].severity == "warning" + assert len(by_focus["pid_untagged_instrument"]) == 1 + + +async def test_incompatible_connection(backend): + pump = await place_symbol(backend, "centrifugal_pump", 100, 100, tag="P-101") + fic = await place_symbol(backend, "instrument", 100, 160, tag="FIC-1") + await draw_line( + backend, + {"handle": fic["handle"]}, + {"handle": pump["handle"], "port": "discharge"}, + line_class="electric", + ) + issues = await backend.drawing_critique(["pid_incompatible_connection"]) + assert len(issues) == 1 and issues[0].severity == "error" + assert issues[0].detail == { + "expected": "signal", + "actual": "process", + "port": "discharge", + "line_class": "electric", + "hint": issues[0].detail["hint"], + } + + +async def test_unconnected_equipment_is_informational(backend): + await place_symbol(backend, "vertical_vessel", 0, 0, tag="V-1") + issues = await backend.drawing_critique(["pid_unconnected_equipment"]) + assert len(issues) == 1 and issues[0].severity == "info" + + +async def test_mechanical_drawing_is_untouched_by_pid_focuses(backend): + from engineering.layers import ensure_engineering_layers + + await ensure_engineering_layers(backend) + await backend.entity_create_circle(0, 0, 20, layer="GEOMETRY") + await backend.entity_create_line(-30, 0, 30, 0, layer="CENTER") + 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. + from engineering.refiner import refine_drawing + + await _clean(backend) + before = await backend.entity_count() + result = ( + await refine_drawing( + backend, max_rounds=1, min_score=100.0, focus=list(PID_FOCUSES), dry_run=True + ) + ).to_dict() + actions = [a for r in result["rounds"] for a in r["actions"]] + assert actions, "the free-end dangling line must reach the refiner as an action" + assert all(a["status"] == "manual_required" for a in actions) + assert await backend.entity_count() == before + + +def test_preflight_flags_a_pid_intent_on_the_wrong_layer_set(): + result = preflight_drawing( + "P&ID for the feed section", + {"units": "mm", "part_type": "pid", "dimensions": {}, "tolerance_policy": "none"}, + layer_set_id="mech", + ) + codes = [c.code for c in result.conflicts] + assert "LAYER_SET_INTENT_MISMATCH" in codes + ok = preflight_drawing( + "P&ID for the feed section", + {"units": "mm", "part_type": "pid", "dimensions": {}, "tolerance_policy": "none"}, + layer_set_id="pid", + ) + 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")