Get around having to register ref to AtttributeIO method signatures

Author: James Souter

src/fastcs/attribute_io.py

@@ -12,14 +12,10 @@ def __init_subclass__(cls) -> None:
         args = get_args(cast(Any, cls).__orig_bases__[0])
         cls.ref_type = args[0]
 
-    async def update(
-        self, attr: AttrR[T, AttributeIORefT], ref: AttributeIORefT
-    ) -> None:
+    async def update(self, attr: AttrR[T, AttributeIORefT]) -> None:
         raise NotImplementedError()
 
-    async def send(
-        self, attr: AttrRW[T, AttributeIORefT], ref: AttributeIORefT, value: T
-    ) -> None:
+    async def send(self, attr: AttrRW[T, AttributeIORefT], value: T) -> None:
         raise NotImplementedError()
 
 

src/fastcs/attributes.py

@@ -37,7 +37,7 @@ def __init__(
             f"Attr type must be one of {ATTRIBUTE_TYPES}, "
             "received type {datatype.dtype}"
         )
-        self.io_ref = io_ref
+        self._io_ref = io_ref
         self._datatype: DataType[T] = datatype
         self._group = group
         self.enabled = True
@@ -47,6 +47,15 @@ def __init__(
         # 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) -> AttributeIORefTD:
+        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

src/fastcs/backend.py

@@ -120,7 +120,9 @@ def _add_attribute_updater_tasks(
 ):
     for attribute in controller_api.attributes.values():
         match attribute:
-            case AttrR(io_ref=AttributeIORef(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)

src/fastcs/controller.py

@@ -58,17 +58,16 @@ async def attribute_initialise(self) -> None:
 
     def _add_io_callbacks(self):
         for attr in self.attributes.values():
-            io = self._attribute_ref_io_map.get(type(attr.io_ref), None)
+            ref = attr.io_ref if attr.has_io_ref() else None
+            io = self._attribute_ref_io_map.get(type(ref), None)
             if isinstance(attr, AttrW):
                 # is it on process or write_display?
-                attr.add_process_callback(self._create_send_callback(io, attr))
-            if attr.io_ref is None or io is None:
-                continue
+                attr.add_process_callback(self._create_send_callback(io, attr, ref))
             if isinstance(attr, AttrR):
-                attr.add_update_callback(self._create_update_callback(io, attr))
+                attr.add_update_callback(self._create_update_callback(io, attr, ref))
 
-    def _create_send_callback(self, io, attr):
-        if attr.io_ref is None:
+    def _create_send_callback(self, io, attr, ref):
+        if ref is None:
 
             async def send_callback(value):
                 await attr.update_display_without_process(value)
@@ -77,13 +76,13 @@ async def send_callback(value):
         else:
 
             async def send_callback(value):
-                await io.send(attr, attr.io_ref, value)
+                await io.send(attr, value)
                 # TODO, should we just then call the above send_callback here?
 
         return send_callback
 
-    def _create_update_callback(self, io, attr):
-        if io is None or attr.io_ref is None:
+    def _create_update_callback(self, io, attr, ref):
+        if io is None or ref is None:
 
             async def error_callback():
                 raise RuntimeError("No AttributeIO registered to handle update")
@@ -92,7 +91,7 @@ async def error_callback():
         else:
 
             async def update_callback():
-                await io.update(attr, attr.io_ref)
+                await io.update(attr)
 
             return update_callback
 
@@ -154,7 +153,7 @@ def _validate_io(self):
         """Validate that each Attribute has an AttributeIORef for which the
         controller has an associated AttributeIO class."""
         for attr in self.attributes.values():
-            if attr.io_ref is None:
+            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 "

src/fastcs/demo/controllers.py

@@ -4,15 +4,18 @@
 import enum
 import json
 from dataclasses import dataclass
+from typing import TypeVar
 
 from fastcs.attribute_io import AttributeIO
 from fastcs.attribute_io_ref import AttributeIORef
 from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.connections import IPConnection, IPConnectionSettings
 from fastcs.controller import Controller, SubController
-from fastcs.datatypes import Enum, Float, Int, T
+from fastcs.datatypes import Enum, Float, Int
 from fastcs.wrappers import command, scan
 
+NumberT = TypeVar("NumberT", int, float)
+
 
 class OnOffEnum(enum.StrEnum):
     Off = "0"
@@ -32,23 +35,25 @@ class TemperatureControllerAttributeIORef(AttributeIORef):
 
 
 class TemperatureControllerAttributeIO(
-    AttributeIO[TemperatureControllerAttributeIORef, T]
+    AttributeIO[TemperatureControllerAttributeIORef, NumberT]
 ):
     def __init__(self, connection: IPConnection, suffix: str):
         self._connection = connection
         self.suffix = suffix
 
     async def send(
-        self, attr: AttrW, ref: TemperatureControllerAttributeIORef, value: T
+        self, attr: AttrW[NumberT, TemperatureControllerAttributeIORef], value: NumberT
     ) -> None:
         await self._connection.send_command(
-            f"{ref.name}{self.suffix}={attr.dtype(value)}\r\n"
+            f"{attr.io_ref.name}{self.suffix}={attr.dtype(value)}\r\n"
         )
 
     async def update(
-        self, attr: AttrR, ref: TemperatureControllerAttributeIORef
+        self, attr: AttrR[NumberT, TemperatureControllerAttributeIORef]
     ) -> None:
-        response = await self._connection.send_query(f"{ref.name}{self.suffix}?\r\n")
+        response = await self._connection.send_query(
+            f"{attr.io_ref.name}{self.suffix}?\r\n"
+        )
         response = response.strip("\r\n")
 
         await attr.set(attr.dtype(response))
@@ -101,7 +106,9 @@ class TemperatureRampController(SubController):
     enabled = AttrRW(
         Enum(OnOffEnum), io_ref=TemperatureControllerAttributeIORef(name="N")
     )
-    target = AttrR(Float(prec=3), io_ref=TemperatureControllerAttributeIORef(name="T"))
+    target = AttrR(
+        Float(prec=3), io_ref=TemperatureControllerAttributeIORef(name="NumberT")
+    )
     actual = AttrR(Float(prec=3), io_ref=TemperatureControllerAttributeIORef(name="A"))
     voltage = AttrR(Float(prec=3))
 

tests/assertable_controller.py

@@ -24,10 +24,10 @@ class MyTestAttributeIORef(AttributeIORef):
 
 
 class MyTestAttributeIO(AttributeIO[MyTestAttributeIORef, T]):
-    async def update(self, attr: AttrR[T], ref: MyTestAttributeIORef):
+    async def update(self, attr: AttrR[T]):
         print(f"update {attr}")
 
-    async def send(self, attr: AttrW[T], ref: MyTestAttributeIORef, value: T):
+    async def send(self, attr: AttrW[T], value: T):
         print(f"sending {attr} = {value}")
 
 

tests/test_attribute.py

@@ -13,7 +13,7 @@
     AttrW,
 )
 from fastcs.controller import Controller
-from fastcs.datatypes import Float, Int, String
+from fastcs.datatypes import Float, Int, String, T
 
 NumberT = TypeVar("NumberT", int, float)
 
@@ -70,8 +70,8 @@ class MyAttributeIORef(AttributeIORef):
         cool: int
 
     class MyAttributeIO(AttributeIO[MyAttributeIORef, int]):
-        async def update(self, attr: AttrR, ref: MyAttributeIORef):
-            print("I am updating", self.ref_type, ref.cool)
+        async def update(self, attr: AttrR[T, MyAttributeIORef]):
+            print("I am updating", self.ref_type, attr.io_ref.cool)
 
     class MyController(Controller):
         my_attr = AttrR(Int(), io_ref=MyAttributeIORef(cool=5))
@@ -127,30 +127,32 @@ class DemoParameterAttributeIO(AttributeIO[DemoParameterAttributeIORef, NumberT]
         async def update(
             self,
             attr: AttrR[NumberT],
-            ref: DemoParameterAttributeIORef,
         ):
             # OK, so this doesn't really work when we have min and maxes...
             await attr.set(attr.get() + 1)
 
         async def send(
             self,
             attr: AttrW[NumberT, DemoParameterAttributeIORef],
-            ref: DemoParameterAttributeIORef,
             value: NumberT,
         ) -> None:
             if (
-                ref.read_only
+                attr.io_ref.read_only
             ):  # TODO, this isn't necessary as we can not call process on this anyway
-                raise RuntimeError(f"Could not set read only attribute {ref.name}")
+                raise RuntimeError(
+                    f"Could not set read only attribute {attr.io_ref.name}"
+                )
 
-            if (io_min := ref.min) is not None and value < io_min:
+            if (io_min := attr.io_ref.min) is not None and value < io_min:
                 raise RuntimeError(
-                    f"Could not set {ref.name} to {value}, min is {ref.min}"
+                    f"Could not set {attr.io_ref.name} to {value}, "
+                    "min is {attr.io_ref.min}"
                 )
 
-            if (io_max := ref.max) is not None and value > io_max:
+            if (io_max := attr.io_ref.max) is not None and value > io_max:
                 raise RuntimeError(
-                    f"Could not set {ref.name} to {value}, max is {ref.max}"
+                    f"Could not set {attr.io_ref.name} to {value}, "
+                    f"max is {attr.io_ref.max}"
                 )
             # TODO: we should always end send with a update_display_without_process...
 

tests/test_backend.py

@@ -102,9 +102,9 @@ class AttributeIORefTimesCalled(AttributeIORef):
         _times_called = 0
 
     class AttributeIOTimesCalled(AttributeIO[AttributeIORefTimesCalled, int]):
-        async def update(self, attr: AttrR[int], ref: AttributeIORefTimesCalled):
-            ref._times_called += 1
-            await attr.set(ref._times_called)
+        async def update(self, attr: AttrR[int, AttributeIORefTimesCalled]):
+            attr.io_ref._times_called += 1
+            await attr.set(attr.io_ref._times_called)
 
     class MyController(Controller):
         update_once = AttrR(Int(), io_ref=AttributeIORefTimesCalled(update_period=ONCE))