Reorder generic args of AttributeIO and simplify defaulted TypeVars

Author: James Souter

src/fastcs/attribute_io.py

@@ -5,12 +5,12 @@
 from fastcs.datatypes import T
 
 
-class AttributeIO(Generic[AttributeIORefT, T]):
+class AttributeIO(Generic[T, AttributeIORefT]):
     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[0]
+        cls.ref_type = args[1]
 
     async def update(self, attr: AttrR[T, AttributeIORefT]) -> None:
         raise NotImplementedError()
@@ -19,4 +19,4 @@ async def send(self, attr: AttrRW[T, AttributeIORefT], value: T) -> None:
         raise NotImplementedError()
 
 
-AnyAttributeIO = AttributeIO[AttributeIORef, T]
+AnyAttributeIO = AttributeIO[T, AttributeIORef]

src/fastcs/attribute_io_ref.py

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

src/fastcs/attributes.py

@@ -4,23 +4,16 @@
 from collections.abc import Callable
 from typing import Generic
 
-from typing_extensions import TypeVar
-
 import fastcs
-from fastcs.attribute_io_ref import AttributeIORef
 
+from .attribute_io_ref import AttributeIORefT
 from .datatypes import ATTRIBUTE_TYPES, AttrSetCallback, AttrUpdateCallback, DataType, T
 
-# TODO rename this: typevar with default
-AttributeIORefTD = TypeVar(
-    "AttributeIORefTD", bound=AttributeIORef, default=AttributeIORef, covariant=True
-)
-
 ONCE = float("inf")
 """Special value to indicate that an attribute should be updated once on start up."""
 
 
-class Attribute(Generic[T, AttributeIORefTD]):
+class Attribute(Generic[T, AttributeIORefT]):
     """Base FastCS attribute.
 
     Instances of this class added to a ``Controller`` will be used by the backend.
@@ -29,7 +22,7 @@ class Attribute(Generic[T, AttributeIORefTD]):
     def __init__(
         self,
         datatype: DataType[T],
-        io_ref: AttributeIORefTD | None = None,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
         description: str | None = None,
     ) -> None:
@@ -48,7 +41,7 @@ def __init__(
         self._update_datatype_callbacks: list[Callable[[DataType[T]], None]] = []
 
     @property
-    def io_ref(self) -> AttributeIORefTD:
+    def io_ref(self) -> AttributeIORefT:
         if self._io_ref is None:
             raise RuntimeError(f"{self} has no AttributeIORef")
         return self._io_ref
@@ -86,13 +79,13 @@ def update_datatype(self, datatype: DataType[T]) -> None:
             callback(datatype)
 
 
-class AttrR(Attribute[T, AttributeIORefTD]):
+class AttrR(Attribute[T, AttributeIORefT]):
     """A read-only ``Attribute``."""
 
     def __init__(
         self,
         datatype: DataType[T],
-        io_ref: AttributeIORefTD | None = None,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
         initial_value: T | None = None,
         description: str | None = None,
@@ -133,13 +126,13 @@ async def update(self):
             await asyncio.gather(*[cb() for cb in self._on_update_callbacks])
 
 
-class AttrW(Attribute[T, AttributeIORefTD]):
+class AttrW(Attribute[T, AttributeIORefT]):
     """A write-only ``Attribute``."""
 
     def __init__(
         self,
         datatype: DataType[T],
-        io_ref: AttributeIORefTD | None = None,
+        io_ref: AttributeIORefT | None = None,
         group: str | None = None,
         description: str | None = None,
     ) -> None:
@@ -180,7 +173,7 @@ def add_write_display_callback(self, callback: AttrSetCallback[T]) -> None:
         self._write_display_callbacks.append(callback)
 
 
-class AttrRW(AttrR[T, AttributeIORefTD], AttrW[T, AttributeIORefTD]):
+class AttrRW(AttrR[T, AttributeIORefT], AttrW[T, AttributeIORefT]):
     """A read-write ``Attribute``."""
 
     async def process(self, value: T) -> None:

src/fastcs/controller.py

@@ -6,7 +6,9 @@
 from typing import get_type_hints
 
 from fastcs.attribute_io import AttributeIO
+from fastcs.attribute_io_ref import AttributeIORefT
 from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
+from fastcs.datatypes import T
 
 
 class BaseController:
@@ -21,7 +23,7 @@ def __init__(
         self,
         path: list[str] | None = None,
         description: str | None = None,
-        ios: Sequence[AttributeIO] | None = None,
+        ios: Sequence[AttributeIO[T, AttributeIORefT]] | None = None,
     ) -> None:
         if (
             description is not None
@@ -192,7 +194,9 @@ class Controller(BaseController):
     """
 
     def __init__(
-        self, description: str | None = None, ios: Sequence[AttributeIO] | None = None
+        self,
+        description: str | None = None,
+        ios: Sequence[AttributeIO[T, AttributeIORefT]] | None = None,
     ) -> None:
         super().__init__(description=description, ios=ios)
 
@@ -210,6 +214,8 @@ class SubController(BaseController):
     root_attribute: Attribute | None = None
 
     def __init__(
-        self, description: str | None = None, ios: Sequence[AttributeIO] | None = None
+        self,
+        description: str | None = None,
+        ios: Sequence[AttributeIO[T, AttributeIORefT]] | None = None,
     ) -> None:
         super().__init__(description=description, ios=ios)

src/fastcs/demo/controllers.py

@@ -35,7 +35,7 @@ class TemperatureControllerAttributeIORef(AttributeIORef):
 
 
 class TemperatureControllerAttributeIO(
-    AttributeIO[TemperatureControllerAttributeIORef, NumberT]
+    AttributeIO[NumberT, TemperatureControllerAttributeIORef]
 ):
     def __init__(self, connection: IPConnection, suffix: str):
         self._connection = connection

tests/assertable_controller.py

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

tests/test_attribute.py

@@ -69,7 +69,7 @@ async def test_attribute_io():
     class MyAttributeIORef(AttributeIORef):
         cool: int
 
-    class MyAttributeIO(AttributeIO[MyAttributeIORef, int]):
+    class MyAttributeIO(AttributeIO[int, MyAttributeIORef]):
         async def update(self, attr: AttrR[T, MyAttributeIORef]):
             print("I am updating", self.ref_type, attr.io_ref.cool)
 
@@ -123,10 +123,10 @@ class DemoParameterAttributeIORef(AttributeIORef, Generic[NumberT]):
         max: NumberT | None = None
         read_only: bool = False
 
-    class DemoParameterAttributeIO(AttributeIO[DemoParameterAttributeIORef, NumberT]):
+    class DemoParameterAttributeIO(AttributeIO[NumberT, DemoParameterAttributeIORef]):
         async def update(
             self,
-            attr: AttrR[NumberT],
+            attr: AttrR[NumberT, DemoParameterAttributeIORef],
         ):
             # OK, so this doesn't really work when we have min and maxes...
             await attr.set(attr.get() + 1)
@@ -145,14 +145,12 @@ async def send(
 
             if (io_min := attr.io_ref.min) is not None and value < io_min:
                 raise RuntimeError(
-                    f"Could not set {attr.io_ref.name} to {value}, "
-                    "min is {attr.io_ref.min}"
+                    f"Could not set {attr.io_ref.name} to {value}, min is {io_min}"
                 )
 
             if (io_max := attr.io_ref.max) is not None and value > io_max:
                 raise RuntimeError(
-                    f"Could not set {attr.io_ref.name} to {value}, "
-                    f"max is {attr.io_ref.max}"
+                    f"Could not set {attr.io_ref.name} to {value}, max is {io_max}"
                 )
             # TODO: we should always end send with a update_display_without_process...
 

tests/test_backend.py

@@ -101,7 +101,7 @@ class AttributeIORefTimesCalled(AttributeIORef):
         update_period: float | None = None
         _times_called = 0
 
-    class AttributeIOTimesCalled(AttributeIO[AttributeIORefTimesCalled, int]):
+    class AttributeIOTimesCalled(AttributeIO[int, AttributeIORefTimesCalled]):
         async def update(self, attr: AttrR[int, AttributeIORefTimesCalled]):
             attr.io_ref._times_called += 1
             await attr.set(attr.io_ref._times_called)