added runtime enum class, metaclass, and tests

src/ophyd_async/core/__init__.py

@@ -50,7 +50,7 @@
     soft_signal_rw,
     wait_for_value,
 )
-from .signal_backend import SignalBackend
+from .signal_backend import RuntimeSubsetEnum, SignalBackend, SubsetEnum
 from .soft_signal_backend import SoftSignalBackend
 from .standard_readable import ConfigSignal, HintedSignal, StandardReadable
 from .utils import (
@@ -68,66 +68,69 @@
 )
 
 __all__ = [
-    "get_mock_put",
-    "callback_on_mock_put",
-    "mock_puts_blocked",
-    "set_mock_values",
-    "reset_mock_put_calls",
-    "SignalBackend",
-    "SoftSignalBackend",
+    "AsyncStatus",
+    "CalculatableTimeout",
+    "CalculateTimeout",
+    "Callback",
+    "ConfigSignal",
+    "DEFAULT_TIMEOUT",
     "DetectorControl",
-    "MockSignalBackend",
     "DetectorTrigger",
     "DetectorWriter",
-    "StandardDetector",
     "Device",
     "DeviceCollector",
     "DeviceVector",
-    "Signal",
-    "SignalR",
-    "SignalW",
-    "SignalRW",
-    "SignalX",
-    "soft_signal_r_and_setter",
-    "soft_signal_rw",
-    "observe_value",
-    "set_and_wait_for_value",
-    "set_mock_put_proceeds",
-    "set_mock_value",
-    "wait_for_value",
-    "AsyncStatus",
-    "WatchableAsyncStatus",
     "DirectoryInfo",
     "DirectoryProvider",
+    "HardwareTriggeredFlyable",
+    "HintedSignal",
+    "MockSignalBackend",
     "NameProvider",
+    "NotConnected",
+    "ReadingValueCallback",
+    "RuntimeSubsetEnum",
+    "SubsetEnum",
     "ShapeProvider",
-    "StaticDirectoryProvider",
+    "Signal",
+    "SignalBackend",
+    "SignalR",
+    "SignalRW",
+    "SignalW",
+    "SignalX",
+    "SoftSignalBackend",
+    "StandardDetector",
     "StandardReadable",
-    "ConfigSignal",
-    "HintedSignal",
+    "StaticDirectoryProvider",
+    "T",
     "TriggerInfo",
     "TriggerLogic",
-    "HardwareTriggeredFlyable",
-    "CalculateTimeout",
-    "CalculatableTimeout",
-    "DEFAULT_TIMEOUT",
-    "Callback",
-    "NotConnected",
-    "ReadingValueCallback",
-    "T",
+    "WatchableAsyncStatus",
+    "assert_configuration",
+    "assert_emitted",
+    "assert_mock_put_called_with",
+    "assert_reading",
+    "assert_value",
+    "callback_on_mock_put",
     "get_dtype",
-    "get_unique",
-    "merge_gathered_dicts",
-    "wait_for_connection",
+    "get_mock_put",
     "get_signal_values",
+    "get_unique",
+    "load_device",
     "load_from_yaml",
+    "merge_gathered_dicts",
+    "mock_puts_blocked",
+    "observe_value",
+    "reset_mock_put_calls",
+    "save_device",
     "save_to_yaml",
+    "set_and_wait_for_value",
+    "set_mock_put_proceeds",
+    "set_mock_value",
+    "set_mock_values",
     "set_signal_values",
+    "soft_signal_r_and_setter",
+    "soft_signal_rw",
+    "wait_for_connection",
+    "wait_for_value",
     "walk_rw_signals",
-    "load_device",
-    "save_device",
-    "assert_reading",
-    "assert_value",
-    "assert_configuration",
-    "assert_emitted",
 ]

src/ophyd_async/core/signal_backend.py

@@ -1,5 +1,13 @@
 from abc import abstractmethod
-from typing import Generic, Optional, Type
+from typing import (
+    TYPE_CHECKING,
+    ClassVar,
+    Generic,
+    Literal,
+    Optional,
+    Tuple,
+    Type,
+)
 
 from bluesky.protocols import DataKey, Reading
 
@@ -45,3 +53,41 @@ async def get_setpoint(self) -> T:
     @abstractmethod
     def set_callback(self, callback: Optional[ReadingValueCallback[T]]) -> None:
         """Observe changes to the current value, timestamp and severity"""
+
+
+class _RuntimeSubsetEnumMeta(type):
+    def __str__(cls):
+        if hasattr(cls, "choices"):
+            return f"SubsetEnum{list(cls.choices)}"
+        return "SubsetEnum"
+
+    def __getitem__(cls, _choices):
+        if isinstance(_choices, str):
+            _choices = (_choices,)
+        else:
+            if not isinstance(_choices, tuple) or not all(
+                isinstance(c, str) for c in _choices
+            ):
+                raise TypeError(
+                    "Choices must be a str or a tuple of str, " f"not {type(_choices)}."
+                )
+            if len(set(_choices)) != len(_choices):
+                raise TypeError("Duplicate elements in runtime enum choices.")
+
+        class _RuntimeSubsetEnum(cls):
+            choices = _choices
+
+        return _RuntimeSubsetEnum
+
+
+class RuntimeSubsetEnum(metaclass=_RuntimeSubsetEnumMeta):
+    choices: ClassVar[Tuple[str, ...]]
+
+    def __init__(self):
+        raise RuntimeError("SubsetEnum cannot be instantiated")
+
+
+if TYPE_CHECKING:
+    SubsetEnum = Literal
+else:
+    SubsetEnum = RuntimeSubsetEnum

src/ophyd_async/core/soft_signal_backend.py

@@ -3,14 +3,23 @@
 import inspect
 import time
 from collections import abc
-from dataclasses import dataclass
 from enum import Enum
-from typing import Dict, Generic, Optional, Type, TypedDict, Union, cast, get_origin
+from typing import (
+    Dict,
+    Generic,
+    Optional,
+    Tuple,
+    Type,
+    TypedDict,
+    Union,
+    cast,
+    get_origin,
+)
 
 import numpy as np
 from bluesky.protocols import DataKey, Dtype, Reading
 
-from .signal_backend import SignalBackend
+from .signal_backend import RuntimeSubsetEnum, SignalBackend
 from .utils import DEFAULT_TIMEOUT, ReadingValueCallback, T, get_dtype
 
 primitive_dtypes: Dict[type, Dtype] = {
@@ -74,37 +83,45 @@ def make_initial_value(self, datatype: Optional[Type[T]]) -> T:
         return cast(T, datatype(shape=0))  # type: ignore
 
 
-@dataclass
 class SoftEnumConverter(SoftConverter):
-    enum_class: Type[Enum]
+    choices: Tuple[str, ...]
 
-    def write_value(self, value: Union[Enum, str]) -> Enum:
+    def __init__(self, datatype: Union[RuntimeSubsetEnum, Enum]):
+        if issubclass(datatype, Enum):
+            self.choices = tuple(v.value for v in datatype)
+        else:
+            self.choices = datatype.choices
+
+    def write_value(self, value: Union[Enum, str]) -> str:
         if isinstance(value, Enum):
+            return value.value
+        else:  # Runtime enum
             return value
-        else:
-            return self.enum_class(value)
 
     def get_datakey(self, source: str, value, **metadata) -> DataKey:
-        choices = [e.value for e in self.enum_class]
         return {
             "source": source,
             "dtype": "string",
             "shape": [],
-            "choices": choices,
+            "choices": self.choices,
             **metadata,
         }
 
     def make_initial_value(self, datatype: Optional[Type[T]]) -> T:
         if datatype is None:
             return cast(T, None)
 
-        return cast(T, list(datatype.__members__.values())[0])  # type: ignore
+        if issubclass(datatype, Enum):
+            return cast(T, list(datatype.__members__.values())[0])  # type: ignore
+        return cast(T, self.choices[0])
 
 
 def make_converter(datatype):
     is_array = get_dtype(datatype) is not None
     is_sequence = get_origin(datatype) == abc.Sequence
-    is_enum = issubclass(datatype, Enum) if inspect.isclass(datatype) else False
+    is_enum = inspect.isclass(datatype) and (
+        issubclass(datatype, Enum) or issubclass(datatype, RuntimeSubsetEnum)
+    )
 
     if is_array or is_sequence:
         return SoftArrayConverter()

src/ophyd_async/epics/_backend/common.py

@@ -1,6 +1,9 @@
+import inspect
 from enum import Enum
 from typing import Dict, Optional, Tuple, Type, TypedDict
 
+from ophyd_async.core.signal_backend import RuntimeSubsetEnum
+
 common_meta = {
     "units",
     "precision",
@@ -30,19 +33,30 @@ def get_supported_values(
     datatype: Optional[Type[str]],
     pv_choices: Tuple[str, ...],
 ) -> Dict[str, str]:
-    if not datatype:
+    if inspect.isclass(datatype) and issubclass(datatype, RuntimeSubsetEnum):
+        if not set(datatype.choices).issubset(set(pv_choices)):
+            raise TypeError(
+                f"{pv} has choices {pv_choices}, "
+                f"which is not a superset of {str(datatype)}."
+            )
         return {x: x or "_" for x in pv_choices}
+    elif inspect.isclass(datatype) and issubclass(datatype, Enum):
+        if not issubclass(datatype, str):
+            raise TypeError(
+                f"{pv} is type Enum but {datatype} does not inherit from String."
+            )
 
-    if not issubclass(datatype, str):
-        raise TypeError(f"{pv} is type Enum but doesn't inherit from String")
-    if issubclass(datatype, Enum):
         choices = tuple(v.value for v in datatype)
         if set(choices) != set(pv_choices):
             raise TypeError(
-                (
-                    f"{pv} has choices {pv_choices}, "
-                    f"which do not match {datatype}, which has {choices}"
-                )
+                f"{pv} has choices {pv_choices}, "
+                f"which do not match {datatype}, which has {choices}."
             )
-        return {x: datatype(x) for x in pv_choices}
-    return {x: x for x in pv_choices}
+        return {x: datatype(x) if x else "_" for x in pv_choices}
+    elif datatype is None:
+        return {x: x or "_" for x in pv_choices}
+
+    raise TypeError(
+        f"{pv} has choices {pv_choices}. "
+        "Use an Enum or SubsetEnum to represent this."
+    )

tests/core/test_soft_signal_backend.py

@@ -31,7 +31,7 @@ def string_d(value):
 
 
 def enum_d(value):
-    return {"dtype": "string", "shape": [], "choices": ["Aaa", "Bbb", "Ccc"]}
+    return {"dtype": "string", "shape": [], "choices": ("Aaa", "Bbb", "Ccc")}
 
 
 def waveform_d(value):