Implement Attributes with allowed values as mbb records

Add check for enum attributes and perform conversions in pythonSoftIOC
wrapper functions.

Co-authored-by: Gary Yendell <gary.yendell@diamond.ac.uk>

src/fastcs/backends/epics/ioc.py

@@ -8,8 +8,13 @@
 from softioc.pythonSoftIoc import RecordWrapper
 
 from fastcs.attributes import AttrR, AttrRW, AttrW
+from fastcs.backends.epics.util import (
+    MBB_MAX_CHOICES,
+    MBB_STATE_FIELDS,
+    convert_if_enum,
+)
 from fastcs.controller import BaseController
-from fastcs.datatypes import Bool, DataType, Float, Int, String
+from fastcs.datatypes import Bool, Float, Int, String, T
 from fastcs.exceptions import FastCSException
 from fastcs.mapping import Mapping
 
@@ -126,20 +131,28 @@ def _create_and_link_attribute_pvs(pv_prefix: str, mapping: Mapping) -> None:
 
 
 def _create_and_link_read_pv(
-    pv_prefix: str, pv_name: str, attr_name: str, attribute: AttrR
+    pv_prefix: str, pv_name: str, attr_name: str, attribute: AttrR[T]
 ) -> None:
-    record = _get_input_record(f"{pv_prefix}:{pv_name}", attribute.datatype)
+    record = _get_input_record(f"{pv_prefix}:{pv_name}", attribute)
 
     _add_attr_pvi_info(record, pv_prefix, attr_name, "r")
 
-    async def async_wrapper(v):
-        record.set(v)
+    async def async_record_set(value: T):
+        record.set(convert_if_enum(attribute, value))
 
-    attribute.set_update_callback(async_wrapper)
+    attribute.set_update_callback(async_record_set)
 
 
-def _get_input_record(pv: str, datatype: DataType) -> RecordWrapper:
-    match datatype:
+def _get_input_record(pv: str, attribute: AttrR) -> RecordWrapper:
+    if (
+        isinstance(attribute.datatype, String)
+        and attribute.allowed_values is not None
+        and len(attribute.allowed_values) <= MBB_MAX_CHOICES
+    ):
+        state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))
+        return builder.mbbIn(pv, **state_keys)
+
+    match attribute.datatype:
         case Bool(znam, onam):
             return builder.boolIn(pv, ZNAM=znam, ONAM=onam)
         case Int():
@@ -149,28 +162,51 @@ def _get_input_record(pv: str, datatype: DataType) -> RecordWrapper:
         case String():
             return builder.longStringIn(pv)
         case _:
-            raise FastCSException(f"Unsupported type {type(datatype)}: {datatype}")
+            raise FastCSException(
+                f"Unsupported type {type(attribute.datatype)}: {attribute.datatype}"
+            )
 
 
 def _create_and_link_write_pv(
-    pv_prefix: str, pv_name: str, attr_name: str, attribute: AttrW
+    pv_prefix: str, pv_name: str, attr_name: str, attribute: AttrW[T]
 ) -> None:
+    async def on_update(value):
+        if (
+            isinstance(attribute.datatype, String)
+            and isinstance(value, int)
+            and attribute.allowed_values is not None
+        ):
+            try:
+                value = attribute.allowed_values[value]
+            except IndexError:
+                raise IndexError(
+                    f"Invalid index {value}, allowed values: {attribute.allowed_values}"
+                ) from None
+
+        await attribute.process_without_display_update(value)
+
     record = _get_output_record(
-        f"{pv_prefix}:{pv_name}",
-        attribute.datatype,
-        on_update=attribute.process_without_display_update,
+        f"{pv_prefix}:{pv_name}", attribute, on_update=on_update
     )
 
     _add_attr_pvi_info(record, pv_prefix, attr_name, "w")
 
-    async def async_wrapper(v):
-        record.set(v, process=False)
+    async def async_record_set(value: T):
+        record.set(convert_if_enum(attribute, value), process=False)
+
+    attribute.set_write_display_callback(async_record_set)
 
-    attribute.set_write_display_callback(async_wrapper)
 
+def _get_output_record(pv: str, attribute: AttrW, on_update: Callable) -> Any:
+    if (
+        isinstance(attribute.datatype, String)
+        and attribute.allowed_values is not None
+        and len(attribute.allowed_values) <= MBB_MAX_CHOICES
+    ):
+        state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))
+        return builder.mbbOut(pv, always_update=True, on_update=on_update, **state_keys)
 
-def _get_output_record(pv: str, datatype: DataType, on_update: Callable) -> Any:
-    match datatype:
+    match attribute.datatype:
         case Bool(znam, onam):
             return builder.boolOut(
                 pv,
@@ -186,7 +222,9 @@ def _get_output_record(pv: str, datatype: DataType, on_update: Callable) -> Any:
         case String():
             return builder.longStringOut(pv, always_update=True, on_update=on_update)
         case _:
-            raise FastCSException(f"Unsupported type {type(datatype)}: {datatype}")
+            raise FastCSException(
+                f"Unsupported type {type(attribute.datatype)}: {attribute.datatype}"
+            )
 
 
 def _create_and_link_command_pvs(pv_prefix: str, mapping: Mapping) -> None:

src/fastcs/backends/epics/util.py

@@ -0,0 +1,51 @@
+from fastcs.attributes import Attribute
+from fastcs.datatypes import String, T
+
+_MBB_FIELD_PREFIXES = (
+    "ZR",
+    "ON",
+    "TW",
+    "TH",
+    "FR",
+    "FV",
+    "SX",
+    "SV",
+    "EI",
+    "NI",
+    "TE",
+    "EL",
+    "TV",
+    "TT",
+    "FT",
+    "FF",
+)
+
+MBB_STATE_FIELDS = tuple(f"{p}ST" for p in _MBB_FIELD_PREFIXES)
+MBB_VALUE_FIELDS = tuple(f"{p}VL" for p in _MBB_FIELD_PREFIXES)
+MBB_MAX_CHOICES = len(_MBB_FIELD_PREFIXES)
+
+
+def convert_if_enum(attribute: Attribute[T], value: T) -> T | int:
+    """Check if `attribute` is a string enum and if so convert `value` to index of enum.
+
+    Args:
+        `attribute`: The attribute to be set
+        `value`: The value
+
+    Returns:
+        The index of the `value` if the `attribute` is an enum, else `value`
+
+    Raises:
+        ValueError: If `attribute` is an enum and `value` is not in its allowed values
+
+    """
+    match attribute:
+        case Attribute(
+            datatype=String(), allowed_values=allowed_values
+        ) if allowed_values is not None:
+            if value in allowed_values:
+                return allowed_values.index(value)
+            else:
+                raise ValueError(f"'{value}' not in allowed values {allowed_values}")
+        case _:
+            return value

tests/backends/epics/test_gui.py

@@ -1,3 +1,4 @@
+import pytest
 from pvi.device import (
     LED,
     ButtonPanel,
@@ -29,6 +30,7 @@ def test_get_components(mapping):
 
     components = gui.extract_mapping_components(mapping.get_controller_mappings()[0])
     assert components == [
+        SignalR(name="BigEnum", read_pv="DEVICE:BigEnum", read_widget=TextRead()),
         SignalR(name="ReadBool", read_pv="DEVICE:ReadBool", read_widget=LED()),
         SignalR(
             name="ReadInt",
@@ -68,3 +70,9 @@ def test_get_components(mapping):
             value="1",
         ),
     ]
+
+
+@pytest.mark.skip
+def test_generate_gui(mapping):
+    gui = EpicsGUI(mapping, "DEVICE")
+    gui.create_gui()

tests/backends/epics/test_ioc.py

@@ -25,13 +25,21 @@ def test_ioc(mocker: MockerFixture, mapping: Mapping):
     builder.aOut.assert_any_call(
         "DEVICE:ReadWriteFloat", always_update=True, on_update=mocker.ANY, PREC=2
     )
+    builder.longIn.assert_any_call("DEVICE:BigEnum")
     builder.longIn.assert_any_call("DEVICE:ReadWriteInt_RBV")
     builder.longOut.assert_called_with(
         "DEVICE:ReadWriteInt", always_update=True, on_update=mocker.ANY
     )
-    builder.longStringIn.assert_called_once_with("DEVICE:StringEnum_RBV")
-    builder.longStringOut.assert_called_once_with(
-        "DEVICE:StringEnum", always_update=True, on_update=mocker.ANY
+    builder.mbbIn.assert_called_once_with(
+        "DEVICE:StringEnum_RBV", ZRST="red", ONST="green", TWST="blue"
+    )
+    builder.mbbOut.assert_called_once_with(
+        "DEVICE:StringEnum",
+        ZRST="red",
+        ONST="green",
+        TWST="blue",
+        always_update=True,
+        on_update=mocker.ANY,
     )
     builder.boolOut.assert_called_once_with(
         "DEVICE:WriteBool",

tests/backends/epics/test_ioc_system.py

@@ -21,4 +21,3 @@ def test_ioc(ioc: None):
         "c": {"w": "DEVICE:Child:C"},
         "d": {"x": "DEVICE:Child:D"},
     }
-    pass

tests/backends/epics/test_util.py

@@ -0,0 +1,17 @@
+import pytest
+
+from fastcs.attributes import AttrR
+from fastcs.backends.epics.util import convert_if_enum
+from fastcs.datatypes import String
+
+
+def test_convert_if_enum():
+    string_attr = AttrR(String())
+    enum_attr = AttrR(String(), allowed_values=["disabled", "enabled"])
+
+    assert convert_if_enum(string_attr, "enabled") == "enabled"
+
+    assert convert_if_enum(enum_attr, "enabled") == 1
+
+    with pytest.raises(ValueError):
+        convert_if_enum(enum_attr, "off")

tests/backends/tango/test_dsr.py

@@ -10,6 +10,7 @@ def test_collect_attributes(mapping):
 
     # Check that attributes are created and of expected type
     assert list(attributes.keys()) == [
+        "BigEnum",
         "ReadBool",
         "ReadInt",
         "ReadWriteFloat",

tests/conftest.py

@@ -50,6 +50,10 @@ class TestController(Controller):
     read_bool: AttrR = AttrR(Bool())
     write_bool: AttrW = AttrW(Bool(), handler=TestSender())
     string_enum: AttrRW = AttrRW(String(), allowed_values=["red", "green", "blue"])
+    big_enum: AttrR = AttrR(
+        Int(),
+        allowed_values=[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17],
+    )
 
     initialised = False
     connected = False