created the PvaTableAbstraction and used it in the SeqTable

Need to add json serialization.

src/ophyd_async/core/__init__.py

@@ -61,7 +61,13 @@
     soft_signal_rw,
     wait_for_value,
 )
-from ._signal_backend import RuntimeSubsetEnum, SignalBackend, SubsetEnum
+from ._signal_backend import (
+    BackendConverterFactory,
+    ProtocolDatatypeAbstraction,
+    RuntimeSubsetEnum,
+    SignalBackend,
+    SubsetEnum,
+)
 from ._soft_signal_backend import SignalMetadata, SoftSignalBackend
 from ._status import AsyncStatus, WatchableAsyncStatus
 from ._utils import (
@@ -103,6 +109,7 @@
     "MockSignalBackend",
     "callback_on_mock_put",
     "get_mock_put",
+    "BackendConverterFactory",
     "mock_puts_blocked",
     "reset_mock_put_calls",
     "set_mock_put_proceeds",
@@ -117,6 +124,7 @@
     "NameProvider",
     "PathInfo",
     "PathProvider",
+    "ProtocolDatatypeAbstraction",
     "ShapeProvider",
     "StaticFilenameProvider",
     "StaticPathProvider",

src/ophyd_async/core/_signal_backend.py

@@ -1,10 +1,55 @@
-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 ProtocolDatatypeAbstraction(ABC, Generic[T]):
+    @abstractmethod
+    def __init__(self):
+        """The abstract datatype must be able to be intialized with no arguments."""
+
+    @abstractmethod
+    def convert_to_protocol_datatype(self) -> T:
+        """
+        Convert the abstract datatype to a form which can be sent
+        over whichever protocol.
+        """
+
+    @classmethod
+    @abstractmethod
+    def convert_from_protocol_datatype(cls, value: T) -> "ProtocolDatatypeAbstraction":
+        """
+        Convert the datatype received from the protocol to a
+        higher level abstract datatype.
+        """
+
+
+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 typing_extensions import TypedDict
 
-from ._signal_backend import RuntimeSubsetEnum, SignalBackend
+from ._signal_backend import (
+    BackendConverterFactory,
+    ProtocolDatatypeAbstraction,
+    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,55 @@ 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 SoftProtocolDatatypeAbstractionConverter(SoftConverter):
+    """
+    No conversion is necessary for ProtocolDatatypeAbstraction datatypes in soft
+    signals.
+    """
+    def __init__(self, datatype: Type[ProtocolDatatypeAbstraction]):
+        self.datatype = datatype
+
+    def reading(self, value: T, timestamp: float, severity: int) -> Reading:
+        return super().reading(value, timestamp, severity)
+
+    def value(self, value: Any) -> Any:
+        return value
+
+    def write_value(self, value):
+        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_convertable_abstract_datatype = inspect.isclass(datatype) and issubclass(
+            datatype,
+            ProtocolDatatypeAbstraction
+        )
+
+        if is_array or is_sequence:
+            return SoftArrayConverter()
+        if is_enum:
+            return SoftEnumConverter(datatype)
+        if is_convertable_abstract_datatype:
+            return SoftProtocolDatatypeAbstractionConverter(datatype)
+
+        return SoftConverter()
 
 
 class SoftSignalBackend(SignalBackend[T]):
@@ -154,7 +195,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/epics/signal/__init__.py

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

src/ophyd_async/epics/signal/_aioca.py

@@ -1,9 +1,10 @@
+import inspect
 import logging
 import sys
 from dataclasses import dataclass
 from enum import Enum
 from math import isnan, nan
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, Dict, List, Optional, Sequence, Type, Union, get_origin
 
 import numpy as np
 from aioca import (
@@ -22,8 +23,10 @@
 
 from ophyd_async.core import (
     DEFAULT_TIMEOUT,
+    BackendConverterFactory,
     NotConnected,
     ReadingValueCallback,
+    RuntimeSubsetEnum,
     SignalBackend,
     T,
     get_dtype,
@@ -183,57 +186,83 @@ def __getattribute__(self, __name: str) -> Any:
         raise NotImplementedError("No PV has been set as connect() has not been called")
 
 
-def make_converter(
-    datatype: Optional[Type], values: Dict[str, AugmentedValue]
-) -> CaConverter:
-    pv = list(values)[0]
-    pv_dbr = get_unique({k: v.datatype for k, v in values.items()}, "datatypes")
-    is_array = bool([v for v in values.values() if v.element_count > 1])
-    if is_array and datatype is str and pv_dbr == dbr.DBR_CHAR:
-        # Override waveform of chars to be treated as string
-        return CaLongStrConverter()
-    elif is_array and pv_dbr == dbr.DBR_STRING:
-        # Waveform of strings, check we wanted this
-        if datatype:
-            datatype_dtype = get_dtype(datatype)
-            if not datatype_dtype or not np.can_cast(datatype_dtype, np.str_):
-                raise TypeError(f"{pv} has type [str] not {datatype.__name__}")
-        return CaArrayConverter(pv_dbr, None)
-    elif is_array:
-        pv_dtype = get_unique({k: v.dtype for k, v in values.items()}, "dtypes")
-        # This is an array
-        if datatype:
-            # Check we wanted an array of this type
-            dtype = get_dtype(datatype)
-            if not dtype:
-                raise TypeError(f"{pv} has type [{pv_dtype}] not {datatype.__name__}")
-            if dtype != pv_dtype:
-                raise TypeError(f"{pv} has type [{pv_dtype}] not [{dtype}]")
-        return CaArrayConverter(pv_dbr, None)
-    elif pv_dbr == dbr.DBR_ENUM and datatype is bool:
-        # Database can't do bools, so are often representated as enums, CA can do int
-        pv_choices_len = get_unique(
-            {k: len(v.enums) for k, v in values.items()}, "number of choices"
-        )
-        if pv_choices_len != 2:
-            raise TypeError(f"{pv} has {pv_choices_len} choices, can't map to bool")
-        return CaBoolConverter(dbr.DBR_SHORT, dbr.DBR_SHORT)
-    elif pv_dbr == dbr.DBR_ENUM:
-        # This is an Enum
-        pv_choices = get_unique(
-            {k: tuple(v.enums) for k, v in values.items()}, "choices"
+class CaConverterFactory(BackendConverterFactory):
+    _ALLOWED_TYPES = (
+        bool,
+        int,
+        float,
+        str,
+        Sequence,
+        Enum,
+        RuntimeSubsetEnum,
+        np.ndarray
+    )
+
+    @classmethod
+    def datatype_allowed(cls, datatype: Type) -> bool:
+        stripped_origin = get_origin(datatype) or datatype
+        return inspect.isclass(stripped_origin) and issubclass(
+                stripped_origin, cls._ALLOWED_TYPES
         )
-        supported_values = get_supported_values(pv, datatype, pv_choices)
-        return CaEnumConverter(dbr.DBR_STRING, None, supported_values)
-    else:
-        value = list(values.values())[0]
-        # Done the dbr check, so enough to check one of the values
-        if datatype and not isinstance(value, datatype):
-            raise TypeError(
-                f"{pv} has type {type(value).__name__.replace('ca_', '')} "
-                + f"not {datatype.__name__}"
+
+    @classmethod
+    def make_converter(
+        cls, datatype: Optional[Type], values: Dict[str, AugmentedValue]
+    ) -> CaConverter:
+        if datatype is not None and not cls.datatype_allowed(datatype):
+            raise TypeError(f"Given datatype {datatype} unsupported in CA.")
+
+        pv = list(values)[0]
+        pv_dbr = get_unique({k: v.datatype for k, v in values.items()}, "datatypes")
+        is_array = bool([v for v in values.values() if v.element_count > 1])
+        if is_array and datatype is str and pv_dbr == dbr.DBR_CHAR:
+            # Override waveform of chars to be treated as string
+            return CaLongStrConverter()
+        elif is_array and pv_dbr == dbr.DBR_STRING:
+            # Waveform of strings, check we wanted this
+            if datatype:
+                datatype_dtype = get_dtype(datatype)
+                if not datatype_dtype or not np.can_cast(datatype_dtype, np.str_):
+                    raise TypeError(f"{pv} has type [str] not {datatype.__name__}")
+            return CaArrayConverter(pv_dbr, None)
+        elif is_array:
+            pv_dtype = get_unique({k: v.dtype for k, v in values.items()}, "dtypes")
+            # This is an array
+            if datatype:
+                # Check we wanted an array of this type
+                dtype = get_dtype(datatype)
+                if not dtype:
+                    raise TypeError(
+                        f"{pv} has type [{pv_dtype}] not {datatype.__name__}"
+                    )
+                if dtype != pv_dtype:
+                    raise TypeError(f"{pv} has type [{pv_dtype}] not [{dtype}]")
+            return CaArrayConverter(pv_dbr, None)
+        elif pv_dbr == dbr.DBR_ENUM and datatype is bool:
+            # Database can't do bools, so are often representated as enums,
+            # CA can do int
+            pv_choices_len = get_unique(
+                {k: len(v.enums) for k, v in values.items()}, "number of choices"
             )
-        return CaConverter(pv_dbr, None)
+            if pv_choices_len != 2:
+                raise TypeError(f"{pv} has {pv_choices_len} choices, can't map to bool")
+            return CaBoolConverter(dbr.DBR_SHORT, dbr.DBR_SHORT)
+        elif pv_dbr == dbr.DBR_ENUM:
+            # This is an Enum
+            pv_choices = get_unique(
+                {k: tuple(v.enums) for k, v in values.items()}, "choices"
+            )
+            supported_values = get_supported_values(pv, datatype, pv_choices)
+            return CaEnumConverter(dbr.DBR_STRING, None, supported_values)
+        else:
+            value = list(values.values())[0]
+            # Done the dbr check, so enough to check one of the values
+            if datatype and not isinstance(value, datatype):
+                raise TypeError(
+                    f"{pv} has type {type(value).__name__.replace('ca_', '')} "
+                    + f"not {datatype.__name__}"
+                )
+            return CaConverter(pv_dbr, None)
 
 
 _tried_pyepics = False
@@ -280,7 +309,9 @@ async def connect(self, timeout: float = DEFAULT_TIMEOUT):
         else:
             # The same, so only need to connect one
             await self._store_initial_value(self.read_pv, timeout=timeout)
-        self.converter = make_converter(self.datatype, self.initial_values)
+        self.converter = CaConverterFactory.make_converter(
+            self.datatype, self.initial_values
+        )
 
     async def put(self, value: Optional[T], wait=True, timeout=None):
         if value is None: