Introduce PvaAbstractions and use them in SeqTable

src/ophyd_async/core/__init__.py

@@ -61,7 +61,12 @@
  soft_signal_rw,
  wait_for_value,
 )
-from ._signal_backend import RuntimeSubsetEnum, SignalBackend, SubsetEnum
+from ._signal_backend import (
+ BackendConverterFactory,
+ RuntimeSubsetEnum,
+ SignalBackend,
+ SubsetEnum,
+)
 from ._soft_signal_backend import SignalMetadata, SoftSignalBackend
 from ._status import AsyncStatus, WatchableAsyncStatus
 from ._utils import (
@@ -103,6 +108,7 @@
  "MockSignalBackend",
  "callback_on_mock_put",
  "get_mock_put",
+ "BackendConverterFactory",
  "mock_puts_blocked",
  "reset_mock_put_calls",
  "set_mock_put_proceeds",

src/ophyd_async/core/_device_save_loader.py

@@ -7,6 +7,7 @@
 from bluesky.plan_stubs import abs_set, wait
 from bluesky.protocols import Location
 from bluesky.utils import Msg
+from pydantic import BaseModel
 
 from ._device import Device
 from ._signal import SignalRW
@@ -18,6 +19,12 @@ def ndarray_representer(dumper: yaml.Dumper, array: npt.NDArray[Any]) -> yaml.No
  )
 
 
+def pydantic_model_abstraction_representer(
+ dumper: yaml.Dumper, model: BaseModel
+) -> yaml.Node:
+ return dumper.represent_data(model.model_dump(mode="python"))
+
+
 class OphydDumper(yaml.Dumper):
  def represent_data(self, data: Any) -> Any:
  if isinstance(data, Enum):
@@ -134,6 +141,11 @@ def save_to_yaml(phases: Sequence[Dict[str, Any]], save_path: str) -> None:
  """
 
  yaml.add_representer(np.ndarray, ndarray_representer, Dumper=yaml.Dumper)
+ yaml.add_multi_representer(
+ BaseModel,
+ pydantic_model_abstraction_representer,
+ Dumper=yaml.Dumper,
+ )
 
  with open(save_path, "w") as file:
  yaml.dump(phases, file, Dumper=OphydDumper, default_flow_style=False)

src/ophyd_async/core/_signal_backend.py

@@ -1,10 +1,34 @@
-from abc import abstractmethod
-from typing import TYPE_CHECKING, ClassVar, Generic, Literal, Optional, Tuple, Type
+from abc import ABC, abstractmethod
+from typing import (
+ TYPE_CHECKING,
+ ClassVar,
+ Generic,
+ Literal,
+ Optional,
+ Tuple,
+ Type,
+)
 
 from ._protocol import DataKey, Reading
 from ._utils import DEFAULT_TIMEOUT, ReadingValueCallback, T
 
 
+class BackendConverterFactory(ABC):
+ """Convert between the signal backend and the signal type"""
+
+ _ALLOWED_TYPES: ClassVar[Tuple[Type]]
+
+ @classmethod
+ @abstractmethod
+ def datatype_allowed(cls, datatype: Type) -> bool:
+ """Check if the datatype is allowed."""
+
+ @classmethod
+ @abstractmethod
+ def make_converter(self, datatype: Type):
+ """Updates the object with callables `to_signal` and `from_signal`."""
+
+
 class SignalBackend(Generic[T]):
  """A read/write/monitor backend for a Signals"""
 

src/ophyd_async/core/_soft_signal_backend.py

@@ -4,13 +4,18 @@
 import time
 from collections import abc
 from enum import Enum
-from typing import Dict, Generic, Optional, Tuple, Type, Union, cast, get_origin
+from typing import Any, Dict, Generic, Optional, Tuple, Type, Union, cast, get_origin
 
 import numpy as np
 from bluesky.protocols import DataKey, Dtype, Reading
+from pydantic import BaseModel
 from typing_extensions import TypedDict
 
-from ._signal_backend import RuntimeSubsetEnum, SignalBackend
+from ._signal_backend import (
+ BackendConverterFactory,
+ RuntimeSubsetEnum,
+ SignalBackend,
+)
 from ._utils import DEFAULT_TIMEOUT, ReadingValueCallback, T, get_dtype
 
 primitive_dtypes: Dict[type, Dtype] = {
@@ -94,7 +99,7 @@ def make_initial_value(self, datatype: Optional[Type[T]]) -> T:
 class SoftEnumConverter(SoftConverter):
  choices: Tuple[str, ...]
 
- def __init__(self, datatype: Union[RuntimeSubsetEnum, Enum]):
+ def __init__(self, datatype: Union[RuntimeSubsetEnum, Type[Enum]]):
  if issubclass(datatype, Enum):
  self.choices = tuple(v.value for v in datatype)
  else:
@@ -122,19 +127,54 @@ def make_initial_value(self, datatype: Optional[Type[T]]) -> T:
  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 = inspect.isclass(datatype) and (
- issubclass(datatype, Enum) or issubclass(datatype, RuntimeSubsetEnum)
- )
+class SoftPydanticModelConverter(SoftConverter):
+ def __init__(self, datatype: Type[BaseModel]):
+ self.datatype = datatype
+
+ def reading(self, value: T, timestamp: float, severity: int) -> Reading:
+ value = self.value(value)
+ return super().reading(value, timestamp, severity)
+
+ def value(self, value: Any) -> Any:
+ if isinstance(value, dict):
+ value = self.datatype(**value)
+ return value
+
+ def write_value(self, value):
+ if isinstance(value, self.datatype):
+ return value.model_dump(mode="python")
+ return value
+
+ def make_initial_value(self, datatype: Type | None) -> Any:
+ return super().make_initial_value(datatype)
 
- if is_array or is_sequence:
- return SoftArrayConverter()
- if is_enum:
- return SoftEnumConverter(datatype)
 
- return SoftConverter()
+class SoftSignalConverterFactory(BackendConverterFactory):
+ _ALLOWED_TYPES = (object,) # Any type is allowed
+
+ @classmethod
+ def datatype_allowed(cls, datatype: Type) -> bool:
+ return True # Any value allowed in a soft signal
+
+ @classmethod
+ def make_converter(cls, datatype):
+ is_array = get_dtype(datatype) is not None
+ is_sequence = get_origin(datatype) == abc.Sequence
+ is_enum = inspect.isclass(datatype) and (
+ issubclass(datatype, Enum) or issubclass(datatype, RuntimeSubsetEnum)
+ )
+ is_pydantic_model = inspect.isclass(datatype) and issubclass(
+ datatype, BaseModel
+ )
+
+ if is_array or is_sequence:
+ return SoftArrayConverter()
+ if is_enum:
+ return SoftEnumConverter(datatype)
+ if is_pydantic_model:
+ return SoftPydanticModelConverter(datatype)
+
+ return SoftConverter()
 
 
 class SoftSignalBackend(SignalBackend[T]):
@@ -154,7 +194,9 @@ def __init__(
  self.datatype = datatype
  self._initial_value = initial_value
  self._metadata = metadata or {}
- self.converter: SoftConverter = make_converter(datatype)
+ self.converter: SoftConverter = SoftSignalConverterFactory.make_converter(
+ datatype
+ )
  if self._initial_value is None:
  self._initial_value = self.converter.make_initial_value(self.datatype)
  else:

src/ophyd_async/core/_utils.py

@@ -145,7 +145,7 @@ def get_dtype(typ: Type) -> Optional[np.dtype]:
 
 
 def get_unique(values: Dict[str, T], types: str) -> T:
- """If all values are the same, return that value, otherwise return TypeError
+ """If all values are the same, return that value, otherwise raise TypeError
 
  >>> get_unique({"a": 1, "b": 1}, "integers")
  1

src/ophyd_async/epics/signal/__init__.py

@@ -1,5 +1,6 @@
 from ._common import LimitPair, Limits, get_supported_values
 from ._p4p import PvaSignalBackend
+from ._p4p_table_model import PvaTable
 from ._signal import (
  epics_signal_r,
  epics_signal_rw,
@@ -13,6 +14,7 @@
  "LimitPair",
  "Limits",
  "PvaSignalBackend",
+ "PvaTable",
  "epics_signal_r",
  "epics_signal_rw",
  "epics_signal_rw_rbv",