Replace Handler with AttributeIO / AttributeIORef

src/fastcs/attribute_io.py

@@ -0,0 +1,24 @@
+from typing import Any, Generic, cast, get_args
+
+from fastcs.attribute_io_ref import AttributeIORef, AttributeIORefT
+from fastcs.attributes import AttrR, AttrRW
+from fastcs.datatypes import T
+
+
+class AttributeIO(Generic[T, AttributeIORefT]):
+    ref_type = AttributeIORef
+
+    def __init_subclass__(cls) -> None:
+        # sets ref_type from subclass generic args
+        # from python 3.12 we can use types.get_original_bases
+        args = get_args(cast(Any, cls).__orig_bases__[0])
+        cls.ref_type = args[1]
+
+    async def update(self, attr: AttrR[T, AttributeIORefT]) -> None:
+        raise NotImplementedError()
+
+    async def send(self, attr: AttrRW[T, AttributeIORefT], value: T) -> None:
+        raise NotImplementedError()
+
+
+AnyAttributeIO = AttributeIO[T, AttributeIORef]

src/fastcs/attribute_io_ref.py

@@ -0,0 +1,13 @@
+from dataclasses import dataclass
+
+from typing_extensions import TypeVar
+
+
+@dataclass(kw_only=True)
+class AttributeIORef:
+    update_period: float | None = None
+
+
+AttributeIORefT = TypeVar(
+    "AttributeIORefT", bound=AttributeIORef, default=AttributeIORef
+)

src/fastcs/attributes.py

@@ -2,67 +2,16 @@
 
 import asyncio
 from collections.abc import Callable
-from enum import Enum
-from typing import Any, Generic
+from typing import Generic
 
-import fastcs
-
-from .datatypes import ATTRIBUTE_TYPES, AttrCallback, DataType, T
+from .attribute_io_ref import AttributeIORefT
+from .datatypes import ATTRIBUTE_TYPES, AttrSetCallback, AttrUpdateCallback, DataType, T
 
 ONCE = float("inf")
 """Special value to indicate that an attribute should be updated once on start up."""
 
 
-class AttrMode(Enum):
-    """Access mode of an ``Attribute``."""
-
-    READ = 1
-    WRITE = 2
-    READ_WRITE = 3
-
-
-class _BaseAttrHandler:
-    async def initialise(self, controller: fastcs.controller.BaseController) -> None:
-        pass
-
-
-class AttrHandlerW(_BaseAttrHandler):
-    """Protocol for setting the value of an ``Attribute``."""
-
-    async def put(self, attr: AttrW[T], value: T) -> None:
-        pass
-
-
-class AttrHandlerR(_BaseAttrHandler):
-    """Protocol for updating the cached readback value of an ``Attribute``."""
-
-    # If update period is None then the attribute will not be updated as a task.
-    update_period: float | None = None
-
-    async def update(self, attr: AttrR[T]) -> None:
-        pass
-
-
-class AttrHandlerRW(AttrHandlerR, AttrHandlerW):
-    """Protocol encapsulating both ``AttrHandlerR`` and ``AttHandlerW``."""
-
-    pass
-
-
-class SimpleAttrHandler(AttrHandlerRW):
-    """Handler for internal parameters"""
-
-    async def put(self, attr: AttrW[T], value: T) -> None:
-        await attr.update_display_without_process(value)
-
-        if isinstance(attr, AttrRW):
-            await attr.set(value)
-
-    async def update(self, attr: AttrR) -> None:
-        raise RuntimeError("SimpleHandler cannot update")
-
-
-class Attribute(Generic[T]):
+class Attribute(Generic[T, AttributeIORefT]):
     """Base FastCS attribute.
 
     Instances of this class added to a ``Controller`` will be used by the backend.
@@ -71,26 +20,33 @@ class Attribute(Generic[T]):
     def __init__(
         self,
         datatype: DataType[T],
-        access_mode: AttrMode,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
-        handler: Any = None,
         description: str | None = None,
     ) -> None:
         assert issubclass(datatype.dtype, ATTRIBUTE_TYPES), (
             f"Attr type must be one of {ATTRIBUTE_TYPES}, "
             "received type {datatype.dtype}"
         )
+        self._io_ref = io_ref
         self._datatype: DataType[T] = datatype
-        self._access_mode: AttrMode = access_mode
         self._group = group
-        self._handler = handler
         self.enabled = True
         self.description = description
 
         # A callback to use when setting the datatype to a different value, for example
         # changing the units on an int. This should be implemented in the backend.
         self._update_datatype_callbacks: list[Callable[[DataType[T]], None]] = []
 
+    @property
+    def io_ref(self) -> AttributeIORefT:
+        if self._io_ref is None:
+            raise RuntimeError(f"{self} has no AttributeIORef")
+        return self._io_ref
+
+    def has_io_ref(self):
+        return self._io_ref is not None
+
     @property
     def datatype(self) -> DataType[T]:
         return self._datatype
@@ -99,18 +55,10 @@ def datatype(self) -> DataType[T]:
     def dtype(self) -> type[T]:
         return self._datatype.dtype
 
-    @property
-    def access_mode(self) -> AttrMode:
-        return self._access_mode
-
     @property
     def group(self) -> str | None:
         return self._group
 
-    async def initialise(self, controller: fastcs.controller.BaseController) -> None:
-        if self._handler is not None:
-            await self._handler.initialise(controller)
-
     def add_update_datatype_callback(
         self, callback: Callable[[DataType[T]], None]
     ) -> None:
@@ -126,71 +74,71 @@ def update_datatype(self, datatype: DataType[T]) -> None:
             callback(datatype)
 
 
-class AttrR(Attribute[T]):
+class AttrR(Attribute[T, AttributeIORefT]):
     """A read-only ``Attribute``."""
 
     def __init__(
         self,
         datatype: DataType[T],
-        access_mode=AttrMode.READ,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
-        handler: AttrHandlerR | None = None,
         initial_value: T | None = None,
         description: str | None = None,
     ) -> None:
         super().__init__(
             datatype,  # type: ignore
-            access_mode,
+            io_ref,
             group,
-            handler,
             description=description,
         )
         self._value: T = (
             datatype.initial_value if initial_value is None else initial_value
         )
-        self._update_callbacks: list[AttrCallback[T]] | None = None
-        self._updater = handler
+        self._on_set_callbacks: list[AttrSetCallback[T]] | None = None
+        self._on_update_callbacks: list[AttrUpdateCallback] | None = None
 
     def get(self) -> T:
         return self._value
 
     async def set(self, value: T) -> None:
         self._value = self._datatype.validate(value)
 
-        if self._update_callbacks is not None:
-            await asyncio.gather(*[cb(self._value) for cb in self._update_callbacks])
+        if self._on_set_callbacks is not None:
+            await asyncio.gather(*[cb(self._value) for cb in self._on_set_callbacks])
 
-    def add_update_callback(self, callback: AttrCallback[T]) -> None:
-        if self._update_callbacks is None:
-            self._update_callbacks = []
-        self._update_callbacks.append(callback)
+    def add_set_callback(self, callback: AttrSetCallback[T]) -> None:
+        if self._on_set_callbacks is None:
+            self._on_set_callbacks = []
+        self._on_set_callbacks.append(callback)
 
-    @property
-    def updater(self) -> AttrHandlerR | None:
-        return self._updater
+    def add_update_callback(self, callback: AttrUpdateCallback):
+        if self._on_update_callbacks is None:
+            self._on_update_callbacks = []
+        self._on_update_callbacks.append(callback)
 
+    async def update(self):
+        if self._on_update_callbacks is not None:
+            await asyncio.gather(*[cb() for cb in self._on_update_callbacks])
 
-class AttrW(Attribute[T]):
+
+class AttrW(Attribute[T, AttributeIORefT]):
     """A write-only ``Attribute``."""
 
     def __init__(
         self,
         datatype: DataType[T],
-        access_mode=AttrMode.WRITE,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
-        handler: AttrHandlerW | None = None,
         description: str | None = None,
     ) -> None:
         super().__init__(
             datatype,  # type: ignore
-            access_mode,
+            io_ref,
             group,
-            handler,
             description=description,
         )
-        self._process_callbacks: list[AttrCallback[T]] | None = None
-        self._write_display_callbacks: list[AttrCallback[T]] | None = None
-        self._setter = handler
+        self._process_callbacks: list[AttrSetCallback[T]] | None = None
+        self._write_display_callbacks: list[AttrSetCallback[T]] | None = None
 
     async def process(self, value: T) -> None:
         await self.process_without_display_update(value)
@@ -206,45 +154,23 @@ async def update_display_without_process(self, value: T) -> None:
         if self._write_display_callbacks:
             await asyncio.gather(*[cb(value) for cb in self._write_display_callbacks])
 
-    def add_process_callback(self, callback: AttrCallback[T]) -> None:
+    def add_process_callback(self, callback: AttrSetCallback[T]) -> None:
         if self._process_callbacks is None:
             self._process_callbacks = []
         self._process_callbacks.append(callback)
 
     def has_process_callback(self) -> bool:
         return bool(self._process_callbacks)
 
-    def add_write_display_callback(self, callback: AttrCallback[T]) -> None:
+    def add_write_display_callback(self, callback: AttrSetCallback[T]) -> None:
         if self._write_display_callbacks is None:
             self._write_display_callbacks = []
         self._write_display_callbacks.append(callback)
 
-    @property
-    def sender(self) -> AttrHandlerW | None:
-        return self._setter
-
 
-class AttrRW(AttrR[T], AttrW[T]):
+class AttrRW(AttrR[T, AttributeIORefT], AttrW[T, AttributeIORefT]):
     """A read-write ``Attribute``."""
 
-    def __init__(
-        self,
-        datatype: DataType[T],
-        access_mode=AttrMode.READ_WRITE,
-        group: str | None = None,
-        handler: AttrHandlerRW | None = None,
-        initial_value: T | None = None,
-        description: str | None = None,
-    ) -> None:
-        super().__init__(
-            datatype,  # type: ignore
-            access_mode,
-            group=group,
-            handler=handler if handler else SimpleAttrHandler(),
-            initial_value=initial_value,
-            description=description,
-        )
-
     async def process(self, value: T) -> None:
         await self.set(value)
 

src/fastcs/backend.py

@@ -2,10 +2,11 @@
 from collections import defaultdict
 from collections.abc import Callable, Coroutine
 
+from fastcs.attribute_io_ref import AttributeIORef
 from fastcs.cs_methods import Command, Put, Scan
 from fastcs.datatypes import T
 
-from .attributes import ONCE, AttrHandlerR, AttrHandlerW, AttrR, AttrW
+from .attributes import ONCE, AttrR, AttrW
 from .controller import BaseController, Controller
 from .controller_api import ControllerAPI
 from .exceptions import FastCSError
@@ -36,7 +37,6 @@ def __init__(
     def _link_process_tasks(self):
         for controller_api in self.controller_api.walk_api():
             _link_put_tasks(controller_api)
-            _link_attribute_sender_class(controller_api)
 
     def __del__(self):
         self._stop_scan_tasks()
@@ -87,30 +87,11 @@ def _link_put_tasks(controller_api: ControllerAPI) -> None:
                 attribute.add_process_callback(method.fn)
             case _:
                 raise FastCSError(
-                    f"Mode {attribute.access_mode} does not "
+                    f"Attribute type {type(attribute)} does not "
                     f"support put operations for {name}"
                 )
 
 
-def _link_attribute_sender_class(controller_api: ControllerAPI) -> None:
-    for attr_name, attribute in controller_api.attributes.items():
-        match attribute:
-            case AttrW(sender=AttrHandlerW()):
-                assert not attribute.has_process_callback(), (
-                    f"Cannot assign both put method and Sender object to {attr_name}"
-                )
-
-                callback = _create_sender_callback(attribute)
-                attribute.add_process_callback(callback)
-
-
-def _create_sender_callback(attribute):
-    async def callback(value):
-        await attribute.sender.put(attribute, value)
-
-    return callback
-
-
 def _get_scan_and_initial_coros(
     root_controller_api: ControllerAPI,
 ) -> tuple[list[Callable], list[Callable]]:
@@ -139,7 +120,9 @@ def _add_attribute_updater_tasks(
 ):
     for attribute in controller_api.attributes.values():
         match attribute:
-            case AttrR(updater=AttrHandlerR(update_period=update_period)) as attribute:
+            case (
+                AttrR(_io_ref=AttributeIORef(update_period=update_period)) as attribute
+            ):
                 callback = _create_updater_callback(attribute)
                 if update_period is ONCE:
                     initial_coros.append(callback)
@@ -148,14 +131,11 @@ def _add_attribute_updater_tasks(
 
 
 def _create_updater_callback(attribute: AttrR[T]):
-    updater = attribute.updater
-    assert updater is not None
-
     async def callback():
         try:
-            await updater.update(attribute)
+            await attribute.update()
         except Exception as e:
-            print(f"Update loop in {updater} stopped:\n{e.__class__.__name__}: {e}")
+            print(f"Update loop in {attribute} stopped:\n{e.__class__.__name__}: {e}")
             raise
 
     return callback