Integrate hinted attribute validation into controller

GDYendell · GDYendell · commit 861a828ab0e4 · 2025-11-14T12:24:08.000Z
Check datatype and access mode in add_attribute, check for
unintrospected type hints after initialise
diff --git a/src/fastcs/control_system.py b/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)
diff --git a/src/fastcs/controller.py b/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():
diff --git a/src/fastcs/util.py b/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__}'."
-            )
diff --git a/tests/test_attribute.py b/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:",
diff --git a/tests/test_controller.py b/tests/test_controller.py
@@ -1,6 +1,6 @@
 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
 
@@ -144,3 +144,21 @@ 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_int: AttrRW[int]
+
+    controller = HintedController()
+
+    with pytest.raises(RuntimeError, match="does not match defined datatype"):
+        controller.add_attribute("read_int", AttrRW(Float()))
+
+    with pytest.raises(RuntimeError, match="does not match defined access mode"):
+        controller.add_attribute("read_int", AttrR(Int()))
+
+    with pytest.raises(RuntimeError, match="failed to introspect hinted attribute"):
+        controller.validate_hinted_attributes()
+
+    controller.add_attribute("read_int", AttrRW(Int()))
diff --git a/tests/test_util.py b/tests/test_util.py
@@ -1,20 +1,10 @@
-import asyncio
-import enum
-
 import numpy as np
 import pytest
 from pvi.device import SignalR
 from pydantic import ValidationError
 
-from fastcs.attributes import Attribute, AttrR, AttrRW
-from fastcs.controller import Controller
-from fastcs.datatypes import Bool, Enum, Float, Int, String
-from fastcs.launch import FastCS
-from fastcs.util import (
-    numpy_to_fastcs_datatype,
-    snake_to_pascal,
-    validate_hinted_attributes,
-)
+from fastcs.datatypes import Bool, Float, Int, String
+from fastcs.util import numpy_to_fastcs_datatype, snake_to_pascal
 
 
 def test_snake_to_pascal():
@@ -66,129 +56,3 @@ def test_pvi_validation_error():
 )
 def test_numpy_to_fastcs_datatype(numpy_type, fastcs_datatype):
     assert fastcs_datatype == numpy_to_fastcs_datatype(numpy_type)
-
-
-@pytest.mark.asyncio
-async def test_hinted_attributes_verified():
-    class ControllerWithWrongType(Controller):
-        hinted_wrong_type: AttrR[int]
-
-        async def initialise(self):
-            self.hinted_wrong_type = AttrR(Float())  # type: ignore
-
-    controller = ControllerWithWrongType()
-    await controller.initialise()
-    with pytest.raises(RuntimeError) as excinfo:
-        validate_hinted_attributes(controller)
-
-    assert str(excinfo.value) == (
-        "Controller 'ControllerWithWrongType' introspection of hinted attribute "
-        "'hinted_wrong_type' does not match defined datatype. "
-        "Expected 'int', got 'float'."
-    )
-
-    class ControllerWithMissingAttr(Controller):
-        hinted_int_missing: AttrR[int]
-
-    controller = ControllerWithMissingAttr()
-    await controller.initialise()
-    with pytest.raises(RuntimeError) as excinfo:
-        validate_hinted_attributes(controller)
-
-    assert str(excinfo.value) == (
-        "Controller `ControllerWithMissingAttr` failed to introspect hinted attribute "
-        "`hinted_int_missing` during initialisation"
-    )
-
-    class ControllerAttrWrongAccessMode(Controller):
-        hinted: AttrR[int]
-
-        async def initialise(self):
-            self.hinted = AttrRW(Int())
-            self.attributes["hinted"] = self.hinted
-
-    controller = ControllerAttrWrongAccessMode()
-    await controller.initialise()
-    with pytest.raises(RuntimeError) as excinfo:
-        validate_hinted_attributes(controller)
-
-    assert str(excinfo.value) == (
-        "Controller 'ControllerAttrWrongAccessMode' introspection of hinted attribute "
-        "'hinted' does not match defined access mode. Expected 'AttrR', got 'AttrRW'."
-    )
-
-    class MyEnum(enum.Enum):
-        A = 0
-        B = 1
-
-    class MyEnum2(enum.Enum):
-        A = 2
-        B = 3
-
-    class ControllerWrongEnumClass(Controller):
-        hinted_enum: AttrRW[MyEnum] = AttrRW(Enum(MyEnum2))
-
-    controller = ControllerWrongEnumClass()
-    await controller.initialise()
-    with pytest.raises(RuntimeError) as excinfo:
-        validate_hinted_attributes(controller)
-
-    assert str(excinfo.value) == (
-        "Controller 'ControllerWrongEnumClass' introspection of hinted attribute "
-        "'hinted_enum' does not match defined datatype. "
-        "Expected 'MyEnum', got 'MyEnum2'."
-    )
-
-
-def test_hinted_attributes_verified_on_subcontrollers():
-    loop = asyncio.get_event_loop()
-
-    class ControllerWithWrongType(Controller):
-        hinted_missing: AttrR[int]
-
-        async def connect(self):
-            return
-
-    class TopController(Controller):
-        async def initialise(self):  # why does this not get called?
-            subcontroller = ControllerWithWrongType()
-            self.add_sub_controller("MySubController", subcontroller)
-
-    fastcs = FastCS(TopController(), [], loop)
-    with pytest.raises(RuntimeError, match="failed to introspect hinted attribute"):
-        fastcs.run()
-
-
-def test_hinted_attribute_access_mode_verified():
-    # test verification works with non-GenericAlias type hints
-    loop = asyncio.get_event_loop()
-
-    class ControllerAttrWrongAccessMode(Controller):
-        read_attr: AttrR
-
-        async def initialise(self):
-            self.read_attr = AttrRW(Int())
-
-    fastcs = FastCS(ControllerAttrWrongAccessMode(), [], loop)
-    with pytest.raises(RuntimeError, match="does not match defined access mode"):
-        fastcs.run()
-
-
-@pytest.mark.asyncio
-async def test_hinted_attributes_with_unspecified_access_mode():
-    class ControllerUnspecifiedAccessMode(Controller):
-        unspecified_access_mode: Attribute
-
-        async def initialise(self):
-            self.unspecified_access_mode = AttrRW(Int())
-
-    controller = ControllerUnspecifiedAccessMode()
-    await controller.initialise()
-    # no assertion thrown
-    with pytest.raises(
-        RuntimeError,
-        match=(
-            "does not match defined access mode. Expected 'Attribute', got 'AttrRW'"
-        ),
-    ):
-        validate_hinted_attributes(controller)
