feat(pid): symbol catalogue framework — instruments, markers, connectors
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
a0b16f399b
commit
6cccadee84
@@ -0,0 +1,156 @@
|
||||
# engineering/pid/geometry.py
|
||||
"""Primitive-spec helpers for authoring symbols (block-local millimetres).
|
||||
|
||||
Each helper returns a ``block_define`` entity/ATTDEF spec (see
|
||||
``backends/block_specs.py``), so a symbol builder composes plain dicts and the
|
||||
same spec is validated and written identically on both engines.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
|
||||
|
||||
def line(x1: float, y1: float, x2: float, y2: float) -> dict:
|
||||
return {"type": "line", "x1": float(x1), "y1": float(y1), "x2": float(x2), "y2": float(y2)}
|
||||
|
||||
|
||||
def circle(cx: float, cy: float, r: float) -> dict:
|
||||
return {"type": "circle", "cx": float(cx), "cy": float(cy), "r": float(r)}
|
||||
|
||||
|
||||
def arc(cx: float, cy: float, r: float, start_deg: float, end_deg: float) -> dict:
|
||||
return {
|
||||
"type": "arc",
|
||||
"cx": float(cx),
|
||||
"cy": float(cy),
|
||||
"r": float(r),
|
||||
"start_deg": float(start_deg),
|
||||
"end_deg": float(end_deg),
|
||||
}
|
||||
|
||||
|
||||
def polyline(points, closed: bool = True, bulges=None) -> dict:
|
||||
spec = {
|
||||
"type": "polyline",
|
||||
"points": [[float(x), float(y)] for x, y in points],
|
||||
"closed": closed,
|
||||
}
|
||||
if bulges is not None:
|
||||
spec["bulges"] = [float(b) for b in bulges]
|
||||
return spec
|
||||
|
||||
|
||||
def text(
|
||||
t: str,
|
||||
x: float,
|
||||
y: float,
|
||||
height: float,
|
||||
align: str = "middle_center",
|
||||
rotation_deg: float = 0.0,
|
||||
) -> dict:
|
||||
return {
|
||||
"type": "text",
|
||||
"text": t,
|
||||
"x": float(x),
|
||||
"y": float(y),
|
||||
"height": float(height),
|
||||
"align": align,
|
||||
"rotation_deg": float(rotation_deg),
|
||||
}
|
||||
|
||||
|
||||
def solid(points) -> dict:
|
||||
return {"type": "solid", "points": [[float(x), float(y)] for x, y in points]}
|
||||
|
||||
|
||||
def dashed_line(
|
||||
x1: float, y1: float, x2: float, y2: float, on: float = 1.0, off: float = 1.0
|
||||
) -> list[dict]:
|
||||
"""A dashed line as short LINE primitives, so the block needs no linetype."""
|
||||
length = math.hypot(x2 - x1, y2 - y1)
|
||||
if length <= 0:
|
||||
raise ValueError("dashed_line: zero-length line")
|
||||
if on <= 0 or off < 0:
|
||||
raise ValueError("dashed_line: 'on' must be > 0 and 'off' >= 0")
|
||||
ux, uy = (x2 - x1) / length, (y2 - y1) / length
|
||||
out: list[dict] = []
|
||||
pos = 0.0
|
||||
while pos < length - 1e-9:
|
||||
end = min(pos + on, length)
|
||||
out.append(line(x1 + ux * pos, y1 + uy * pos, x1 + ux * end, y1 + uy * end))
|
||||
pos = end + off
|
||||
return out
|
||||
|
||||
|
||||
def regular_polygon(
|
||||
cx: float, cy: float, r: float, n: int, start_deg: float = 0.0
|
||||
) -> list[tuple[float, float]]:
|
||||
return [
|
||||
(
|
||||
cx + r * math.cos(math.radians(start_deg + 360.0 * i / n)),
|
||||
cy + r * math.sin(math.radians(start_deg + 360.0 * i / n)),
|
||||
)
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
|
||||
def attdef(
|
||||
tag: str,
|
||||
x: float,
|
||||
y: float,
|
||||
height: float,
|
||||
align: str = "center",
|
||||
invisible: bool = False,
|
||||
default: str = "",
|
||||
) -> dict:
|
||||
return {
|
||||
"tag": tag,
|
||||
"prompt": tag,
|
||||
"default": default,
|
||||
"x": float(x),
|
||||
"y": float(y),
|
||||
"height": float(height),
|
||||
"align": align,
|
||||
"invisible": invisible,
|
||||
}
|
||||
|
||||
|
||||
def _arc_extent(cx, cy, r, start_deg, end_deg):
|
||||
"""Points along the arc at 1° steps plus both ends (CCW from start to end)."""
|
||||
pts = []
|
||||
sweep = (end_deg - start_deg) % 360.0 or 360.0
|
||||
for k in range(0, int(sweep) + 1):
|
||||
a = math.radians(start_deg + k)
|
||||
pts.append((cx + r * math.cos(a), cy + r * math.sin(a)))
|
||||
a = math.radians(end_deg)
|
||||
pts.append((cx + r * math.cos(a), cy + r * math.sin(a)))
|
||||
return pts
|
||||
|
||||
|
||||
def bbox_of(primitives) -> tuple[float, float, float, float]:
|
||||
"""Extent of the drawn geometry (text counts by its insertion point only)."""
|
||||
xs: list[float] = []
|
||||
ys: list[float] = []
|
||||
for p in primitives:
|
||||
kind = p["type"]
|
||||
if kind == "line":
|
||||
xs += [p["x1"], p["x2"]]
|
||||
ys += [p["y1"], p["y2"]]
|
||||
elif kind == "circle":
|
||||
xs += [p["cx"] - p["r"], p["cx"] + p["r"]]
|
||||
ys += [p["cy"] - p["r"], p["cy"] + p["r"]]
|
||||
elif kind == "arc":
|
||||
for x, y in _arc_extent(p["cx"], p["cy"], p["r"], p["start_deg"], p["end_deg"]):
|
||||
xs.append(x)
|
||||
ys.append(y)
|
||||
elif kind in ("polyline", "solid"):
|
||||
for x, y in p["points"]:
|
||||
xs.append(x)
|
||||
ys.append(y)
|
||||
elif kind == "text":
|
||||
xs.append(p["x"])
|
||||
ys.append(p["y"])
|
||||
if not xs:
|
||||
return (0.0, 0.0, 0.0, 0.0)
|
||||
return (min(xs), min(ys), max(xs), max(ys))
|
||||
@@ -0,0 +1,404 @@
|
||||
# engineering/pid/symbols.py
|
||||
"""The P&ID symbol catalogue: symbols as data, blocks on demand.
|
||||
|
||||
Every symbol is a ``SymbolSpec`` — typed primitives in symbol-local mm, ATTDEFs,
|
||||
named ports with an outward direction, a declared bbox — authored from
|
||||
ISO 10628-2 (equipment, valves) or ISA-5.1 (instrumentation). Builders live in
|
||||
``symbols_valves.py``, ``symbols_equipment.py`` and ``symbols_vessels.py`` and
|
||||
register here; this module holds the framework plus the instrument bubble,
|
||||
markers and connectors.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .geometry import (
|
||||
attdef,
|
||||
bbox_of,
|
||||
circle,
|
||||
dashed_line,
|
||||
line,
|
||||
polyline,
|
||||
regular_polygon,
|
||||
solid,
|
||||
)
|
||||
|
||||
#: Written into every INSERT's XDATA; bump on any change to shipped geometry or ports.
|
||||
CATALOG_VERSION = "1"
|
||||
BUBBLE_RADIUS = 5.0
|
||||
FAMILY_LAYER = {
|
||||
"valve": "PROCESS-VALVES",
|
||||
"rotating": "PROCESS-EQUIPMENT",
|
||||
"vessel": "PROCESS-EQUIPMENT",
|
||||
"heat": "PROCESS-EQUIPMENT",
|
||||
"misc": "PROCESS-EQUIPMENT",
|
||||
"instrument": "INSTRUMENT-SYMBOL",
|
||||
"connector": "PROCESS-EQUIPMENT",
|
||||
"marker": "line",
|
||||
}
|
||||
INSTRUMENT_TYPES = ("discrete", "dcs", "computer", "plc")
|
||||
INSTRUMENT_LOCATIONS = ("field", "primary", "auxiliary", "primary_rear", "auxiliary_rear")
|
||||
TAG_HEIGHT = 2.5
|
||||
DESC_HEIGHT = 2.0
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Port:
|
||||
name: str
|
||||
x: float
|
||||
y: float
|
||||
direction_deg: float | None # outward direction a line leaves along; None = radial
|
||||
kind: str = "process" # "process" | "signal"
|
||||
radius: float = 0.0 # radial ports only
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"name": self.name,
|
||||
"x": self.x,
|
||||
"y": self.y,
|
||||
"direction_deg": self.direction_deg,
|
||||
"kind": self.kind,
|
||||
"radius": self.radius,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SymbolSpec:
|
||||
name: str
|
||||
family: str
|
||||
symbol: str
|
||||
variant: str | None
|
||||
primitives: tuple[dict, ...]
|
||||
attdefs: tuple[dict, ...]
|
||||
ports: tuple[Port, ...]
|
||||
bbox: tuple[float, float, float, float]
|
||||
layer_class: str
|
||||
source: str
|
||||
parametric: bool = False
|
||||
params: dict = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"name": self.name,
|
||||
"family": self.family,
|
||||
"symbol": self.symbol,
|
||||
"variant": self.variant,
|
||||
"ports": [p.to_dict() for p in self.ports],
|
||||
"bbox": list(self.bbox),
|
||||
"layer_class": self.layer_class,
|
||||
"source": self.source,
|
||||
"parametric": self.parametric,
|
||||
"params": dict(self.params),
|
||||
}
|
||||
|
||||
|
||||
Builder = Callable[..., SymbolSpec]
|
||||
_REGISTRY: dict[str, tuple[str, Builder, dict[str, list[str]], dict | None]] = {}
|
||||
|
||||
|
||||
def register(
|
||||
symbol: str,
|
||||
family: str,
|
||||
builder: Builder,
|
||||
variants: dict[str, list[str]] | None = None,
|
||||
params_schema: dict | None = None,
|
||||
) -> None:
|
||||
if family not in FAMILY_LAYER:
|
||||
raise ValueError(f"unknown family {family!r}")
|
||||
_REGISTRY[symbol] = (family, builder, dict(variants or {}), params_schema)
|
||||
|
||||
|
||||
def resolve(symbol: str, **options) -> SymbolSpec:
|
||||
"""Build the SymbolSpec for ``symbol`` with the given variant options."""
|
||||
if symbol not in _REGISTRY:
|
||||
raise ValueError(
|
||||
f"unknown symbol {symbol!r}; valid symbols: {', '.join(sorted(_REGISTRY))}"
|
||||
)
|
||||
family, builder, variants, schema = _REGISTRY[symbol]
|
||||
for key, value in options.items():
|
||||
if value is None:
|
||||
continue
|
||||
if key == "params":
|
||||
if schema is None:
|
||||
raise ValueError(f"{symbol} is not parametric; 'params' is only for vessels")
|
||||
continue
|
||||
if key not in variants:
|
||||
raise ValueError(f"{symbol} takes no option {key!r}")
|
||||
if value not in variants[key]:
|
||||
raise ValueError(
|
||||
f"{symbol}: {key} must be one of {', '.join(variants[key])}, got {value!r}"
|
||||
)
|
||||
return builder(**{k: v for k, v in options.items() if v is not None})
|
||||
|
||||
|
||||
def list_symbols(family: str | None = None) -> list[dict]:
|
||||
rows = []
|
||||
for symbol, (fam, builder, variants, schema) in sorted(_REGISTRY.items()):
|
||||
if family and fam != family:
|
||||
continue
|
||||
base = builder()
|
||||
rows.append(
|
||||
{
|
||||
"symbol": symbol,
|
||||
"family": fam,
|
||||
"variants": variants,
|
||||
"ports": [p.to_dict() for p in base.ports],
|
||||
"parametric": base.parametric,
|
||||
"params_schema": schema or {},
|
||||
"source": base.source,
|
||||
"block_name": base.name,
|
||||
}
|
||||
)
|
||||
return rows
|
||||
|
||||
|
||||
def all_specs() -> list[SymbolSpec]:
|
||||
"""Every base symbol plus every enumerable variant (for tests and the contact sheet)."""
|
||||
specs: list[SymbolSpec] = []
|
||||
for _symbol, (_fam, builder, variants, _schema) in sorted(_REGISTRY.items()):
|
||||
if not variants:
|
||||
specs.append(builder())
|
||||
continue
|
||||
keys = sorted(variants)
|
||||
combos: list[dict] = [{}]
|
||||
for key in keys:
|
||||
combos = [{**c, key: v} for c in combos for v in variants[key]]
|
||||
for combo in combos:
|
||||
try:
|
||||
specs.append(builder(**combo))
|
||||
except ValueError:
|
||||
continue # a combination the builder refuses (e.g. an actuator on a check valve)
|
||||
return specs
|
||||
|
||||
|
||||
def transform_port(port: Port, x: float, y: float, rotation_deg: float, scale: float) -> dict:
|
||||
"""Symbol-local port -> WCS, for an INSERT at (x, y) rotated by ``rotation_deg``
|
||||
and uniformly scaled."""
|
||||
a = math.radians(rotation_deg)
|
||||
lx, ly = port.x * scale, port.y * scale
|
||||
wx = x + lx * math.cos(a) - ly * math.sin(a)
|
||||
wy = y + lx * math.sin(a) + ly * math.cos(a)
|
||||
direction = None if port.direction_deg is None else (port.direction_deg + rotation_deg) % 360.0
|
||||
return {
|
||||
"name": port.name,
|
||||
"x": wx,
|
||||
"y": wy,
|
||||
"direction_deg": direction,
|
||||
"kind": port.kind,
|
||||
"radius": port.radius * scale,
|
||||
"inferred": False,
|
||||
}
|
||||
|
||||
|
||||
def params_hash(params: dict) -> str:
|
||||
"""First 8 hex chars of SHA-256 over the canonical JSON of ``params`` (spec 4.5)."""
|
||||
canonical = json.dumps(params, sort_keys=True, separators=(",", ":"))
|
||||
return hashlib.sha256(canonical.encode()).hexdigest()[:8].upper()
|
||||
|
||||
|
||||
def tag_attdefs(ymax: float, with_desc: bool = True) -> tuple[dict, ...]:
|
||||
"""``TAG`` (and ``DESC``) above the body, per spec 4.7."""
|
||||
out = [attdef("TAG", 0.0, ymax + 3.5, TAG_HEIGHT)]
|
||||
if with_desc:
|
||||
out.append(attdef("DESC", 0.0, ymax + 6.5, DESC_HEIGHT))
|
||||
return tuple(out)
|
||||
|
||||
|
||||
def make_spec(
|
||||
name: str,
|
||||
family: str,
|
||||
symbol: str,
|
||||
variant: str | None,
|
||||
primitives,
|
||||
attdefs,
|
||||
ports,
|
||||
source: str,
|
||||
bbox=None,
|
||||
parametric: bool = False,
|
||||
params: dict | None = None,
|
||||
) -> SymbolSpec:
|
||||
prims = tuple(primitives)
|
||||
return SymbolSpec(
|
||||
name=name,
|
||||
family=family,
|
||||
symbol=symbol,
|
||||
variant=variant,
|
||||
primitives=prims,
|
||||
attdefs=tuple(attdefs),
|
||||
ports=tuple(ports),
|
||||
bbox=tuple(bbox) if bbox else bbox_of(prims),
|
||||
layer_class=FAMILY_LAYER[family],
|
||||
source=source,
|
||||
parametric=parametric,
|
||||
params=dict(params or {}),
|
||||
)
|
||||
|
||||
|
||||
# ── instrument bubble (ISA-5.1-2009 Table 5.4.1) ─────────────────────────────
|
||||
|
||||
|
||||
def build_instrument(type: str = "discrete", location: str = "field") -> SymbolSpec:
|
||||
if type not in INSTRUMENT_TYPES:
|
||||
raise ValueError(f"instrument: type must be one of {', '.join(INSTRUMENT_TYPES)}")
|
||||
if location not in INSTRUMENT_LOCATIONS:
|
||||
raise ValueError(f"instrument: location must be one of {', '.join(INSTRUMENT_LOCATIONS)}")
|
||||
r = BUBBLE_RADIUS
|
||||
prims: list[dict] = [circle(0, 0, r)]
|
||||
square = polyline([(-r, -r), (r, -r), (r, r), (-r, r)])
|
||||
if type == "dcs":
|
||||
prims.append(square)
|
||||
elif type == "computer":
|
||||
prims.append(polyline(regular_polygon(0, 0, r, 6, 0.0)))
|
||||
elif type == "plc":
|
||||
prims += [square, polyline([(0, r), (r, 0), (0, -r), (-r, 0)])]
|
||||
ys = {
|
||||
"field": [],
|
||||
"primary": [0.0],
|
||||
"auxiliary": [0.8, -0.8],
|
||||
"primary_rear": [0.0],
|
||||
"auxiliary_rear": [0.8, -0.8],
|
||||
}[location]
|
||||
for y in ys:
|
||||
prims += dashed_line(-r, y, r, y) if location.endswith("_rear") else [line(-r, y, r, y)]
|
||||
attdefs = (
|
||||
attdef("FUNC", 0.0, 2.0, DESC_HEIGHT, align="middle_center"),
|
||||
attdef("LOOP", 0.0, -2.0, DESC_HEIGHT, align="middle_center"),
|
||||
)
|
||||
ports = (Port("signal", 0.0, 0.0, None, "signal", r),)
|
||||
return make_spec(
|
||||
f"PID_INST_{type.upper()}_{location.upper()}",
|
||||
"instrument",
|
||||
"instrument",
|
||||
f"{type}_{location}",
|
||||
prims,
|
||||
attdefs,
|
||||
ports,
|
||||
"ISA-5.1-2009 Table 5.4.1",
|
||||
bbox=(-r, -r, r, r),
|
||||
)
|
||||
|
||||
|
||||
register(
|
||||
"instrument",
|
||||
"instrument",
|
||||
build_instrument,
|
||||
variants={"type": list(INSTRUMENT_TYPES), "location": list(INSTRUMENT_LOCATIONS)},
|
||||
)
|
||||
|
||||
|
||||
# ── markers (decoration on a line; never graph nodes) ────────────────────────
|
||||
|
||||
MARKERS = ("arrow_flow", "mark_pneumatic", "mark_capillary", "mark_hydraulic", "mark_data")
|
||||
|
||||
|
||||
def build_marker(which: str) -> SymbolSpec:
|
||||
shapes = {
|
||||
"arrow_flow": [solid([(0, 0), (-3, 1.2), (-3, -1.2)])],
|
||||
"mark_pneumatic": [line(-1.5, -1.5, 0, 1.5), line(0.5, -1.5, 2, 1.5)],
|
||||
"mark_capillary": [line(-1.2, -1.2, 1.2, 1.2), line(-1.2, 1.2, 1.2, -1.2)],
|
||||
"mark_hydraulic": [line(-1, 1.5, -1, -1.5), line(-1, -1.5, 1.5, -1.5)],
|
||||
"mark_data": [circle(0, 0, 0.8)],
|
||||
}
|
||||
if which not in shapes:
|
||||
raise ValueError(f"unknown marker {which!r}; valid markers: {', '.join(MARKERS)}")
|
||||
return make_spec(
|
||||
f"PID_MARKER_{which.upper()}",
|
||||
"marker",
|
||||
which,
|
||||
None,
|
||||
shapes[which],
|
||||
(),
|
||||
(),
|
||||
"ISA-5.1-2009 Table 5.3.1",
|
||||
)
|
||||
|
||||
|
||||
def _marker_builder(which: str) -> Builder:
|
||||
def build() -> SymbolSpec:
|
||||
return build_marker(which)
|
||||
|
||||
return build
|
||||
|
||||
|
||||
for _which in MARKERS:
|
||||
register(_which, "marker", _marker_builder(_which))
|
||||
|
||||
|
||||
# ── connectors ───────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def build_connector(which: str = "offpage", direction: str = "out") -> SymbolSpec:
|
||||
if which == "offpage":
|
||||
if direction == "out":
|
||||
pts = [(-8, -3), (4, -3), (8, 0), (4, 3), (-8, 3)]
|
||||
port = Port("process", -8.0, 0.0, 180.0, "process")
|
||||
elif direction == "in":
|
||||
pts = [(8, -3), (-4, -3), (-8, 0), (-4, 3), (8, 3)]
|
||||
port = Port("process", 8.0, 0.0, 0.0, "process")
|
||||
else:
|
||||
raise ValueError(f"offpage: direction must be one of in, out, got {direction!r}")
|
||||
attdefs = (
|
||||
attdef(
|
||||
"TAG",
|
||||
-2.0 if direction == "out" else 2.0,
|
||||
0.0,
|
||||
DESC_HEIGHT,
|
||||
align="middle_center",
|
||||
),
|
||||
attdef("LINK", 0.0, -5.0, DESC_HEIGHT, invisible=True),
|
||||
)
|
||||
return make_spec(
|
||||
f"PID_CONNECTOR_OFFPAGE_{direction.upper()}",
|
||||
"connector",
|
||||
"offpage",
|
||||
direction,
|
||||
[polyline(pts)],
|
||||
attdefs,
|
||||
(port,),
|
||||
"PIP PIC001 off-page connector",
|
||||
bbox=(-8, -3, 8, 3),
|
||||
)
|
||||
if which == "tie_in":
|
||||
return make_spec(
|
||||
"PID_CONNECTOR_TIE_IN",
|
||||
"connector",
|
||||
"tie_in",
|
||||
None,
|
||||
[circle(0, 0, 3)],
|
||||
(attdef("TAG", 0.0, 0.0, DESC_HEIGHT, align="middle_center"),),
|
||||
(Port("in", -3.0, 0.0, 180.0), Port("out", 3.0, 0.0, 0.0)),
|
||||
"PIP PIC001 tie-in point",
|
||||
)
|
||||
if which == "spec_break":
|
||||
return make_spec(
|
||||
"PID_CONNECTOR_SPEC_BREAK",
|
||||
"connector",
|
||||
"spec_break",
|
||||
None,
|
||||
[line(0, -3, 0, 3)],
|
||||
(attdef("TAG", 0.0, 4.5, DESC_HEIGHT),),
|
||||
(Port("in", -0.3, 0.0, 180.0), Port("out", 0.3, 0.0, 0.0)),
|
||||
"PIP PIC001 specification break",
|
||||
bbox=(-0.3, -3, 0.3, 4.5),
|
||||
)
|
||||
raise ValueError(f"unknown connector {which!r}")
|
||||
|
||||
|
||||
register(
|
||||
"offpage",
|
||||
"connector",
|
||||
lambda direction="out": build_connector("offpage", direction),
|
||||
variants={"direction": ["in", "out"]},
|
||||
)
|
||||
register("tie_in", "connector", lambda: build_connector("tie_in"))
|
||||
register("spec_break", "connector", lambda: build_connector("spec_break"))
|
||||
|
||||
# Builders for valves, equipment and vessels register themselves on import.
|
||||
from . import symbols_equipment as _symbols_equipment # noqa: E402,F401
|
||||
from . import symbols_valves as _symbols_valves # noqa: E402,F401
|
||||
from . import symbols_vessels as _symbols_vessels # noqa: E402,F401
|
||||
@@ -0,0 +1 @@
|
||||
"""Rotating equipment, heat transfer and miscellaneous inline symbols - filled in by Task 9."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Valve bodies and actuators - filled in by Task 8."""
|
||||
@@ -0,0 +1 @@
|
||||
"""Parametric vessels - filled in by Task 10."""
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Every catalogue symbol: valid primitives, honest bbox, ports on the boundary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
from backends.block_specs import validate_attdef_specs, validate_entity_specs
|
||||
from engineering.pid.geometry import bbox_of
|
||||
from engineering.pid.symbols import (
|
||||
BUBBLE_RADIUS,
|
||||
CATALOG_VERSION,
|
||||
INSTRUMENT_LOCATIONS,
|
||||
INSTRUMENT_TYPES,
|
||||
Port,
|
||||
all_specs,
|
||||
list_symbols,
|
||||
resolve,
|
||||
transform_port,
|
||||
)
|
||||
|
||||
NAME_RE = re.compile(r"^PID_[A-Z0-9_]+$")
|
||||
SOURCE_RE = re.compile(r"^(ISO 10628-2|ISA-5\.1|PIP PIC001)")
|
||||
EPS = 1e-6
|
||||
|
||||
|
||||
def _ids():
|
||||
return [spec.name for spec in all_specs()]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("spec", all_specs(), ids=_ids())
|
||||
def test_primitives_and_attdefs_validate(spec):
|
||||
validate_entity_specs(list(spec.primitives))
|
||||
validate_attdef_specs(list(spec.attdefs))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("spec", all_specs(), ids=_ids())
|
||||
def test_declared_bbox_contains_every_primitive(spec):
|
||||
xmin, ymin, xmax, ymax = bbox_of(spec.primitives)
|
||||
dxmin, dymin, dxmax, dymax = spec.bbox
|
||||
assert dxmin - EPS <= xmin and dymin - EPS <= ymin
|
||||
assert xmax <= dxmax + EPS and ymax <= dymax + EPS
|
||||
assert dxmax > dxmin and dymax > dymin
|
||||
|
||||
|
||||
@pytest.mark.parametrize("spec", all_specs(), ids=_ids())
|
||||
def test_ports_sit_on_the_boundary_or_at_the_centre(spec):
|
||||
if spec.family == "marker":
|
||||
assert spec.ports == ()
|
||||
return
|
||||
assert spec.ports, "every non-marker symbol needs at least one port"
|
||||
xmin, ymin, xmax, ymax = spec.bbox
|
||||
for port in spec.ports:
|
||||
if port.direction_deg is None:
|
||||
assert port.radius > 0 and (port.x, port.y) == (0.0, 0.0)
|
||||
continue
|
||||
on_edge = (
|
||||
abs(port.x - xmin) < EPS
|
||||
or abs(port.x - xmax) < EPS
|
||||
or abs(port.y - ymin) < EPS
|
||||
or abs(port.y - ymax) < EPS
|
||||
)
|
||||
assert on_edge, (
|
||||
f"{spec.name}.{port.name} at ({port.x}, {port.y}) is inside bbox {spec.bbox}"
|
||||
)
|
||||
assert port.direction_deg in (0.0, 90.0, 180.0, 270.0)
|
||||
assert port.kind in ("process", "signal")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("spec", all_specs(), ids=_ids())
|
||||
def test_naming_source_and_layer(spec):
|
||||
assert NAME_RE.match(spec.name)
|
||||
assert SOURCE_RE.match(spec.source), spec.source
|
||||
assert spec.layer_class in (
|
||||
"PROCESS-VALVES",
|
||||
"PROCESS-EQUIPMENT",
|
||||
"INSTRUMENT-SYMBOL",
|
||||
"line",
|
||||
)
|
||||
tags = [a["tag"] for a in spec.attdefs]
|
||||
assert len(tags) == len(set(tags))
|
||||
for prim in spec.primitives:
|
||||
assert prim.get("layer", "0") == "0"
|
||||
|
||||
|
||||
def test_catalogue_names_are_unique_and_versioned():
|
||||
names = [s.name for s in all_specs()]
|
||||
assert len(names) == len(set(names))
|
||||
assert CATALOG_VERSION == "1"
|
||||
|
||||
|
||||
def test_instrument_generator_covers_four_types_by_five_locations():
|
||||
specs = {
|
||||
(t, loc): resolve("instrument", type=t, location=loc)
|
||||
for t in INSTRUMENT_TYPES
|
||||
for loc in INSTRUMENT_LOCATIONS
|
||||
}
|
||||
assert len(specs) == 20
|
||||
field = specs[("discrete", "field")]
|
||||
assert field.name == "PID_INST_DISCRETE_FIELD"
|
||||
assert [a["tag"] for a in field.attdefs] == ["FUNC", "LOOP"]
|
||||
assert field.ports == (Port("signal", 0.0, 0.0, None, "signal", BUBBLE_RADIUS),)
|
||||
assert sum(1 for p in field.primitives if p["type"] == "circle") == 1
|
||||
primary = specs[("discrete", "primary")]
|
||||
assert sum(1 for p in primary.primitives if p["type"] == "line") == 1
|
||||
aux_rear = specs[("plc", "auxiliary_rear")]
|
||||
assert sum(1 for p in aux_rear.primitives if p["type"] == "line") > 2, "dashed pair"
|
||||
assert sum(1 for p in aux_rear.primitives if p["type"] == "polyline") == 2, "square + diamond"
|
||||
hexagon = next(p for p in specs[("computer", "field")].primitives if p["type"] == "polyline")
|
||||
assert len(hexagon["points"]) == 6
|
||||
|
||||
|
||||
def test_unknown_symbol_or_option_names_the_valid_values():
|
||||
with pytest.raises(ValueError, match="instrument"):
|
||||
resolve("thermometer")
|
||||
with pytest.raises(ValueError, match="field"):
|
||||
resolve("instrument", type="dcs", location="ceiling")
|
||||
|
||||
|
||||
def test_markers_and_connectors():
|
||||
arrow = resolve("arrow_flow")
|
||||
assert arrow.family == "marker" and arrow.layer_class == "line"
|
||||
assert any(p["type"] == "solid" for p in arrow.primitives)
|
||||
for which in ("mark_pneumatic", "mark_capillary", "mark_hydraulic", "mark_data"):
|
||||
assert resolve(which).family == "marker"
|
||||
out = resolve("offpage", direction="out")
|
||||
inn = resolve("offpage", direction="in")
|
||||
assert out.name == "PID_CONNECTOR_OFFPAGE_OUT" and inn.name == "PID_CONNECTOR_OFFPAGE_IN"
|
||||
assert out.ports[0] == Port("process", -8.0, 0.0, 180.0, "process")
|
||||
assert inn.ports[0] == Port("process", 8.0, 0.0, 0.0, "process")
|
||||
assert [a["tag"] for a in out.attdefs] == ["TAG", "LINK"]
|
||||
assert next(a for a in out.attdefs if a["tag"] == "LINK")["invisible"] is True
|
||||
|
||||
|
||||
def test_transform_port_applies_rotation_and_scale_about_the_insertion():
|
||||
port = Port("out", 4.0, 0.0, 0.0, "process")
|
||||
moved = transform_port(port, 100.0, 50.0, 90.0, 2.0)
|
||||
assert moved["x"] == pytest.approx(100.0) and moved["y"] == pytest.approx(58.0)
|
||||
assert moved["direction_deg"] == pytest.approx(90.0)
|
||||
radial = transform_port(Port("signal", 0.0, 0.0, None, "signal", 5.0), 10.0, 10.0, 45.0, 3.0)
|
||||
assert radial["direction_deg"] is None and radial["radius"] == pytest.approx(15.0)
|
||||
|
||||
|
||||
def test_list_symbols_reports_variants_and_ports():
|
||||
rows = {row["symbol"]: row for row in list_symbols()}
|
||||
assert rows["instrument"]["variants"] == {
|
||||
"type": list(INSTRUMENT_TYPES),
|
||||
"location": list(INSTRUMENT_LOCATIONS),
|
||||
}
|
||||
assert rows["offpage"]["variants"] == {"direction": ["in", "out"]}
|
||||
assert rows["arrow_flow"]["ports"] == []
|
||||
assert list_symbols(family="marker") and all(
|
||||
r["family"] == "marker" for r in list_symbols(family="marker")
|
||||
)
|
||||
Reference in New Issue
Block a user