Redesign attribute callbacks

Author: GDYendell

docs/conf.py

@@ -88,6 +88,8 @@
     ("py:class", "p4p.nt.ndarray.NTNDArray"),
     ("py:class", "p4p.nt.NTTable"),
     # Problems in FastCS itself
+    ("py:class", "T"),
+    ("py:class", "AttrUpdateCallback"),
     ("py:class", "fastcs.transport.epics.pva.pvi_tree._PviSignalInfo"),
     ("py:class", "fastcs.logging._logging.LogLevel"),
     ("py:class", "fastcs.logging._graylog.GraylogEndpoint"),

docs/snippets/static13.py

@@ -100,7 +100,7 @@ async def update_voltages(self):
     @command()
     async def disable_all(self) -> None:
         for rc in self._ramp_controllers:
-            await rc.enabled.process(OnOffEnum.Off)
+            await rc.enabled.put(OnOffEnum.Off, sync_setpoint=True)
             # TODO: The requests all get concatenated and the sim doesn't handle it
             await asyncio.sleep(0.1)
 

src/fastcs/attribute_io.py

@@ -1,7 +1,7 @@
 from typing import Any, Generic, cast, get_args
 
 from fastcs.attribute_io_ref import AttributeIORef, AttributeIORefT
-from fastcs.attributes import AttrR, AttrRW
+from fastcs.attributes import AttrR, AttrW
 from fastcs.datatypes import T
 from fastcs.tracer import Tracer
 
@@ -32,7 +32,7 @@ def __init__(self):
     async def update(self, attr: AttrR[T, AttributeIORefT]) -> None:
         raise NotImplementedError()
 
-    async def send(self, attr: AttrRW[T, AttributeIORefT], value: T) -> None:
+    async def send(self, attr: AttrW[T, AttributeIORefT], value: T) -> None:
         raise NotImplementedError()
 
 

src/fastcs/attribute_io_ref.py

@@ -20,5 +20,5 @@ class AttributeIORef:
 
 
 AttributeIORefT = TypeVar(
-    "AttributeIORefT", bound=AttributeIORef, default=AttributeIORef
+    "AttributeIORefT", bound=AttributeIORef, default=AttributeIORef, covariant=True
 )

src/fastcs/attributes.py

@@ -1,17 +1,11 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Callable
-from typing import Generic
+from collections.abc import Awaitable, Callable
+from typing import Any, Generic
 
 from fastcs.attribute_io_ref import AttributeIORefT
-from fastcs.datatypes import (
-    ATTRIBUTE_TYPES,
-    AttrSetCallback,
-    AttrUpdateCallback,
-    DataType,
-    T,
-)
+from fastcs.datatypes import ATTRIBUTE_TYPES, DataType, T
 from fastcs.tracer import Tracer
 
 ONCE = float("inf")
@@ -97,6 +91,10 @@ def __repr__(self):
         return f"{self.__class__.__name__}({self._name}, {self._datatype})"
 
 
+AttrUpdateCallback = Callable[["AttrR[T, Any]"], Awaitable[None]]
+AttrOnSetCallback = Callable[[T], Awaitable[None]]
+
+
 class AttrR(Attribute[T, AttributeIORefT]):
     """A read-only ``Attribute``."""
 
@@ -108,42 +106,66 @@ def __init__(
         initial_value: T | None = None,
         description: str | None = None,
     ) -> None:
-        super().__init__(
-            datatype,  # type: ignore
-            io_ref,
-            group,
-            description=description,
-        )
+        super().__init__(datatype, io_ref, group, description=description)
         self._value: T = (
             datatype.initial_value if initial_value is None else initial_value
         )
-        self._on_set_callbacks: list[AttrSetCallback[T]] | None = None
-        self._on_update_callbacks: list[AttrUpdateCallback] | None = None
+        self._update_callback: AttrUpdateCallback[T] | None = None
+        """Callback to update the value of the attribute from the source"""
+        self._on_set_callbacks: list[AttrOnSetCallback[T]] | None = None
+        """Callbacks to publish changes to the value of the attribute"""
 
     def get(self) -> T:
+        """Get the cached value of the attribute."""
         return self._value
 
     async def set(self, value: T) -> None:
+        """Set the value of the attibute
+
+        This sets the cached value of the attribute presented in the API. It should
+        generally only be called from an IO or a controller that is updating the value
+        from some underlying source.
+
+        To request a change to the setpoint of the attribute, use the ``put`` method,
+        which will attempt to apply the change to the underlying source.
+
+        """
         self.log_event("Attribute set", attribute=self, value=value)
 
         self._value = self._datatype.validate(value)
 
         if self._on_set_callbacks is not None:
             await asyncio.gather(*[cb(self._value) for cb in self._on_set_callbacks])
 
-    def add_set_callback(self, callback: AttrSetCallback[T]) -> None:
+    def add_on_set_callback(self, callback: AttrOnSetCallback[T]) -> None:
+        """Add a callback to be called when the attribute is set
+
+        The callback will be called with the new value when the attribute is set.
+
+        """
         if self._on_set_callbacks is None:
             self._on_set_callbacks = []
         self._on_set_callbacks.append(callback)
 
-    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])
+        """Update the attribute value via its IO, if it has one"""
+        if self._update_callback is not None:
+            await self._update_callback(self)
+
+    def set_update_callback(self, callback: AttrUpdateCallback[T]):
+        """Set the callback to update the value of the attribute from the source
+
+        The callback will be called with the attribute when it needs updating.
+
+        """
+        if self._update_callback is not None:
+            raise RuntimeError("Attribute already has an update callback")
+
+        self._update_callback = callback
+
+
+AttrOnPutCallback = Callable[["AttrW[T, Any]", T], Awaitable[None]]
+AttrSyncSetpointCallback = Callable[[T], Awaitable[None]]
 
 
 class AttrW(Attribute[T, AttributeIORefT]):
@@ -162,41 +184,83 @@ def __init__(
             group,
             description=description,
         )
-        self._process_callbacks: list[AttrSetCallback[T]] | None = None
-        self._write_display_callbacks: list[AttrSetCallback[T]] | None = None
+        self._on_put_callback: AttrOnPutCallback[T] | None = None
+        """Callback to action the put of a new value to attribute"""
+        self._sync_setpoint_callbacks: list[AttrSyncSetpointCallback[T]] = []
+        """Callbacks to publish changes to the setpoint of the attribute"""
+
+    async def put(self, setpoint: T, sync_setpoint: bool = False) -> None:
+        """Set the setpoint of the attribute
+
+        This should be called by clients to the attribute such as transports to apply a
+        change to the attribute. The ``_on_put_callback`` will be called with this new
+        setpoint, which may or may not take effect depending on the validity of the new
+        value. For example, if the attribute has an IO to some device, the value might
+        be rejected.
+
+        To directly change the value of the attribute, for example from an update loop
+        that has read a new value from some underlying source, call the ``set`` method.
+
+        """
+        setpoint = self._datatype.validate(setpoint)
+        if self._on_put_callback is not None:
+            await self._on_put_callback(self, setpoint)
+
+        if sync_setpoint:
+            await self._call_sync_setpoint_callbacks(setpoint)
+
+    async def _call_sync_setpoint_callbacks(self, setpoint: T) -> None:
+        if self._sync_setpoint_callbacks:
+            await asyncio.gather(
+                *[cb(setpoint) for cb in self._sync_setpoint_callbacks]
+            )
 
-    async def process(self, value: T) -> None:
-        await self.process_without_display_update(value)
-        await self.update_display_without_process(value)
+    def set_on_put_callback(self, callback: AttrOnPutCallback[T]) -> None:
+        """Set the callback to call when the setpoint is changed
 
-    async def process_without_display_update(self, value: T) -> None:
-        value = self._datatype.validate(value)
-        if self._process_callbacks:
-            await asyncio.gather(*[cb(value) for cb in self._process_callbacks])
+        The callback will be called with the attribute and the new setpoint.
 
-    async def update_display_without_process(self, value: T) -> None:
-        value = self._datatype.validate(value)
-        if self._write_display_callbacks:
-            await asyncio.gather(*[cb(value) for cb in self._write_display_callbacks])
+        """
+        if self._on_put_callback is not None:
+            raise RuntimeError("Attribute already has an on put callback")
 
-    def add_process_callback(self, callback: AttrSetCallback[T]) -> None:
-        if self._process_callbacks is None:
-            self._process_callbacks = []
-        self._process_callbacks.append(callback)
+        self._on_put_callback = callback
 
-    def has_process_callback(self) -> bool:
-        return bool(self._process_callbacks)
+    def add_sync_setpoint_callback(self, callback: AttrSyncSetpointCallback[T]) -> None:
+        """Add a callback to publish changes to the setpoint of the attribute
 
-    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)
+        The callback will be called with the new setpoint.
+
+        """
+        self._sync_setpoint_callbacks.append(callback)
 
 
 class AttrRW(AttrR[T, AttributeIORefT], AttrW[T, AttributeIORefT]):
     """A read-write ``Attribute``."""
 
-    async def process(self, value: T) -> None:
+    def __init__(
+        self,
+        datatype: DataType[T],
+        io_ref: AttributeIORefT | None = None,
+        group: str | None = None,
+        initial_value: T | None = None,
+        description: str | None = None,
+    ):
+        super().__init__(datatype, io_ref, group, initial_value, description)
+
+        self._setpoint_initialised = False
+
+        if io_ref is None:
+            self.set_on_put_callback(self._internal_put)
+
+    async def _internal_put(self, attr: AttrW[T, AttributeIORefT], value: T):
+        """Set value directly when Attribute has no IO"""
+        assert attr is self
         await self.set(value)
 
-        await super().process(value)  # type: ignore
+    async def set(self, value: T):
+        await super().set(value)
+
+        if not self._setpoint_initialised:
+            await self._call_sync_setpoint_callbacks(value)
+            self._setpoint_initialised = True

src/fastcs/controller.py

@@ -7,7 +7,11 @@
 
 from fastcs.attribute_io import AttributeIO
 from fastcs.attribute_io_ref import AttributeIORefT
-from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
+from fastcs.attributes import (
+    Attribute,
+    AttrR,
+    AttrW,
+)
 from fastcs.datatypes import T
 from fastcs.tracer import Tracer
 
@@ -58,39 +62,20 @@ async def attribute_initialise(self) -> None:
     def _add_io_callbacks(self):
         for attr in self.attributes.values():
             ref = attr.io_ref if attr.has_io_ref() else None
+            if ref is None:
+                continue
+
             io = self._attribute_ref_io_map.get(type(ref))
+            if io is None:
+                raise ValueError(
+                    f"{self.__class__.__name__} does not have an AttributeIO "
+                    f"to handle {attr.io_ref.__class__.__name__}"
+                )
+
             if isinstance(attr, AttrW):
-                attr.add_process_callback(self._create_send_callback(io, attr, ref))
+                attr.set_on_put_callback(io.send)
             if isinstance(attr, AttrR):
-                attr.add_update_callback(self._create_update_callback(io, attr, ref))
-
-    def _create_send_callback(self, io, attr, ref):
-        if ref is None:
-
-            async def send_callback(value):
-                await attr.update_display_without_process(value)
-                if isinstance(attr, AttrRW):
-                    await attr.set(value)
-        else:
-
-            async def send_callback(value):
-                await io.send(attr, value)
-
-        return send_callback
-
-    def _create_update_callback(self, io, attr, ref):
-        if ref is None:
-
-            async def error_callback():
-                raise RuntimeError("Can't call update on Attributes without an io_ref")
-
-            return error_callback
-        else:
-
-            async def update_callback():
-                await io.update(attr)
-
-            return update_callback
+                attr.set_update_callback(io.update)
 
     @property
     def path(self) -> list[str]:
@@ -145,14 +130,6 @@ def _validate_io(self, ios: Sequence[AttributeIO[T, AttributeIORefT]]):
                     f"More than one AttributeIO class handles {ref_type.__name__}"
                 )
 
-        for attr in self.attributes.values():
-            if not attr.has_io_ref():
-                continue
-            assert type(attr.io_ref) in self._attribute_ref_io_map, (
-                f"{self.__class__.__name__} does not have an AttributeIO to handle "
-                f"{attr.io_ref.__class__.__name__}"
-            )
-
     def add_attribute(self, name, attribute: Attribute):
         if name in self.attributes and attribute is not self.attributes[name]:
             raise ValueError(

src/fastcs/datatypes.py

@@ -2,7 +2,6 @@
 
 import enum
 from abc import abstractmethod
-from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from functools import cached_property
 from typing import Any, Generic, TypeVar
@@ -24,10 +23,6 @@
 ATTRIBUTE_TYPES: tuple[type] = T.__constraints__  # type: ignore
 
 
-AttrSetCallback = Callable[[T], Awaitable[None]]
-AttrUpdateCallback = Callable[[], Awaitable[None]]
-
-
 @dataclass(frozen=True)
 class DataType(Generic[T]):
     """Generic datatype mapping to a python type, with additional metadata."""

src/fastcs/demo/controllers.py

@@ -87,7 +87,7 @@ def __init__(self, settings: TemperatureControllerSettings) -> None:
     @command()
     async def cancel_all(self) -> None:
         for rc in self._ramp_controllers:
-            await rc.enabled.process(OnOffEnum.Off)
+            await rc.enabled.put(OnOffEnum.Off, sync_setpoint=True)
             # TODO: The requests all get concatenated and the sim doesn't handle it
             await asyncio.sleep(0.1)