diff --git a/src/fastcs/attributes/__init__.py b/src/fastcs/attributes/__init__.py index a65c5e162..b25e66f72 100644 --- a/src/fastcs/attributes/__init__.py +++ b/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 diff --git a/src/fastcs/attributes/attribute.py b/src/fastcs/attributes/attribute.py index 6668cff1b..546dab165 100644 --- a/src/fastcs/attributes/attribute.py +++ b/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 diff --git a/src/fastcs/attributes/hinted_attribute.py b/src/fastcs/attributes/hinted_attribute.py new file mode 100644 index 000000000..d58fadd46 --- /dev/null +++ b/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`""" diff --git a/src/fastcs/controllers/base_controller.py b/src/fastcs/controllers/base_controller.py index 6ea709672..ff7507df4 100755 --- a/src/fastcs/controllers/base_controller.py +++ b/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,7 +54,10 @@ 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() @@ -52,21 +65,32 @@ def __init__( 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}. " diff --git a/src/fastcs/datatypes/__init__.py b/src/fastcs/datatypes/__init__.py index 7952c6c90..fc108c9d5 100644 --- a/src/fastcs/datatypes/__init__.py +++ b/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 diff --git a/src/fastcs/datatypes/datatype.py b/src/fastcs/datatypes/datatype.py index d8ac378c7..1559a5c08 100644 --- a/src/fastcs/datatypes/datatype.py +++ b/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]): diff --git a/tests/test_controllers.py b/tests/test_controllers.py index b5d21d029..bc084324e 100644 --- a/tests/test_controllers.py +++ b/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()