Integrate hinted attribute validation into controller

Check datatype and access mode in add_attribute, check for
unintrospected type hints after initialise

Author: GDYendell

src/fastcs/control_system.py

@@ -13,7 +13,6 @@
 from fastcs.logging import logger as _fastcs_logger
 from fastcs.tracer import Tracer
 from fastcs.transport import Transport
-from fastcs.util import validate_hinted_attributes
 
 tracer = Tracer(name=__name__)
 logger = _fastcs_logger.bind(logger_name=__name__)
@@ -85,7 +84,7 @@ def _stop_scan_tasks(self):
 
     async def serve(self, interactive: bool = True) -> None:
         await self._controller.initialise()
-        validate_hinted_attributes(self._controller)
+        self._controller.validate_hinted_attributes()
         self._controller.connect_attribute_ios()
 
         self.controller_api = build_controller_api(self._controller)

src/fastcs/controller.py

@@ -3,12 +3,12 @@
 from collections import Counter
 from collections.abc import Iterator, Mapping, MutableMapping, Sequence
 from copy import deepcopy
-from typing import get_type_hints
+from typing import _GenericAlias, get_args, get_origin, get_type_hints  # type: ignore
 
 from fastcs.attribute_io import AttributeIO
 from fastcs.attribute_io_ref import AttributeIORefT
 from fastcs.attributes import Attribute, AttrR, AttrW
-from fastcs.datatypes import T
+from fastcs.datatypes import DataType, T
 from fastcs.tracer import Tracer
 
 
@@ -39,16 +39,46 @@ def __init__(
         self._path: list[str] = path or []
         self.__sub_controller_tree: dict[str, BaseController] = {}
 
+        self.__hinted_attributes = self._parse_attribute_type_hints()
         self._bind_attrs()
 
         ios = ios or []
         self._attribute_ref_io_map = {io.ref_type: io for io in ios}
         self._validate_io(ios)
 
+    def _parse_attribute_type_hints(
+        self,
+    ) -> dict[str, tuple[type[Attribute], type[DataType]]]:
+        hinted_attributes = {}
+        for name, hint in get_type_hints(type(self)).items():
+            if not isinstance(hint, _GenericAlias):  # e.g. AttrR[int]
+                continue
+
+            origin = get_origin(hint)
+            if not isinstance(origin, type) or not issubclass(origin, Attribute):
+                continue
+
+            hinted_attributes[name] = (origin, get_args(hint)[0])
+
+        return hinted_attributes
+
     async def initialise(self):
         """Hook to dynamically add attributes before building the API"""
         pass
 
+    def validate_hinted_attributes(self):
+        """Validate ``Attribute`` type-hints were introspected during initialisation"""
+        for name in self.__hinted_attributes:
+            attr = getattr(self, name, None)
+            if attr is None:
+                raise RuntimeError(
+                    f"Controller `{self.__class__.__name__}` failed to introspect "
+                    f"hinted attribute `{name}` during initialisation"
+                )
+
+        for subcontroller in self.sub_controllers.values():
+            subcontroller.validate_hinted_attributes()  # noqa: SLF001
+
     def connect_attribute_ios(self) -> None:
         """Connect ``Attribute`` callbacks to ``AttributeIO``s"""
         for attr in self.attributes.values():
@@ -126,24 +156,39 @@ def _validate_io(self, ios: Sequence[AttributeIO[T, AttributeIORefT]]):
                     f"More than one AttributeIO class handles {ref_type.__name__}"
                 )
 
-    def add_attribute(self, name, attribute: Attribute):
-        if name in self.attributes and attribute is not self.attributes[name]:
+    def add_attribute(self, name, attr: Attribute):
+        if name in self.attributes:
             raise ValueError(
-                f"Cannot add attribute {attribute}. "
+                f"Cannot add attribute {attr}. "
                 f"Controller {self} has has existing attribute {name}: "
                 f"{self.attributes[name]}"
             )
+        elif name in self.__hinted_attributes:
+            attr_class, attr_dtype = self.__hinted_attributes[name]
+            if not isinstance(attr, attr_class):
+                raise RuntimeError(
+                    f"Controller '{self.__class__.__name__}' introspection of "
+                    f"hinted attribute '{name}' does not match defined access mode. "
+                    f"Expected '{attr_class.__name__}', got '{type(attr).__name__}'."
+                )
+            if attr_dtype is not None and attr_dtype != attr.datatype.dtype:
+                raise RuntimeError(
+                    f"Controller '{self.__class__.__name__}' introspection of "
+                    f"hinted attribute '{name}' does not match defined datatype. "
+                    f"Expected '{attr_dtype.__name__}', "
+                    f"got '{attr.datatype.dtype.__name__}'."
+                )
         elif name in self.__sub_controller_tree.keys():
             raise ValueError(
-                f"Cannot add attribute {attribute}. "
+                f"Cannot add attribute {attr}. "
                 f"Controller {self} has existing sub controller {name}: "
                 f"{self.__sub_controller_tree[name]}"
             )
 
-        attribute.set_name(name)
-        attribute.set_path(self.path)
-        self.attributes[name] = attribute
-        super().__setattr__(name, attribute)
+        attr.set_name(name)
+        attr.set_path(self.path)
+        self.attributes[name] = attr
+        super().__setattr__(name, attr)
 
     def add_sub_controller(self, name: str, sub_controller: BaseController):
         if name in self.__sub_controller_tree.keys():

src/fastcs/util.py

@@ -1,10 +1,7 @@
 import re
-from typing import _GenericAlias, get_args, get_origin, get_type_hints  # type: ignore
 
 import numpy as np
 
-from fastcs.attributes import Attribute
-from fastcs.controller import BaseController
 from fastcs.datatypes import Bool, DataType, Float, Int, String
 
 
@@ -29,47 +26,3 @@ def numpy_to_fastcs_datatype(np_type) -> DataType:
         return Bool()
     else:
         return String()
-
-
-def validate_hinted_attributes(controller: BaseController):
-    """Validate ``Attribute`` type-hints  match dynamically intropected instances
-
-    For each type-hinted attribute, validate that a corresponding instance exists in the
-    controller with the correct access mode and datatype.
-    """
-    for subcontroller in controller.sub_controllers.values():
-        validate_hinted_attributes(subcontroller)
-    hints = {
-        k: v
-        for k, v in get_type_hints(type(controller)).items()
-        if isinstance(v, _GenericAlias | type)
-    }
-    for name, hint in hints.items():
-        if isinstance(hint, type):
-            attr_class = hint
-            attr_dtype = None
-        else:
-            attr_class = get_origin(hint)
-            attr_dtype = get_args(hint)[0]
-        if not issubclass(attr_class, Attribute):
-            continue
-
-        attr = getattr(controller, name, None)
-        if attr is None:
-            raise RuntimeError(
-                f"Controller `{controller.__class__.__name__}` failed to introspect "
-                f"hinted attribute `{name}` during initialisation"
-            )
-        if attr_class is not type(attr):
-            raise RuntimeError(
-                f"Controller '{controller.__class__.__name__}' introspection of "
-                f"hinted attribute '{name}' does not match defined access mode. "
-                f"Expected '{attr_class.__name__}', got '{type(attr).__name__}'."
-            )
-        if attr_dtype is not None and attr_dtype != attr.datatype.dtype:
-            raise RuntimeError(
-                f"Controller '{controller.__class__.__name__}' introspection of hinted "
-                f"attribute '{name}' does not match defined datatype. "
-                f"Expected '{attr_dtype.__name__}', "
-                f"got '{attr.datatype.dtype.__name__}'."
-            )

tests/test_attribute.py

@@ -208,9 +208,7 @@ async def initialise(self):
                         initial_value=parameter_response.get("value", None),
                     )
 
-                    self.attributes[ref.name] = attr
-                    setattr(self, ref.name, attr)
-
+                    self.add_attribute(ref.name, attr)
                 except Exception as e:
                     print(
                         "Exception constructing attribute from parameter response:",

tests/test_controller.py

@@ -1,8 +1,10 @@
+import enum
+
 import pytest
 
-from fastcs.attributes import AttrR
+from fastcs.attributes import AttrR, AttrRW
 from fastcs.controller import Controller, ControllerVector
-from fastcs.datatypes import Float, Int
+from fastcs.datatypes import Enum, Float, Int
 
 
 def test_controller_nesting():
@@ -144,3 +146,39 @@ def test_controller_vector_iter():
 
     for index, child in controller_vector.items():
         assert sub_controllers[index] == child
+
+
+def test_controller_introspection_hint_validation():
+    class HintedController(Controller):
+        read_write_int: AttrRW[int]
+
+    controller = HintedController()
+
+    with pytest.raises(RuntimeError, match="does not match defined datatype"):
+        controller.add_attribute("read_write_int", AttrRW(Float()))
+
+    with pytest.raises(RuntimeError, match="does not match defined access mode"):
+        controller.add_attribute("read_write_int", AttrR(Int()))
+
+    with pytest.raises(RuntimeError, match="failed to introspect hinted attribute"):
+        controller.validate_hinted_attributes()
+
+    controller.add_attribute("read_write_int", AttrRW(Int()))
+
+
+def test_controller_introspection_hint_validation_enum():
+    class GoodEnum(enum.IntEnum):
+        VAL = 0
+
+    class BadEnum(enum.IntEnum):
+        VAL = 0
+
+    class HintedController(Controller):
+        enum: AttrRW[GoodEnum]
+
+    controller = HintedController()
+
+    with pytest.raises(RuntimeError, match="does not match defined datatype"):
+        controller.add_attribute("enum", AttrRW(Enum(BadEnum)))
+
+    controller.add_attribute("enum", AttrRW(Enum(GoodEnum)))