Source code for ophyd_async.core.signal_backend

from abc import abstractmethod
from typing import (
    TYPE_CHECKING,
    ClassVar,
    Generic,
    Literal,
    Optional,
    Tuple,
    Type,
)

from bluesky.protocols import DataKey, Reading

from .utils import DEFAULT_TIMEOUT, ReadingValueCallback, T


[docs] class SignalBackend(Generic[T]): """A read/write/monitor backend for a Signals""" #: Datatype of the signal value datatype: Optional[Type[T]] = None #: Like ca://PV_PREFIX:SIGNAL
[docs] @abstractmethod def source(self, name: str) -> str: """Return source of signal. Signals may pass a name to the backend, which can be used or discarded."""
[docs] @abstractmethod async def connect(self, timeout: float = DEFAULT_TIMEOUT): """Connect to underlying hardware"""
[docs] @abstractmethod async def put(self, value: Optional[T], wait=True, timeout=None): """Put a value to the PV, if wait then wait for completion for up to timeout"""
[docs] @abstractmethod async def get_datakey(self, source: str) -> DataKey: """Metadata like source, dtype, shape, precision, units"""
[docs] @abstractmethod async def get_reading(self) -> Reading: """The current value, timestamp and severity"""
[docs] @abstractmethod async def get_value(self) -> T: """The current value"""
[docs] @abstractmethod async def get_setpoint(self) -> T: """The point that a signal was requested to move to."""
[docs] @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
[docs] 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