Merge pull request #40 from DiamondLightSource/dropdowns

Create mbb records for enum Attributes

Author: GDYendell

src/fastcs/attributes.py

@@ -51,6 +51,7 @@ def __init__(
         access_mode: AttrMode,
         group: str | None = None,
         handler: Any = None,
+        allowed_values: list[T] | None = None,
     ) -> None:
         assert (
             datatype.dtype in ATTRIBUTE_TYPES
@@ -59,6 +60,7 @@ def __init__(
         self._access_mode: AttrMode = access_mode
         self._group = group
         self.enabled = True
+        self._allowed_values: list[T] | None = allowed_values
 
     @property
     def datatype(self) -> DataType[T]:
@@ -76,6 +78,10 @@ def access_mode(self) -> AttrMode:
     def group(self) -> str | None:
         return self._group
 
+    @property
+    def allowed_values(self) -> list[T] | None:
+        return self._allowed_values
+
 
 class AttrR(Attribute[T]):
     """A read-only ``Attribute``."""
@@ -86,8 +92,15 @@ def __init__(
         access_mode=AttrMode.READ,
         group: str | None = None,
         handler: Updater | None = None,
+        allowed_values: list[T] | None = None,
     ) -> None:
-        super().__init__(datatype, access_mode, group, handler)  # type: ignore
+        super().__init__(
+            datatype,  # type: ignore
+            access_mode,
+            group,
+            handler,
+            allowed_values=allowed_values,  # type: ignore
+        )
         self._value: T = datatype.dtype()
         self._update_callback: AttrCallback[T] | None = None
         self._updater = handler
@@ -118,19 +131,19 @@ def __init__(
         access_mode=AttrMode.WRITE,
         group: str | None = None,
         handler: Sender | None = None,
-        allowed_values: list[str] | None = None,
+        allowed_values: list[T] | None = None,
     ) -> None:
-        super().__init__(datatype, access_mode, group, handler)  # type: ignore
+        super().__init__(
+            datatype,  # type: ignore
+            access_mode,
+            group,
+            handler,
+            allowed_values=allowed_values,  # type: ignore
+        )
         self._process_callback: AttrCallback[T] | None = None
         self._write_display_callback: AttrCallback[T] | None = None
         self._sender = handler
 
-        self._allowed_values = allowed_values
-
-    @property
-    def allowed_values(self) -> list[str] | None:
-        return self._allowed_values
-
     async def process(self, value: T) -> None:
         await self.process_without_display_update(value)
         await self.update_display_without_process(value)
@@ -166,9 +179,15 @@ def __init__(
         access_mode=AttrMode.READ_WRITE,
         group: str | None = None,
         handler: Handler | None = None,
-        allowed_values: list[str] | None = None,
+        allowed_values: list[T] | None = None,
     ) -> None:
-        super().__init__(datatype, access_mode, group, handler, allowed_values)  # type: ignore
+        super().__init__(
+            datatype,  # type: ignore
+            access_mode,
+            group,
+            handler,
+            allowed_values,  # type: ignore
+        )
 
     async def process(self, value: T) -> None:
         await self.set(value)

src/fastcs/backends/epics/ioc.py

@@ -8,8 +8,14 @@
 from softioc.pythonSoftIoc import RecordWrapper
 
 from fastcs.attributes import AttrR, AttrRW, AttrW
+from fastcs.backends.epics.util import (
+    MBB_STATE_FIELDS,
+    attr_is_enum,
+    enum_index_to_value,
+    enum_value_to_index,
+)
 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
 
@@ -148,20 +154,30 @@ 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)
+    if attr_is_enum(attribute):
 
+        async def async_record_set(value: T):
+            record.set(enum_value_to_index(attribute, value))
+    else:
+
+        async def async_record_set(value: T):  # type: ignore
+            record.set(value)
+
+    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)
+    attribute.set_update_callback(async_record_set)
 
-    attribute.set_update_callback(async_wrapper)
 
+def _get_input_record(pv: str, attribute: AttrR) -> RecordWrapper:
+    if attr_is_enum(attribute):
+        # https://github.com/python/mypy/issues/16789
+        state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))  # type: ignore
+        return builder.mbbIn(pv, **state_keys)
 
-def _get_input_record(pv: str, datatype: DataType) -> RecordWrapper:
-    match datatype:
+    match attribute.datatype:
         case Bool(znam, onam):
             return builder.boolIn(pv, ZNAM=znam, ONAM=onam)
         case Int():
@@ -171,28 +187,46 @@ 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:
+    if attr_is_enum(attribute):
+
+        async def on_update(value):
+            await attribute.process_without_display_update(
+                enum_index_to_value(attribute, value)
+            )
+
+        async def async_write_display(value: T):
+            record.set(enum_value_to_index(attribute, value), process=False)
+
+    else:
+
+        async def on_update(value):
+            await attribute.process_without_display_update(value)
+
+        async def async_write_display(value: T):  # type: ignore
+            record.set(value, process=False)
+
     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)
+    attribute.set_write_display_callback(async_write_display)
 
-    attribute.set_write_display_callback(async_wrapper)
 
+def _get_output_record(pv: str, attribute: AttrW, on_update: Callable) -> Any:
+    if attr_is_enum(attribute):
+        state_keys = dict(zip(MBB_STATE_FIELDS, attribute.allowed_values, strict=False))  # type: ignore
+        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,
@@ -208,7 +242,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,99 @@
+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 attr_is_enum(attribute: Attribute) -> bool:
+    """Check if the `Attribute` has a `String` datatype and has `allowed_values` set.
+
+    Args:
+        attribute: The `Attribute` to check
+
+    Returns:
+        `True` if `Attribute` is an enum, else `False`
+
+    """
+    match attribute:
+        case Attribute(
+            datatype=String(), allowed_values=allowed_values
+        ) if allowed_values is not None and len(allowed_values) <= MBB_MAX_CHOICES:
+            return True
+        case _:
+            return False
+
+
+def enum_value_to_index(attribute: Attribute[T], value: T) -> int:
+    """Convert the given value to the index within the allowed_values of the Attribute
+
+    Args:
+        `attribute`: The attribute
+        `value`: The value to convert
+
+    Returns:
+        The index of the `value`
+
+    Raises:
+        ValueError: If `attribute` has no allowed values or `value` is not a valid
+            option
+
+    """
+    if attribute.allowed_values is None:
+        raise ValueError(
+            "Cannot convert value to index for Attribute without allowed values"
+        )
+
+    try:
+        return attribute.allowed_values.index(value)
+    except ValueError:
+        raise ValueError(
+            f"{value} not in allowed values of {attribute}: {attribute.allowed_values}"
+        ) from None
+
+
+def enum_index_to_value(attribute: Attribute[T], index: int) -> T:
+    """Lookup the value from the allowed_values of an attribute at the given index.
+
+    Parameters:
+        attribute: The `Attribute` to lookup the index from
+        index: The index of the value to retrieve
+
+    Returns:
+        The value at the specified index in the allowed values list.
+
+    Raises:
+        IndexError: If the index is out of bounds
+
+    """
+    if attribute.allowed_values is None:
+        raise ValueError(
+            "Cannot lookup value by index for Attribute without allowed values"
+        )
+
+    try:
+        return attribute.allowed_values[index]
+    except IndexError:
+        raise IndexError(
+            f"Invalid index {index} into allowed values: {attribute.allowed_values}"
+        ) from None

tests/backends/epics/test_gui.py

@@ -29,6 +29,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",