From b8de3968f98ecd8ddcf0710983c5af18b1fe2fb8 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Tue, 15 Sep 2026 13:14:12 +0300 Subject: [PATCH] feat(pid): ISA-5.1 identification-letter grammar with a pinned fixture Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- engineering/pid/tags.py | 364 +++++++++++++++++++++++++++++++++++++ tests/data/isa51_tags.json | 85 +++++++++ tests/test_pid_tags.py | 87 +++++++++ 3 files changed, 536 insertions(+) create mode 100644 engineering/pid/tags.py create mode 100644 tests/data/isa51_tags.json create mode 100644 tests/test_pid_tags.py diff --git a/engineering/pid/tags.py b/engineering/pid/tags.py new file mode 100644 index 0000000..bba8e57 --- /dev/null +++ b/engineering/pid/tags.py @@ -0,0 +1,364 @@ +"""ISA-5.1-2009 identification letters (Table 4.1) as a deterministic grammar. + + tag := first modifier? readout* output* trailing? + modifier := D | F | K | Q | S + readout := A | B | E | G | I | L | N | O | P | R | U | W | X + output := C | K | S | T | U | V | X | Y | Z + trailing := HH | LL | H | L | M | O | C + +The disambiguation rules (safety S vs switch S, trailing L vs light L, ZSO/ZSC, +ratio F only after F, rate-of-change K only before an output) are the ones in +the design spec; the fixture ``tests/data/isa51_tags.json`` pins every one of +them against the standard's own examples. +""" + +from __future__ import annotations + +import re + +#: First letter → measured/initiating variable. ``None`` = user's choice. +FIRST_LETTERS: dict[str, str | None] = { + "A": "Analysis", + "B": "Burner", + "C": None, + "D": None, + "E": "Voltage", + "F": "Flow", + "G": None, + "H": "Hand", + "I": "Current", + "J": "Power", + "K": "Time", + "L": "Level", + "M": None, + "N": None, + "O": None, + "P": "Pressure", + "Q": "Quantity", + "R": "Radiation", + "S": "Speed", + "T": "Temperature", + "U": "Multivariable", + "V": "Vibration", + "W": "Weight", + "X": "Unclassified", + "Y": "Event", + "Z": "Position", +} +MODIFIERS: dict[str, str] = { + "D": "Differential", + "F": "Ratio", + "K": "Rate of change", + "Q": "Totalizing", + "S": "Safety", +} +#: Readout / passive function letters → (adjective form, noun form). ``None`` = user's choice. +READOUT: dict[str, tuple[str, str] | None] = { + "A": ("Alarm", "Alarm"), + "B": None, + "E": ("Element", "Element"), + "G": ("Gauge", "Gauge"), + "I": ("Indicating", "Indicator"), + "L": ("Light", "Light"), + "N": None, + "O": ("Orifice", "Orifice"), + "P": ("Test point", "Test point"), + "R": ("Recording", "Recorder"), + "U": ("Multifunction", "Multifunction"), + "W": ("Well", "Well"), + "X": None, +} +#: Output / active function letters → word. ``None`` = user's choice. +OUTPUT: dict[str, str | None] = { + "C": "Controller", + "K": "Control station", + "S": "Switch", + "T": "Transmitter", + "U": "Multifunction", + "V": "Valve", + "X": None, + "Y": "Relay", + "Z": "Driver", +} +TRAILING: dict[str, str] = { + "HH": "High-high", + "LL": "Low-low", + "H": "High", + "L": "Low", + "M": "Middle", + "O": "Open", + "C": "Closed", +} +DEFAULT_EQUIPMENT_PREFIXES: dict[str, str] = { + "P": "pump", + "V": "vessel", + "T": "tank", + "E": "exchanger", + "K": "compressor", + "C": "column", + "R": "reactor", + "F": "filter", + "D": "drum", + "M": "mixer", + "H": "heater", + "S": "separator", + "B": "blower", + "X": "package", +} + +_INSTRUMENT_RE = re.compile( + r"^(?:(?P\d{1,4})-)?(?P[A-Z]{1,5})[- ]?(?P\d{1,5})(?:[- ]?(?P[A-Z]))?$" +) +_EQUIPMENT_RE = re.compile( + r"^(?:(?P\d{1,4})-)?(?P[A-Z]{1,3})-(?P\d{1,5})(?P[A-Z]{0,2})$" +) + + +def _normalise(tag) -> str: + return str(tag or "").strip().upper() + + +def _parse_letters(letters: str) -> dict: + """Split ``letters`` into first/modifier/readout/output/trailing or report the first error.""" + errors: list[str] = [] + first = letters[0] + i, n = 1, len(letters) + if n == 1: + return {"errors": [f"{letters}: a tag needs at least one function letter after {first}"]} + modifier = None + if letters[i] in MODIFIERS: + candidate = letters[i] + following = letters[i + 1] if i + 1 < n else "" + take = ( + candidate in ("D", "Q") + or (candidate == "S" and following in ("V", "E")) + or (candidate == "F" and first == "F") + or (candidate == "K" and following in OUTPUT) + ) + if take: + modifier = candidate + i += 1 + readout: list[str] = [] + while i < n and letters[i] in READOUT: + letter = letters[i] + if letter in ("H", "L", "M") and readout and readout[-1] == "A": + break # trailing high/low/middle after an alarm + if readout and readout[-1] == letter: + errors.append(f"{letters}: repeated letter {letter} at position {i + 1}") + return {"errors": errors} + readout.append(letter) + i += 1 + output: list[str] = [] + while i < n and letters[i] in OUTPUT: + letter = letters[i] + if letter == "C" and output and output[-1] == "S" and i == n - 1: + break # ZSC: closed, not controller + if output and output[-1] == letter: + errors.append(f"{letters}: repeated letter {letter} at position {i + 1}") + return {"errors": errors} + if output and output[-1] == "V": + # A valve is a final control element: nothing acts after it (PSVX is not a tag). + errors.append( + f"{letters}: unexpected letter {letter} at position {i + 1} after valve V" + ) + return {"errors": errors} + output.append(letter) + i += 1 + trailing = letters[i:] + if trailing: + if trailing not in TRAILING: + errors.append(f"{letters}: unexpected letter {trailing[0]} at position {i + 1}") + return {"errors": errors} + after_alarm = bool(readout) and readout[-1] == "A" and not output + after_switch = bool(output) and output[-1] == "S" + if trailing in ("O", "C") and not (after_switch and first == "Z"): + # Open/closed are position states: ZSO / ZSC only (LSO is not a tag). + errors.append( + f"{letters}: trailing {trailing} (open/closed) is only valid after a switch " + "on a position (Z) tag" + ) + return {"errors": errors} + if trailing in ("H", "L", "M", "HH", "LL") and not (after_alarm or after_switch): + errors.append( + f"{letters}: trailing {trailing} needs an alarm (A) or switch (S) before it" + ) + return {"errors": errors} + return { + "first": first, + "modifier": modifier, + "readout": readout, + "output": output, + "trailing": trailing or None, + "errors": [], + } + + +def _describe(parts: dict) -> tuple[str, bool]: + words: list[str] = [] + # A user's-choice first letter (C, D, G, M, N, O) contributes no word; X keeps + # its "Unclassified" word but is just as much the user's own meaning. + user_defined = parts["first"] == "X" + first_word = FIRST_LETTERS[parts["first"]] + if first_word is None: + user_defined = True + else: + words.append(first_word) + if parts["modifier"]: + words.append(MODIFIERS[parts["modifier"]]) + readout, output = parts["readout"], parts["output"] + for index, letter in enumerate(readout): + forms = READOUT[letter] + if forms is None: + user_defined = True + continue + last_function = index == len(readout) - 1 and not output + words.append(forms[1] if last_function else forms[0]) + for index, letter in enumerate(output): + word = OUTPUT[letter] + if word is None: + user_defined = True + continue + if letter == "C" and index + 1 < len(output) and output[index + 1] == "V": + word = "Control" + words.append(word) + if parts["trailing"]: + words.append(TRAILING[parts["trailing"]]) + return " ".join(words), user_defined + + +def _parse_instrument(raw: str) -> dict: + out = { + "tag": raw, + "valid": False, + "kind": "instrument", + "area": None, + "letters": None, + "first": None, + "modifier": None, + "readout": [], + "output": [], + "trailing": None, + "loop": None, + "suffix": None, + "description": None, + "user_defined": False, + "errors": [], + } + tag = _normalise(raw) + match = _INSTRUMENT_RE.match(tag) + if not match: + out["errors"] = [f"{raw!r}: not in [area-]LETTERS[-]loop[suffix] format"] + return out + out.update( + area=match.group("area"), + letters=match.group("letters"), + loop=match.group("loop"), + suffix=match.group("suffix"), + ) + parts = _parse_letters(match.group("letters")) + if parts["errors"]: + out["errors"] = parts["errors"] + return out + description, user_defined = _describe(parts) + out.update( + valid=True, + first={"letter": parts["first"], "meaning": FIRST_LETTERS[parts["first"]]}, + modifier=( + {"letter": parts["modifier"], "meaning": MODIFIERS[parts["modifier"]]} + if parts["modifier"] + else None + ), + readout=[ + {"letter": letter, "meaning": (READOUT[letter] or (None, None))[1]} + for letter in parts["readout"] + ], + output=[{"letter": letter, "meaning": OUTPUT[letter]} for letter in parts["output"]], + trailing=parts["trailing"], + description=description, + user_defined=user_defined, + ) + return out + + +def _parse_equipment(raw: str, prefixes: dict[str, str]) -> dict: + out = { + "tag": raw, + "valid": False, + "kind": "equipment", + "area": None, + "prefix": None, + "equipment_kind": None, + "number": None, + "suffix": "", + "errors": [], + } + match = _EQUIPMENT_RE.match(_normalise(raw)) + if not match: + out["errors"] = [f"{raw!r}: not in [area-]PREFIX-number[suffix] format"] + return out + prefix = match.group("prefix") + out.update( + area=match.group("area"), + prefix=prefix, + number=match.group("number"), + suffix=match.group("suffix") or "", + ) + kind = prefixes.get(prefix) + if kind is None: + out["errors"] = [f"{raw!r}: prefix {prefix} is not in the equipment prefix map"] + return out + out.update(valid=True, equipment_kind=kind) + return out + + +def parse_tag(tag: str, kind: str = "auto", equipment_prefixes: dict | None = None) -> dict: + """Parse an ISA-5.1 instrument tag or an equipment tag. + + ``kind``: ``"instrument"``, ``"equipment"`` or ``"auto"`` (instrument first, + then equipment, else the instrument parse with its errors). + """ + prefixes = {k.upper(): v for k, v in (equipment_prefixes or DEFAULT_EQUIPMENT_PREFIXES).items()} + if kind == "instrument": + return _parse_instrument(tag) + if kind == "equipment": + return _parse_equipment(tag, prefixes) + if kind != "auto": + raise ValueError("kind must be 'auto', 'instrument' or 'equipment'") + instrument = _parse_instrument(tag) + if instrument["valid"]: + return instrument + equipment = _parse_equipment(tag, prefixes) + return equipment if equipment["valid"] else instrument + + +def split_instrument_tag(tag: str) -> tuple[str, str]: + """``FUNC`` and ``LOOP`` attribute texts for a bubble: letters, and loop+suffix. + Anything unparseable goes wholesale into ``FUNC`` so nothing is dropped.""" + match = _INSTRUMENT_RE.match(_normalise(tag)) + if not match: + return _normalise(tag), "" + return match.group("letters"), match.group("loop") + (match.group("suffix") or "") + + +def describe_tables() -> dict: + """The letter tables, for the ``autocad://standards/isa51`` resource.""" + return { + "standard": "ISA-5.1-2009 Table 4.1", + "grammar": "first modifier? readout* output* trailing?", + "first_letters": {k: v or "user's choice" for k, v in FIRST_LETTERS.items()}, + "modifiers": dict(MODIFIERS), + "readout": {k: (v[1] if v else "user's choice") for k, v in READOUT.items()}, + "output": {k: v or "user's choice" for k, v in OUTPUT.items()}, + "trailing": dict(TRAILING), + "equipment_prefixes": dict(DEFAULT_EQUIPMENT_PREFIXES), + "rules": [ + "S in second position is the safety modifier only before V or E (PSV, TSE); otherwise a switch.", + "D in second position is always differential.", + "F in second position is ratio only when the first letter is F (FFC).", + "K in second position is rate of change only before an output letter.", + "H/L/M after an alarm (A) or switch (S) are trailing high/low/middle; L elsewhere is a light.", + "O/C as the last letter after a switch on a position (Z) tag are open/closed (ZSO, ZSC).", + "V (valve) is a final element: no output letter may follow it.", + "A first letter in C/D/G/M/N/O/X or a function letter in B/N/X marks the tag user_defined.", + ], + } diff --git a/tests/data/isa51_tags.json b/tests/data/isa51_tags.json new file mode 100644 index 0000000..843dc8a --- /dev/null +++ b/tests/data/isa51_tags.json @@ -0,0 +1,85 @@ +{ + "valid": [ + ["FIC-101", "Flow Indicating Controller", "101"], + ["FT-101", "Flow Transmitter", "101"], + ["FE-101", "Flow Element", "101"], + ["FQI-102", "Flow Totalizing Indicator", "102"], + ["FFC-103", "Flow Ratio Controller", "103"], + ["FY-104", "Flow Relay", "104"], + ["FCV-105", "Flow Control Valve", "105"], + ["FIT-106", "Flow Indicating Transmitter", "106"], + ["PT-201", "Pressure Transmitter", "201"], + ["PI-201", "Pressure Indicator", "201"], + ["PG-202", "Pressure Gauge", "202"], + ["PDT-203", "Pressure Differential Transmitter", "203"], + ["PDI-203", "Pressure Differential Indicator", "203"], + ["PDIC-204", "Pressure Differential Indicating Controller", "204"], + ["PSV-205", "Pressure Safety Valve", "205"], + ["PSL-206", "Pressure Switch Low", "206"], + ["PSHH-207", "Pressure Switch High-high", "207"], + ["PCV-208", "Pressure Control Valve", "208"], + ["PIC-209", "Pressure Indicating Controller", "209"], + ["LT-301", "Level Transmitter", "301"], + ["LG-302", "Level Gauge", "302"], + ["LAHH-303", "Level Alarm High-high", "303"], + ["LAL-304", "Level Alarm Low", "304"], + ["LSLL-305", "Level Switch Low-low", "305"], + ["LIC-306", "Level Indicating Controller", "306"], + ["LCV-307", "Level Control Valve", "307"], + ["LS-308", "Level Switch", "308"], + ["TE-401", "Temperature Element", "401"], + ["TW-402", "Temperature Well", "402"], + ["TIC-403", "Temperature Indicating Controller", "403"], + ["TSE-404", "Temperature Safety Element", "404"], + ["TAH-405", "Temperature Alarm High", "405"], + ["TT-406", "Temperature Transmitter", "406"], + ["TCV-407", "Temperature Control Valve", "407"], + ["AIT-501", "Analysis Indicating Transmitter", "501"], + ["HV-601", "Hand Valve", "601"], + ["HS-602", "Hand Switch", "602"], + ["HIC-603", "Hand Indicating Controller", "603"], + ["ZSO-701", "Position Switch Open", "701"], + ["ZSC-702", "Position Switch Closed", "702"], + ["ZT-703", "Position Transmitter", "703"], + ["SIC-801", "Speed Indicating Controller", "801"], + ["WIT-802", "Weight Indicating Transmitter", "802"], + ["VT-803", "Vibration Transmitter", "803"], + ["JI-804", "Power Indicator", "804"], + ["EI-805", "Voltage Indicator", "805"], + ["II-806", "Current Indicator", "806"], + ["BE-807", "Burner Element", "807"], + ["RE-808", "Radiation Element", "808"], + ["UY-809", "Multivariable Relay", "809"], + ["PL-810", "Pressure Light", "810"] + ], + "user_defined": [ + ["XV-901", "Unclassified Valve", "901"], + ["NV-902", "Valve", "902"], + ["CIC-903", "Indicating Controller", "903"] + ], + "invalid": [ + ["FCI-101", "I"], + ["FIH-102", "H"], + ["F-103", "function"], + ["FICC-104", "repeated"], + ["PSVX-105", "X"], + ["ABCDEF-106", "D"], + ["FQQ-107", "Q"], + ["1FC", "format"], + ["FIC", "format"], + ["fic-101-", "format"], + ["LSO-108", "O"] + ], + "formats": [ + ["10-FIC-101A", "10", "FIC", "101", "A"], + ["FIC 101", null, "FIC", "101", null], + ["FIC101", null, "FIC", "101", null], + ["fic-101b", null, "FIC", "101", "B"] + ], + "equipment": [ + ["P-101A", "pump", "101", "A"], + ["V-201", "vessel", "201", ""], + ["E-301", "exchanger", "301", ""], + ["10-K-401", "compressor", "401", ""] + ] +} diff --git a/tests/test_pid_tags.py b/tests/test_pid_tags.py new file mode 100644 index 0000000..1d80954 --- /dev/null +++ b/tests/test_pid_tags.py @@ -0,0 +1,87 @@ +"""ISA-5.1 identification-letter grammar, pinned against a reader-checkable fixture.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from engineering.pid.tags import ( + DEFAULT_EQUIPMENT_PREFIXES, + describe_tables, + parse_tag, + split_instrument_tag, +) + +FIXTURE = json.loads((Path(__file__).parent / "data" / "isa51_tags.json").read_text("utf-8")) + + +@pytest.mark.parametrize("tag, description, loop", FIXTURE["valid"]) +def test_valid_tags_parse_with_the_standard_description(tag, description, loop): + out = parse_tag(tag) + assert out["valid"] is True, out["errors"] + assert out["kind"] == "instrument" + assert out["description"] == description + assert out["loop"] == loop + assert out["user_defined"] is False + + +@pytest.mark.parametrize("tag, description, loop", FIXTURE["user_defined"]) +def test_user_choice_letters_parse_as_user_defined(tag, description, loop): + out = parse_tag(tag) + assert out["valid"] is True and out["user_defined"] is True + assert out["description"] == description + + +@pytest.mark.parametrize("tag, fragment", FIXTURE["invalid"]) +def test_invalid_tags_name_the_offending_letter(tag, fragment): + out = parse_tag(tag, kind="instrument") + assert out["valid"] is False + assert out["tag"] == tag + assert any(fragment in err for err in out["errors"]), out["errors"] + + +@pytest.mark.parametrize("tag, area, letters, loop, suffix", FIXTURE["formats"]) +def test_tag_formats(tag, area, letters, loop, suffix): + out = parse_tag(tag) + assert (out["area"], out["letters"], out["loop"], out["suffix"]) == ( + area, + letters, + loop, + suffix, + ) + + +@pytest.mark.parametrize("tag, kind, number, suffix", FIXTURE["equipment"]) +def test_equipment_tags(tag, kind, number, suffix): + out = parse_tag(tag) + assert out["kind"] == "equipment" and out["valid"] is True + assert out["equipment_kind"] == kind and out["number"] == number and out["suffix"] == suffix + + +def test_equipment_prefixes_can_be_replaced_per_call(): + out = parse_tag("Q-1", kind="equipment", equipment_prefixes={"Q": "quench tower"}) + assert out["valid"] is True and out["equipment_kind"] == "quench tower" + default = parse_tag("Q-1", kind="equipment") + assert default["valid"] is False and "prefix" in default["errors"][0] + assert DEFAULT_EQUIPMENT_PREFIXES["P"] == "pump" + + +def test_auto_kind_prefers_instrument_then_equipment_then_reports_instrument_errors(): + assert parse_tag("PT-1")["kind"] == "instrument" + assert parse_tag("P-1")["kind"] == "equipment" + bad = parse_tag("FCI-1") + assert bad["kind"] == "instrument" and bad["valid"] is False + + +def test_split_instrument_tag_feeds_func_and_loop_attributes(): + assert split_instrument_tag("10-FIC-101A") == ("FIC", "101A") + assert split_instrument_tag("PT 7") == ("PT", "7") + assert split_instrument_tag("garbage") == ("GARBAGE", "") + + +def test_describe_tables_is_json_serialisable_and_names_every_first_letter(): + tables = describe_tables() + json.dumps(tables) + assert set(tables["first_letters"]) == set("ABCDEFGHIJKLMNOPQRSTUVWXYZ")