merge: track-a task 6

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd

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<area>\d{1,4})-)?(?P<letters>[A-Z]{1,5})[- ]?(?P<loop>\d{1,5})(?:[- ]?(?P<suffix>[A-Z]))?$"
+)
+_EQUIPMENT_RE = re.compile(
+    r"^(?:(?P<area>\d{1,4})-)?(?P<prefix>[A-Z]{1,3})-(?P<number>\d{1,5})(?P<suffix>[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.",
+        ],
+    }

tests/data/isa51_tags.json

@@ -0,0 +1,86 @@
+{
+  "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", "unexpected letter I at position 3"],
+    ["FIH-102", "trailing H needs an alarm (A) or switch (S) before it"],
+    ["F-103", "needs at least one function letter after F"],
+    ["FICC-104", "repeated letter C at position 4"],
+    ["PSVX-105", "unexpected letter X at position 4 after valve V"],
+    ["ABCDE-106", "unexpected letter D at position 4"],
+    ["ABCDEF-106", "not in [area-]LETTERS[-]loop[suffix] format"],
+    ["FQQ-107", "unexpected letter Q at position 3"],
+    ["1FC", "not in [area-]LETTERS[-]loop[suffix] format"],
+    ["FIC", "not in [area-]LETTERS[-]loop[suffix] format"],
+    ["fic-101-", "not in [area-]LETTERS[-]loop[suffix] format"],
+    ["LSO-108", "trailing O (open/closed) is only valid after a switch on a position (Z) tag"]
+  ],
+  "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", ""]
+  ]
+}

tests/test_pid_tags.py

@@ -0,0 +1,103 @@
+"""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
+    # The grammar stops at the first offending letter, so the fixture pins
+    # exactly one error and the fragment must sit in *that* one.
+    assert len(out["errors"]) == 1, out["errors"]
+    assert fragment in out["errors"][0], out["errors"]
+
+
+@pytest.mark.parametrize("tag, fragment", FIXTURE["invalid"])
+def test_invalid_fixture_fragments_bind_the_letter_and_position(tag, fragment):
+    # Every grammar error is prefixed with the tag's own letters, so a fragment
+    # that is a substring of the tag ("I" for FCI) matches whatever the grammar
+    # says and pins nothing. A row must quote the rule text, and a grammar row
+    # must quote the reported position as well as the letter.
+    assert fragment not in tag.upper(), f"{tag!r}: fragment {fragment!r} is part of the tag"
+    if "format" not in fragment:
+        assert "position" in fragment or "letter" in fragment or "trailing" in fragment, (
+            f"{tag!r}: fragment {fragment!r} names neither a letter nor a position"
+        )
+
+
+@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")