feat(pid): pid_from_spec — a whole P&ID in one transaction, graph and critique returned
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
6aded6bbf0
commit
d80ae17411
@@ -19,7 +19,7 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
|
||||
|
||||
</div>
|
||||
|
||||
> **v1.5 release snapshot:** 164 tools · 6 resources · 5 prompt templates · 1369 collected tests.
|
||||
> **v1.5 release snapshot:** 165 tools · 6 resources · 5 prompt templates · 1369 collected tests.
|
||||
> 154 is the **registered** count; a default install advertises 149 over `tools/list`,
|
||||
> because `ENABLE_3D` is unset. `system_about` is the runtime authority.
|
||||
|
||||
|
||||
@@ -1408,6 +1408,15 @@ TOOL_ALIASES: dict[str, ToolAliases] = {
|
||||
acad=(),
|
||||
synonyms=("equipment list", "equipment schedule", "valve list"),
|
||||
),
|
||||
"pid_from_spec": ToolAliases(
|
||||
acad=(),
|
||||
synonyms=(
|
||||
"whole p&id in one call",
|
||||
"p&id from json",
|
||||
"draw the process from a spec",
|
||||
"batch p&id",
|
||||
),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 26 KiB |
@@ -3,7 +3,7 @@
|
||||
"schema_version": 1,
|
||||
"package_version": "1.5.1",
|
||||
"totals": {
|
||||
"tools": 164,
|
||||
"tools": 165,
|
||||
"resources": 6,
|
||||
"prompts": 5,
|
||||
"groups": 20
|
||||
@@ -21,7 +21,7 @@
|
||||
"entity_query": 9,
|
||||
"layers": 14,
|
||||
"layouts": 12,
|
||||
"pid": 7,
|
||||
"pid": 8,
|
||||
"premium": 12,
|
||||
"solids": 5,
|
||||
"system": 7,
|
||||
@@ -1731,6 +1731,21 @@
|
||||
"readOnlyHint": true
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "pid_from_spec",
|
||||
"group": "pid",
|
||||
"cost": "mutate",
|
||||
"summary": "Draw a whole P&ID from one declarative spec in one transaction; returns graph + critique.",
|
||||
"tags": [
|
||||
"batch",
|
||||
"create",
|
||||
"pid"
|
||||
],
|
||||
"annotations": {
|
||||
"title": "P&ID: From Spec",
|
||||
"readOnlyHint": false
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "pid_graph",
|
||||
"group": "pid",
|
||||
|
||||
@@ -0,0 +1,436 @@
|
||||
"""``pid_from_spec``: a whole P&ID from one declarative document, in one transaction."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .critique import PID_FOCUSES, issues_for
|
||||
from .drawlines import draw_line
|
||||
from .graph import build_graph
|
||||
from .insert import place_symbol
|
||||
from .lines import AXIS, LINE_CLASSES, count_crossings, route, snap_axis
|
||||
from .symbols import resolve, transform_port
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from backends.base import AutoCADBackend
|
||||
|
||||
EXAMPLE_SPEC: dict = {
|
||||
"sheet": {
|
||||
"layer_set": "pid",
|
||||
"sheet_size": "A3",
|
||||
"scale": 1.0,
|
||||
"intent": "Feed section P&ID",
|
||||
},
|
||||
"equipment": [
|
||||
{
|
||||
"id": "P-101",
|
||||
"symbol": "centrifugal_pump",
|
||||
"x": 100,
|
||||
"y": 100,
|
||||
"rotation": 0,
|
||||
"tag": "P-101",
|
||||
"desc": "Feed pump",
|
||||
},
|
||||
{
|
||||
"id": "V-201",
|
||||
"symbol": "vertical_vessel",
|
||||
"x": 220,
|
||||
"y": 120,
|
||||
"tag": "V-201",
|
||||
"params": {
|
||||
"width": 20,
|
||||
"height": 40,
|
||||
"nozzles": [
|
||||
{"name": "N1", "side": "top", "fraction": 0.5},
|
||||
{"name": "N2", "side": "left", "fraction": 0.3},
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
"valves": [
|
||||
{
|
||||
"id": "FCV-101",
|
||||
"symbol": "globe",
|
||||
"actuator": "diaphragm",
|
||||
"x": 160,
|
||||
"y": 104,
|
||||
"tag": "FCV-101",
|
||||
"fail": "FC",
|
||||
},
|
||||
],
|
||||
"instruments": [
|
||||
{
|
||||
"id": "FIC-101",
|
||||
"type": "dcs",
|
||||
"location": "primary",
|
||||
"x": 160,
|
||||
"y": 140,
|
||||
"tag": "FIC-101",
|
||||
},
|
||||
],
|
||||
"lines": [
|
||||
{
|
||||
"from": "P-101.discharge",
|
||||
"to": "FCV-101.in",
|
||||
"class": "process_major",
|
||||
"number": "100-P-1001-CS1",
|
||||
},
|
||||
{
|
||||
"from": "FCV-101.out",
|
||||
"to": "V-201.N2",
|
||||
"class": "process_major",
|
||||
"number": "100-P-1001-CS1",
|
||||
},
|
||||
{"from": "FIC-101", "to": "FCV-101.signal", "class": "electric"},
|
||||
],
|
||||
"connectors": [
|
||||
{
|
||||
"id": "OP-1",
|
||||
"direction": "out",
|
||||
"x": 300,
|
||||
"y": 120,
|
||||
"tag": "TO P&ID-002",
|
||||
"link": "L-17",
|
||||
"from": "V-201.N1",
|
||||
"class": "process_major",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
_SECTIONS = ("equipment", "valves", "instruments", "connectors")
|
||||
_TOP_KEYS = {"sheet", "lines", *_SECTIONS}
|
||||
_ITEM_OPTIONS = {
|
||||
"equipment": {"symbol", "x", "y", "rotation", "scale", "tag", "desc", "params", "shape"},
|
||||
"valves": {"symbol", "x", "y", "rotation", "scale", "tag", "desc", "actuator", "fail"},
|
||||
"instruments": {"x", "y", "rotation", "scale", "tag", "type", "location"},
|
||||
"connectors": {
|
||||
"x",
|
||||
"y",
|
||||
"rotation",
|
||||
"scale",
|
||||
"tag",
|
||||
"link",
|
||||
"direction",
|
||||
"from",
|
||||
"to",
|
||||
"class",
|
||||
"number",
|
||||
},
|
||||
}
|
||||
_LINE_KEYS = {
|
||||
"from",
|
||||
"to",
|
||||
"class",
|
||||
"number",
|
||||
"size",
|
||||
"service",
|
||||
"spec",
|
||||
"insulation",
|
||||
"route",
|
||||
"stub",
|
||||
"label",
|
||||
"arrow",
|
||||
}
|
||||
|
||||
|
||||
def _num(value, where: str) -> float:
|
||||
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
||||
raise ValueError(f"{where} must be a number")
|
||||
return float(value)
|
||||
|
||||
|
||||
def _ref(value, where: str, ids: set[str]) -> tuple[str, str | None]:
|
||||
if not isinstance(value, str) or not value:
|
||||
raise ValueError(f"{where} must be 'ID' or 'ID.port'")
|
||||
item_id, _, port = value.partition(".")
|
||||
if item_id not in ids:
|
||||
raise ValueError(f"{where}: unknown id {item_id!r}")
|
||||
return item_id, port or None
|
||||
|
||||
|
||||
def _line_class(line: dict, where: str) -> str:
|
||||
cls = line.get("class", "process_major")
|
||||
if cls not in LINE_CLASSES:
|
||||
raise ValueError(f"{where}.class must be one of {', '.join(LINE_CLASSES)}")
|
||||
return cls
|
||||
|
||||
|
||||
def validate_spec(spec: dict) -> dict:
|
||||
"""Normalise a spec document; every error names the offending path.
|
||||
|
||||
Nothing is created here, so a malformed spec never reaches the drawing.
|
||||
"""
|
||||
if not isinstance(spec, dict):
|
||||
raise ValueError("spec must be an object")
|
||||
unknown = set(spec) - _TOP_KEYS
|
||||
if unknown:
|
||||
raise ValueError(f"unknown top-level key(s): {', '.join(sorted(unknown))}")
|
||||
out = {key: copy.deepcopy(spec.get(key) or ([] if key != "sheet" else {})) for key in _TOP_KEYS}
|
||||
if not isinstance(out["sheet"], dict):
|
||||
raise ValueError("sheet must be an object")
|
||||
ids: set[str] = set()
|
||||
for section in _SECTIONS:
|
||||
items = out[section]
|
||||
if not isinstance(items, list):
|
||||
raise ValueError(f"{section} must be a list")
|
||||
for i, item in enumerate(items):
|
||||
where = f"{section}[{i}]"
|
||||
if not isinstance(item, dict):
|
||||
raise ValueError(f"{where} must be an object")
|
||||
item_id = item.get("id")
|
||||
if not isinstance(item_id, str) or not item_id:
|
||||
raise ValueError(f"{where}.id must be a non-empty string")
|
||||
if item_id in ids:
|
||||
raise ValueError(f"{where}.id {item_id!r} is repeated")
|
||||
ids.add(item_id)
|
||||
bad = set(item) - _ITEM_OPTIONS[section] - {"id"}
|
||||
if bad:
|
||||
raise ValueError(f"{where}: unknown key(s) {', '.join(sorted(bad))}")
|
||||
for key in ("x", "y"):
|
||||
if key not in item:
|
||||
raise ValueError(f"{where}.{key} is required")
|
||||
item[key] = _num(item[key], f"{where}.{key}")
|
||||
for key in ("rotation", "scale"):
|
||||
if key in item:
|
||||
item[key] = _num(item[key], f"{where}.{key}")
|
||||
if section in ("equipment", "valves") and not isinstance(item.get("symbol"), str):
|
||||
raise ValueError(f"{where}.symbol is required")
|
||||
if section == "connectors":
|
||||
if item.get("direction") not in ("in", "out"):
|
||||
raise ValueError(f"{where}.direction must be 'in' or 'out'")
|
||||
has_from, has_to = "from" in item, "to" in item
|
||||
if item["direction"] == "out" and (not has_from or has_to):
|
||||
raise ValueError(f"{where}: an 'out' connector carries exactly 'from'")
|
||||
if item["direction"] == "in" and (not has_to or has_from):
|
||||
raise ValueError(f"{where}: an 'in' connector carries exactly 'to'")
|
||||
lines = out["lines"]
|
||||
if not isinstance(lines, list):
|
||||
raise ValueError("lines must be a list")
|
||||
for i, line in enumerate(lines):
|
||||
where = f"lines[{i}]"
|
||||
if not isinstance(line, dict):
|
||||
raise ValueError(f"{where} must be an object")
|
||||
bad = set(line) - _LINE_KEYS
|
||||
if bad:
|
||||
raise ValueError(f"{where}: unknown key(s) {', '.join(sorted(bad))}")
|
||||
for key in ("from", "to"):
|
||||
if key not in line:
|
||||
raise ValueError(f"{where}.{key} is required")
|
||||
_ref(line[key], f"{where}.{key}", ids)
|
||||
line["class"] = _line_class(line, where)
|
||||
if "stub" in line:
|
||||
line["stub"] = _num(line["stub"], f"{where}.stub")
|
||||
for i, conn in enumerate(out["connectors"]):
|
||||
key = "from" if conn["direction"] == "out" else "to"
|
||||
_ref(conn[key], f"connectors[{i}].{key}", ids)
|
||||
conn["class"] = _line_class(conn, f"connectors[{i}]")
|
||||
return out
|
||||
|
||||
|
||||
def _spec_for(section: str, item: dict):
|
||||
"""Resolve the SymbolSpec an item would place (no backend)."""
|
||||
if section == "instruments":
|
||||
return resolve("instrument", type=item.get("type"), location=item.get("location"))
|
||||
if section == "connectors":
|
||||
return resolve("offpage", direction=item["direction"])
|
||||
return resolve(
|
||||
item["symbol"],
|
||||
actuator=item.get("actuator"),
|
||||
params=item.get("params"),
|
||||
shape=item.get("shape"),
|
||||
)
|
||||
|
||||
|
||||
def _connector_lines(norm: dict) -> list[tuple[str, dict]]:
|
||||
"""The implicit line each connector carries, as ``(where, line)``."""
|
||||
out = []
|
||||
for i, conn in enumerate(norm["connectors"]):
|
||||
if conn["direction"] == "out":
|
||||
line = {"from": conn["from"], "to": conn["id"]}
|
||||
else:
|
||||
line = {"from": conn["id"], "to": conn["to"]}
|
||||
line.update({"class": conn["class"], "number": conn.get("number")})
|
||||
out.append((f"connectors[{i}]", line))
|
||||
return out
|
||||
|
||||
|
||||
def _all_lines(norm: dict) -> list[tuple[str, dict]]:
|
||||
return [(f"lines[{i}]", line) for i, line in enumerate(norm["lines"])] + _connector_lines(norm)
|
||||
|
||||
|
||||
def _plan_lines(norm: dict) -> list[dict]:
|
||||
"""Route every line in memory from catalogue ports — the dry run."""
|
||||
placed: dict[str, dict] = {}
|
||||
for section in _SECTIONS:
|
||||
for i, item in enumerate(norm[section]):
|
||||
try:
|
||||
spec = _spec_for(section, item)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"{section}[{i}]: {exc}") from exc
|
||||
ports = {
|
||||
p.name: transform_port(
|
||||
p, item["x"], item["y"], item.get("rotation", 0.0), item.get("scale", 1.0)
|
||||
)
|
||||
for p in spec.ports
|
||||
}
|
||||
placed[item["id"]] = {"family": spec.family, "ports": ports}
|
||||
|
||||
def endpoint(ref: str) -> dict:
|
||||
item_id, _, port = ref.partition(".")
|
||||
entry = placed[item_id]
|
||||
if not port:
|
||||
if len(entry["ports"]) == 1:
|
||||
port = next(iter(entry["ports"]))
|
||||
elif entry["family"] == "instrument" and "signal" in entry["ports"]:
|
||||
port = "signal"
|
||||
else:
|
||||
raise ValueError(f"{ref}: choose a port: {', '.join(entry['ports'])}")
|
||||
if port not in entry["ports"]:
|
||||
raise ValueError(f"{ref}: no port {port!r}; ports: {', '.join(entry['ports'])}")
|
||||
return entry["ports"][port]
|
||||
|
||||
def anchor(ep: dict, other: dict):
|
||||
if ep["radius"] > 0:
|
||||
axis = snap_axis(other["x"] - ep["x"], other["y"] - ep["y"])
|
||||
ux, uy = AXIS[axis]
|
||||
return (ep["x"] + ux * ep["radius"], ep["y"] + uy * ep["radius"]), axis
|
||||
return (ep["x"], ep["y"]), ep["direction_deg"]
|
||||
|
||||
planned: list[dict] = []
|
||||
for index, (where, line) in enumerate(_all_lines(norm)):
|
||||
try:
|
||||
start, end = endpoint(line["from"]), endpoint(line["to"])
|
||||
s, sd = anchor(start, end)
|
||||
e, ed = anchor(end, start)
|
||||
vertices = route(
|
||||
s, sd, e, ed, stub=float(line.get("stub", 5.0)), mode=line.get("route", "auto")
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"{where}: {exc}") from exc
|
||||
crossings = count_crossings(vertices, [p["vertices"] for p in planned])
|
||||
planned.append(
|
||||
{
|
||||
"index": index,
|
||||
"where": where,
|
||||
"from": line["from"],
|
||||
"to": line["to"],
|
||||
"class": line["class"],
|
||||
"vertices": [[x, y] for x, y in vertices],
|
||||
"crossings": crossings,
|
||||
}
|
||||
)
|
||||
return planned
|
||||
|
||||
|
||||
async def _place(backend: AutoCADBackend, section: str, item: dict) -> dict:
|
||||
common = (
|
||||
item["x"],
|
||||
item["y"],
|
||||
item.get("rotation", 0.0),
|
||||
item.get("scale", 1.0),
|
||||
item.get("tag"),
|
||||
)
|
||||
if section == "instruments":
|
||||
return await place_symbol(
|
||||
backend,
|
||||
"instrument",
|
||||
*common,
|
||||
type=item.get("type"),
|
||||
location=item.get("location"),
|
||||
)
|
||||
if section == "connectors":
|
||||
return await place_symbol(
|
||||
backend,
|
||||
"offpage",
|
||||
*common,
|
||||
link=item.get("link"),
|
||||
direction=item["direction"],
|
||||
)
|
||||
return await place_symbol(
|
||||
backend,
|
||||
item["symbol"],
|
||||
*common,
|
||||
desc=item.get("desc"),
|
||||
fail=item.get("fail"),
|
||||
actuator=item.get("actuator"),
|
||||
params=item.get("params"),
|
||||
shape=item.get("shape"),
|
||||
)
|
||||
|
||||
|
||||
async def run_spec(backend: AutoCADBackend, spec: dict, dry_run: bool = False) -> dict:
|
||||
"""Validate, then place every item and draw every line inside one transaction.
|
||||
|
||||
``dry_run`` routes from catalogue ports in memory and touches nothing.
|
||||
"""
|
||||
norm = validate_spec(spec)
|
||||
if dry_run:
|
||||
planned = _plan_lines(norm)
|
||||
return {
|
||||
"dry_run": True,
|
||||
"items": sum(len(norm[s]) for s in _SECTIONS),
|
||||
"lines": planned,
|
||||
"crossings_total": sum(p["crossings"] for p in planned),
|
||||
}
|
||||
sheet = norm["sheet"]
|
||||
if backend.get_plan_spec() is None:
|
||||
await backend.drawing_plan(
|
||||
sheet.get("intent", "P&ID"),
|
||||
sheet.get("sheet_size", "A3"),
|
||||
float(sheet.get("scale", 1.0)),
|
||||
sheet.get("layer_set", "pid"),
|
||||
)
|
||||
begun = await backend.transaction_begin()
|
||||
if not isinstance(begun, dict) or not begun.get("ok"):
|
||||
raise RuntimeError(f"pid_from_spec: could not open a transaction: {begun}")
|
||||
handles: dict[str, str] = {}
|
||||
crossings_total = 0
|
||||
try:
|
||||
for section in _SECTIONS:
|
||||
for i, item in enumerate(norm[section]):
|
||||
try:
|
||||
placed = await _place(backend, section, item)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"{section}[{i}]: {exc}") from exc
|
||||
handles[item["id"]] = placed["handle"]
|
||||
|
||||
def ref(value: str) -> dict:
|
||||
item_id, _, port = value.partition(".")
|
||||
if port:
|
||||
return {"handle": handles[item_id], "port": port}
|
||||
return {"handle": handles[item_id]}
|
||||
|
||||
for where, line in _all_lines(norm):
|
||||
try:
|
||||
drawn = await draw_line(
|
||||
backend,
|
||||
ref(line["from"]),
|
||||
ref(line["to"]),
|
||||
line["class"],
|
||||
line.get("route", "auto"),
|
||||
float(line.get("stub", 5.0)),
|
||||
line.get("number"),
|
||||
line.get("size"),
|
||||
line.get("service"),
|
||||
line.get("spec"),
|
||||
line.get("insulation"),
|
||||
None,
|
||||
bool(line.get("label", True)),
|
||||
line.get("arrow"),
|
||||
)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"{where}: {exc}") from exc
|
||||
crossings_total += drawn["crossings"]
|
||||
except Exception:
|
||||
await backend.transaction_rollback()
|
||||
raise
|
||||
await backend.transaction_commit()
|
||||
graph = await build_graph(backend)
|
||||
critique = [issue.to_dict() for focus in PID_FOCUSES for issue in issues_for(focus, graph)]
|
||||
return {
|
||||
"handles": handles,
|
||||
"graph": graph,
|
||||
"critique": critique,
|
||||
"crossings_total": crossings_total,
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
"""Render the P&ID the README shows, using only the server's own tools.
|
||||
|
||||
The sheet is ``EXAMPLE_SPEC`` from ``engineering/pid/spec.py`` -- the same
|
||||
document ``pid_from_spec`` draws in one transaction -- placed and routed
|
||||
through the same backend methods the MCP tools call, then rendered headlessly.
|
||||
If a symbol builder or the router regresses, the picture changes with it.
|
||||
|
||||
Reproduce:
|
||||
|
||||
AUTOCAD_MCP_BACKEND=ezdxf uv run --frozen python scripts/render_readme_pid.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
DEFAULT_OUTPUT = ROOT / "docs" / "assets" / "pid-showcase.png"
|
||||
|
||||
|
||||
async def render(output: Path) -> dict:
|
||||
from backends.ezdxf_backend import EzdxfBackend
|
||||
from engineering.layers import apply_layer_set
|
||||
from engineering.pid.spec import EXAMPLE_SPEC, run_spec
|
||||
|
||||
backend = EzdxfBackend()
|
||||
await backend.connect()
|
||||
try:
|
||||
await backend.drawing_new()
|
||||
await apply_layer_set(backend, "pid")
|
||||
result = await run_spec(backend, EXAMPLE_SPEC)
|
||||
png = await backend.view_screenshot()
|
||||
if not png:
|
||||
raise RuntimeError("view_screenshot returned no PNG (is matplotlib installed?)")
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
output.write_bytes(png)
|
||||
return {
|
||||
"output": str(output),
|
||||
"bytes": len(png),
|
||||
"dangling": result["graph"]["stats"]["dangling"],
|
||||
"critique": len(result["critique"]),
|
||||
}
|
||||
finally:
|
||||
await backend.disconnect()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
||||
args = parser.parse_args()
|
||||
print(asyncio.run(render(args.output)))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -6369,7 +6369,7 @@ async def solid_boolean(
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ── SECTION 17: P&ID (7 tools) ──────────────────────────────────────────────
|
||||
# ── SECTION 17: P&ID (8 tools) ──────────────────────────────────────────────
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@@ -6681,6 +6681,38 @@ async def pid_equipment_list(
|
||||
)
|
||||
|
||||
|
||||
@cad_tool(
|
||||
summary=(
|
||||
"Draw a whole P&ID from one declarative spec in one transaction; returns graph + critique."
|
||||
),
|
||||
cost="mutate",
|
||||
)
|
||||
@mcp.tool(
|
||||
annotations={"title": "P&ID: From Spec", "readOnlyHint": False},
|
||||
tags={"pid", "create", "batch"},
|
||||
)
|
||||
async def pid_from_spec(
|
||||
spec: Annotated[
|
||||
dict,
|
||||
"{sheet, equipment[], valves[], instruments[], lines[], connectors[]} — see the P&ID prompt",
|
||||
],
|
||||
dry_run: Annotated[bool, "Validate and route in memory; draw nothing"] = False,
|
||||
ctx: Context = None,
|
||||
) -> dict:
|
||||
"""Equipment, valves, instruments and connectors placed, then every line
|
||||
drawn port-to-port, inside one transaction: any bad item rolls the whole
|
||||
sheet back and the error names it (`lines[2].to`). The response carries the
|
||||
graph read back from the drawing and the P&ID critique issues, so the
|
||||
caller sees dangling ends or duplicate tags in the same round trip.
|
||||
`dry_run` returns the planned vertices and crossings without touching the
|
||||
drawing.
|
||||
"""
|
||||
from engineering.pid.spec import run_spec
|
||||
|
||||
await ctx.info("P&ID from spec" + (" (dry run)" if dry_run else ""))
|
||||
return await run_spec(_backend(ctx), spec, dry_run)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ── RESOURCES ───────────────────────────────────────────────────────────────
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""pid_from_spec: one call, one transaction, graph + critique back."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from fastmcp import Client
|
||||
|
||||
import server
|
||||
from engineering.pid.spec import EXAMPLE_SPEC, run_spec, validate_spec
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def test_example_draws_a_connected_pid(backend):
|
||||
result = await run_spec(backend, EXAMPLE_SPEC)
|
||||
assert set(result["handles"]) == {"P-101", "V-201", "FCV-101", "FIC-101", "OP-1"}
|
||||
graph = result["graph"]
|
||||
assert graph["stats"]["nodes_by_kind"] == {
|
||||
"connector": 1,
|
||||
"equipment": 2,
|
||||
"instrument": 1,
|
||||
"valve": 1,
|
||||
}
|
||||
assert len(graph["edges"]) == 4 and graph["stats"]["dangling"] == 0
|
||||
assert result["critique"] == [] and result["crossings_total"] == 0
|
||||
assert backend.get_plan_spec()["layer_set_id"] == "pid"
|
||||
|
||||
|
||||
async def test_bad_reference_rolls_back(backend):
|
||||
before = await backend.entity_count()
|
||||
bad = copy.deepcopy(EXAMPLE_SPEC)
|
||||
bad["lines"].append({"from": "P-101.discharge", "to": "V-201.NOPE", "class": "process_major"})
|
||||
with pytest.raises(ValueError, match=r"lines\[3\]"):
|
||||
await run_spec(backend, bad)
|
||||
assert await backend.entity_count() == before
|
||||
|
||||
|
||||
async def test_dry_run_routes_in_memory_and_writes_nothing(backend):
|
||||
before = await backend.entity_count()
|
||||
result = await run_spec(backend, EXAMPLE_SPEC, dry_run=True)
|
||||
assert result["dry_run"] is True and len(result["lines"]) == 4
|
||||
assert all(len(line["vertices"]) >= 2 for line in result["lines"])
|
||||
assert await backend.entity_count() == before
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mutate, fragment",
|
||||
[
|
||||
(
|
||||
lambda s: s["equipment"].append({"id": "P-101", "symbol": "fan", "x": 0, "y": 0}),
|
||||
"equipment[2].id",
|
||||
),
|
||||
(lambda s: s["lines"][0].update({"class": "steam"}), "lines[0].class"),
|
||||
(lambda s: s["instruments"][0].pop("x"), "instruments[0].x"),
|
||||
(lambda s: s.update({"bogus": []}), "bogus"),
|
||||
(lambda s: s["connectors"][0].update({"to": "V-201.N1"}), "connectors[0]"),
|
||||
],
|
||||
)
|
||||
async def test_validation_names_the_path(mutate, fragment):
|
||||
spec = copy.deepcopy(EXAMPLE_SPEC)
|
||||
mutate(spec)
|
||||
with pytest.raises(ValueError) as exc:
|
||||
validate_spec(spec)
|
||||
assert fragment in str(exc.value)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def client(monkeypatch):
|
||||
monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
|
||||
async with Client(server.mcp) as connected:
|
||||
await connected.call_tool("drawing_new", {})
|
||||
yield connected
|
||||
|
||||
|
||||
async def test_finalize_scores_the_example_at_ninety_or_better(client, tmp_path):
|
||||
result = (await client.call_tool("pid_from_spec", {"spec": EXAMPLE_SPEC})).structured_content
|
||||
assert result["graph"]["stats"]["dangling"] == 0
|
||||
# The validator's first step refuses an unsaved drawing (`not_saved`), so
|
||||
# finalize saves first — the same shape `tests/test_pid_critique.py` uses.
|
||||
final = (
|
||||
await client.call_tool("drawing_finalize", {"save_path": str(tmp_path / "pid.dxf")})
|
||||
).structured_content
|
||||
assert final["score"]["score"] >= 90, final["score"]
|
||||
assert not [i for i in final.get("critique", []) if str(i.get("focus", "")).startswith("pid_")]
|
||||
@@ -512,8 +512,9 @@ async def test_tool_group_sizes_are_unchanged():
|
||||
`pid_symbol_insert`, then 2 -> 3 when `pid_line_draw` joined, 3 -> 4
|
||||
when the reader `pid_graph` joined and 4 -> 7 when the three graph
|
||||
deliverables (`pid_instrument_index` / `pid_line_list` /
|
||||
`pid_equipment_list`) joined; the `pid` tag is ranked first in
|
||||
`_GROUP_TAG_PRIORITY` so those tools file under
|
||||
`pid_equipment_list`) joined and 7 -> 8 when `pid_from_spec` (a whole
|
||||
sheet from one declarative spec, in one transaction) joined; the `pid`
|
||||
tag is ranked first in `_GROUP_TAG_PRIORITY` so those tools file under
|
||||
`pid` rather than under their secondary `query` / `create` tags. Every
|
||||
other number here has been unchanged since the snapshot was taken.
|
||||
"""
|
||||
@@ -531,7 +532,7 @@ async def test_tool_group_sizes_are_unchanged():
|
||||
"entity_query": 9,
|
||||
"layers": 14,
|
||||
"layouts": 12,
|
||||
"pid": 7,
|
||||
"pid": 8,
|
||||
"premium": 12,
|
||||
"solids": 5,
|
||||
"system": 7,
|
||||
@@ -540,4 +541,4 @@ async def test_tool_group_sizes_are_unchanged():
|
||||
"validation": 1,
|
||||
"view": 4,
|
||||
}
|
||||
assert sum(sizes.values()) == 164
|
||||
assert sum(sizes.values()) == 165
|
||||
|
||||
Reference in New Issue
Block a user