Skip to content

Commit

Permalink
added runtime enum class, metaclass, and tests (#341)
Browse files Browse the repository at this point in the history
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']`
  • Loading branch information
evalott100 authored Jun 20, 2024
1 parent d55dcb1 commit 9165c5a
Show file tree
Hide file tree
Showing 8 changed files with 266 additions and 76 deletions.
91 changes: 47 additions & 44 deletions src/ophyd_async/core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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",
]
48 changes: 47 additions & 1 deletion src/ophyd_async/core/signal_backend.py
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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
41 changes: 29 additions & 12 deletions src/ophyd_async/core/soft_signal_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -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] = {
Expand Down Expand Up @@ -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()
Expand Down
34 changes: 24 additions & 10 deletions src/ophyd_async/epics/_backend/common.py
Original file line number Diff line number Diff line change
@@ -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",
Expand Down Expand Up @@ -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."
)
2 changes: 1 addition & 1 deletion tests/core/test_soft_signal_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Loading

0 comments on commit 9165c5a

Please sign in to comment.