added runtime enum class, metaclass, and tests (#341)

added `SubsetEnum` class (with metaclass), and added logic to `make_converter`s to support them

Also made it so that
	* if the record is mbb then the signal has to be
	  `Enum`,`SubsetEnum`, or None
	* `SubsetEnum` metaclass caches with a tuple, so
	  order matters and `SubsetEnum['A', 'B'] is not
	  `SubsetEnum['B', 'A']`

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):