Merge pull request #281 from DiamondLightSource/type-hint-validation

Improve Attribute hint validation and add Controller hint validation

Author: GDYendell

src/fastcs/attributes/__init__.py

@@ -6,3 +6,4 @@
 from .attribute_io import AttributeIO as AttributeIO
 from .attribute_io_ref import AttributeIORef as AttributeIORef
 from .attribute_io_ref import AttributeIORefT as AttributeIORefT
+from .hinted_attribute import HintedAttribute as HintedAttribute

src/fastcs/attributes/attribute.py

@@ -2,7 +2,7 @@
 from typing import Generic
 
 from fastcs.attributes.attribute_io_ref import AttributeIORefT
-from fastcs.datatypes import DATATYPE_DTYPES, DataType, DType_T
+from fastcs.datatypes import DataType, DType, DType_T
 from fastcs.logging import bind_logger
 from fastcs.tracer import Tracer
 
@@ -24,9 +24,8 @@ def __init__(
     ) -> None:
         super().__init__()
 
-        assert issubclass(datatype.dtype, DATATYPE_DTYPES), (
-            f"Attr type must be one of {DATATYPE_DTYPES}, "
-            "received type {datatype.dtype}"
+        assert issubclass(datatype.dtype, DType), (
+            f"Attr type must be one of {DType}, received type {datatype.dtype}"
         )
         self._io_ref = io_ref
         self._datatype: DataType[DType_T] = datatype

src/fastcs/attributes/hinted_attribute.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass
+
+from fastcs.attributes.attribute import Attribute
+from fastcs.datatypes import DType
+
+
+@dataclass(kw_only=True)
+class HintedAttribute:
+    """An `Attribute` type hint found on a `Controller` class
+
+    e.g. ``attr: AttrR[int]``
+
+    """
+
+    attr_type: type[Attribute]
+    """The type of the `Attribute` in the type hint - e.g. `AttrR`"""
+    dtype: type[DType] | None
+    """The dtype of the `Attribute` in the type hint, if any - e.g. `int`"""

src/fastcs/controllers/base_controller.py

@@ -5,10 +5,20 @@
 from copy import deepcopy
 from typing import _GenericAlias, get_args, get_origin, get_type_hints  # type: ignore
 
-from fastcs.attributes import Attribute, AttributeIO, AttributeIORefT, AttrR, AttrW
-from fastcs.datatypes import DataType, DType_T
+from fastcs.attributes import (
+    Attribute,
+    AttributeIO,
+    AttributeIORefT,
+    AttrR,
+    AttrW,
+    HintedAttribute,
+)
+from fastcs.datatypes import DType_T
+from fastcs.logging import bind_logger
 from fastcs.tracer import Tracer
 
+logger = bind_logger(logger_name=__name__)
+
 
 class BaseController(Tracer):
     """Base class for controllers
@@ -44,29 +54,43 @@ def __init__(
         # Internal state that should not be accessed directly by base classes
         self.__attributes: dict[str, Attribute] = {}
         self.__sub_controllers: dict[str, BaseController] = {}
-        self.__hinted_attributes = self._parse_attribute_type_hints()
+
+        self.__hinted_attributes: dict[str, HintedAttribute] = {}
+        self.__hinted_sub_controllers: dict[str, type[BaseController]] = {}
+        self._find_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 = {}
+    def _find_type_hints(self):
+        """Find `Attribute` and `Controller` type hints for introspection validation"""
         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])
+            if isinstance(hint, _GenericAlias):  # e.g. AttrR[int]
+                args = get_args(hint)
+                hint = get_origin(hint)
+            else:
+                args = None
+
+            if isinstance(hint, type) and issubclass(hint, Attribute):
+                if args is None:
+                    dtype = None
+                else:
+                    if len(args) == 2:
+                        dtype = args[0]
+                    else:
+                        raise TypeError(
+                            f"Invalid type hint for attribute {name}: {hint}"
+                        )
+
+                self.__hinted_attributes[name] = HintedAttribute(
+                    attr_type=hint, dtype=dtype
+                )
 
-        return hinted_attributes
+            elif isinstance(hint, type) and issubclass(hint, BaseController):
+                self.__hinted_sub_controllers[name] = hint
 
     def _bind_attrs(self) -> None:
         """Search for Attributes and Methods to bind them to this instance.
@@ -132,21 +156,51 @@ async def initialise(self):
 
     def post_initialise(self):
         """Hook to call after all attributes added, before serving the application"""
-        self._validate_hinted_attributes()
+        self._validate_type_hints()
         self._connect_attribute_ios()
 
-    def _validate_hinted_attributes(self):
-        """Validate ``Attribute`` type-hints were introspected during initialisation"""
+    def _validate_type_hints(self):
+        """Validate all `Attribute` and `Controller` type-hints were introspected"""
         for name in self.__hinted_attributes:
-            attr = getattr(self, name, None)
-            if attr is None or not isinstance(attr, Attribute):
-                raise RuntimeError(
-                    f"Controller `{self.__class__.__name__}` failed to introspect "
-                    f"hinted attribute `{name}` during initialisation"
-                )
+            self._validate_hinted_attribute(name)
+
+        for name in self.__hinted_sub_controllers:
+            self._validate_hinted_controller(name)
 
         for subcontroller in self.sub_controllers.values():
-            subcontroller._validate_hinted_attributes()  # noqa: SLF001
+            subcontroller._validate_type_hints()  # noqa: SLF001
+
+    def _validate_hinted_attribute(self, name: str):
+        """Check that an `Attribute` with the given name exists on the controller"""
+        attr = getattr(self, name, None)
+        if attr is None or not isinstance(attr, Attribute):
+            raise RuntimeError(
+                f"Controller `{self.__class__.__name__}` failed to introspect "
+                f"hinted attribute `{name}` during initialisation"
+            )
+        else:
+            logger.debug(
+                "Validated hinted attribute",
+                name=name,
+                controller=self,
+                attribute=attr,
+            )
+
+    def _validate_hinted_controller(self, name: str):
+        """Check that a sub controller with the given name exists on the controller"""
+        controller = getattr(self, name, None)
+        if controller is None or not isinstance(controller, BaseController):
+            raise RuntimeError(
+                f"Controller `{self.__class__.__name__}` failed to introspect "
+                f"hinted controller `{name}` during initialisation"
+            )
+        else:
+            logger.debug(
+                "Validated hinted sub controller",
+                name=name,
+                controller=self,
+                sub_controller=controller,
+            )
 
     def _connect_attribute_ios(self) -> None:
         """Connect ``Attribute`` callbacks to ``AttributeIO``s"""
@@ -191,18 +245,18 @@ def add_attribute(self, name, attr: Attribute):
                 f"{self.__attributes[name]}"
             )
         elif name in self.__hinted_attributes:
-            attr_class, attr_dtype = self.__hinted_attributes[name]
-            if not isinstance(attr, attr_class):
+            hint = self.__hinted_attributes[name]
+            if not isinstance(attr, hint.attr_type):
                 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__}'."
+                    f"Expected '{hint.attr_type.__name__}' got '{type(attr).__name__}'."
                 )
-            if attr_dtype is not None and attr_dtype != attr.datatype.dtype:
+            if hint.dtype is not None and hint.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"Expected '{hint.dtype.__name__}', "
                     f"got '{attr.datatype.dtype.__name__}'."
                 )
         elif name in self.__sub_controllers.keys():
@@ -228,6 +282,15 @@ def add_sub_controller(self, name: str, sub_controller: BaseController):
                 f"Controller {self} has existing sub controller {name}: "
                 f"{self.__sub_controllers[name]}"
             )
+        elif name in self.__hinted_sub_controllers:
+            hint = self.__hinted_sub_controllers[name]
+            if not isinstance(sub_controller, hint):
+                raise RuntimeError(
+                    f"Controller '{self.__class__.__name__}' introspection of "
+                    f"hinted sub controller '{name}' does not match defined type. "
+                    f"Expected '{hint.__name__}' got "
+                    f"'{sub_controller.__class__.__name__}'."
+                )
         elif name in self.__attributes:
             raise ValueError(
                 f"Cannot add sub controller {sub_controller}. "

src/fastcs/datatypes/__init__.py

@@ -1,7 +1,7 @@
 from ._util import numpy_to_fastcs_datatype as numpy_to_fastcs_datatype
 from .bool import Bool as Bool
-from .datatype import DATATYPE_DTYPES as DATATYPE_DTYPES
 from .datatype import DataType as DataType
+from .datatype import DType as DType
 from .datatype import DType_T as DType_T
 from .enum import Enum as Enum
 from .float import Float as Float

src/fastcs/datatypes/datatype.py

@@ -4,7 +4,15 @@
 from typing import Any, Generic, TypeVar
 
 import numpy as np
-from numpy.typing import DTypeLike
+
+DType = (
+    int  # Int
+    | float  # Float
+    | bool  # Bool
+    | str  # String
+    | enum.Enum  # Enum
+    | np.ndarray  # Waveform / Table
+)
 
 DType_T = TypeVar(
     "DType_T",
@@ -13,13 +21,10 @@
     bool,  # Bool
     str,  # String
     enum.Enum,  # Enum
-    np.ndarray,  # Waveform
-    list[tuple[str, DTypeLike]],  # Table
+    np.ndarray,  # Waveform / Table
 )
 """A builtin (or numpy) type supported by a corresponding FastCS Attribute DataType"""
 
-DATATYPE_DTYPES: tuple[type] = DType_T.__constraints__  # type: ignore
-
 
 @dataclass(frozen=True)
 class DataType(Generic[DType_T]):

tests/test_controllers.py

@@ -148,7 +148,7 @@ def test_controller_vector_iter():
         assert sub_controllers[index] == child
 
 
-def test_controller_introspection_hint_validation():
+def test_attribute_hint_validation():
     class HintedController(Controller):
         read_write_int: AttrRW[int]
 
@@ -162,15 +162,15 @@ class HintedController(Controller):
 
     with pytest.raises(RuntimeError, match="failed to introspect hinted attribute"):
         controller.read_write_int = 5  # type: ignore
-        controller._validate_hinted_attributes()
+        controller._validate_type_hints()
 
     with pytest.raises(RuntimeError, match="failed to introspect hinted attribute"):
-        controller._validate_hinted_attributes()
+        controller._validate_type_hints()
 
     controller.add_attribute("read_write_int", AttrRW(Int()))
 
 
-def test_controller_introspection_hint_validation_enum():
+def test_enum_attribute_hint_validation():
     class GoodEnum(enum.IntEnum):
         VAL = 0
 
@@ -186,3 +186,20 @@ class HintedController(Controller):
         controller.add_attribute("enum", AttrRW(Enum(BadEnum)))
 
     controller.add_attribute("enum", AttrRW(Enum(GoodEnum)))
+
+
+@pytest.mark.asyncio
+async def test_sub_controller_hint_validation():
+    class HintedController(Controller):
+        child: SomeSubController
+
+    controller = HintedController()
+
+    with pytest.raises(RuntimeError, match="failed to introspect hinted controller"):
+        controller._validate_type_hints()
+
+    with pytest.raises(RuntimeError, match="does not match defined type"):
+        controller.add_sub_controller("child", Controller())
+
+    controller.add_sub_controller("child", SomeSubController())
+    controller._validate_type_hints()