From 6cccadee841b766d15a5513e808e0b8f7c04b414 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Tue, 15 Sep 2026 15:14:55 +0300 Subject: [PATCH] =?UTF-8?q?feat(pid):=20symbol=20catalogue=20framework=20?= =?UTF-8?q?=E2=80=94=20instruments,=20markers,=20connectors?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- engineering/pid/geometry.py | 156 +++++++++++ engineering/pid/symbols.py | 404 +++++++++++++++++++++++++++ engineering/pid/symbols_equipment.py | 1 + engineering/pid/symbols_valves.py | 1 + engineering/pid/symbols_vessels.py | 1 + tests/test_pid_symbols.py | 155 ++++++++++ 6 files changed, 718 insertions(+) create mode 100644 engineering/pid/geometry.py create mode 100644 engineering/pid/symbols.py create mode 100644 engineering/pid/symbols_equipment.py create mode 100644 engineering/pid/symbols_valves.py create mode 100644 engineering/pid/symbols_vessels.py create mode 100644 tests/test_pid_symbols.py diff --git a/engineering/pid/geometry.py b/engineering/pid/geometry.py new file mode 100644 index 0000000..12fc410 --- /dev/null +++ b/engineering/pid/geometry.py @@ -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)) diff --git a/engineering/pid/symbols.py b/engineering/pid/symbols.py new file mode 100644 index 0000000..7086690 --- /dev/null +++ b/engineering/pid/symbols.py @@ -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 diff --git a/engineering/pid/symbols_equipment.py b/engineering/pid/symbols_equipment.py new file mode 100644 index 0000000..b0c84a2 --- /dev/null +++ b/engineering/pid/symbols_equipment.py @@ -0,0 +1 @@ +"""Rotating equipment, heat transfer and miscellaneous inline symbols - filled in by Task 9.""" diff --git a/engineering/pid/symbols_valves.py b/engineering/pid/symbols_valves.py new file mode 100644 index 0000000..d242dbf --- /dev/null +++ b/engineering/pid/symbols_valves.py @@ -0,0 +1 @@ +"""Valve bodies and actuators - filled in by Task 8.""" diff --git a/engineering/pid/symbols_vessels.py b/engineering/pid/symbols_vessels.py new file mode 100644 index 0000000..f3342c7 --- /dev/null +++ b/engineering/pid/symbols_vessels.py @@ -0,0 +1 @@ +"""Parametric vessels - filled in by Task 10.""" diff --git a/tests/test_pid_symbols.py b/tests/test_pid_symbols.py new file mode 100644 index 0000000..8db28cb --- /dev/null +++ b/tests/test_pid_symbols.py @@ -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") + )