From 082158c51f25a5cd7d4a114d5e704d2c4a1529ba Mon Sep 17 00:00:00 2001 From: Gary Yendell Date: Fri, 5 Dec 2025 16:30:18 +0000 Subject: [PATCH 1/3] Add DType union to replace DATATYPE_DTYPES In future, this should be used in place of DType_T in non-generic contexts and DType_T should be bound to DType, but attempting this revealed some existing typing problems that were difficult to resolve. Remove incorrect Table dtype definition. It is actually just ndarray like Waveform. --- src/fastcs/attributes/attribute.py | 7 +++---- src/fastcs/datatypes/__init__.py | 2 +- src/fastcs/datatypes/datatype.py | 15 ++++++++++----- 3 files changed, 14 insertions(+), 10 deletions(-) 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/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]): From 55641b8b2da75719bf722d36d95e60692ed58bed Mon Sep 17 00:00:00 2001 From: Gary Yendell Date: Wed, 3 Dec 2025 16:52:02 +0000 Subject: [PATCH 2/3] Improve Attribute validation Add HintedAttribute dataclass Allow for validation of Attribute type hints without a dtype --- src/fastcs/attributes/__init__.py | 1 + src/fastcs/attributes/hinted_attribute.py | 18 +++++ src/fastcs/controllers/base_controller.py | 87 ++++++++++++++++------- 3 files changed, 80 insertions(+), 26 deletions(-) create mode 100644 src/fastcs/attributes/hinted_attribute.py 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/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..1d1ebb0aa 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,7 @@ 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 = self._validate_attribute_type_hints() self._bind_attrs() @@ -52,19 +62,33 @@ 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 = {} - for name, hint in get_type_hints(type(self)).items(): - if not isinstance(hint, _GenericAlias): # e.g. AttrR[int] - continue + def _validate_attribute_type_hints(self) -> dict[str, HintedAttribute]: + """Validate `Attribute` type hints for introspection - origin = get_origin(hint) - if not isinstance(origin, type) or not issubclass(origin, Attribute): - continue + Returns: + A dictionary of `HintedAttribute` from type hints on this controller class - hinted_attributes[name] = (origin, get_args(hint)[0]) + """ + hinted_attributes = {} + for name, hint in get_type_hints(type(self)).items(): + 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}" + ) + + hinted_attributes[name] = HintedAttribute(attr_type=hint, dtype=dtype) return hinted_attributes @@ -136,18 +160,29 @@ def post_initialise(self): self._connect_attribute_ios() def _validate_hinted_attributes(self): - """Validate ``Attribute`` type-hints were introspected during initialisation""" + """Validate all `Attribute` 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 subcontroller in self.sub_controllers.values(): subcontroller._validate_hinted_attributes() # 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 _connect_attribute_ios(self) -> None: """Connect ``Attribute`` callbacks to ``AttributeIO``s""" for attr in self.__attributes.values(): @@ -191,18 +226,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(): From 2499e6d2f3d893b8a2babc4a5ccb78ae083c5231 Mon Sep 17 00:00:00 2001 From: Gary Yendell Date: Wed, 3 Dec 2025 17:21:42 +0000 Subject: [PATCH 3/3] Add validation of hinted sub controllers --- src/fastcs/controllers/base_controller.py | 58 +++++++++++++++++------ tests/test_controllers.py | 25 ++++++++-- 2 files changed, 64 insertions(+), 19 deletions(-) diff --git a/src/fastcs/controllers/base_controller.py b/src/fastcs/controllers/base_controller.py index 1d1ebb0aa..ff7507df4 100755 --- a/src/fastcs/controllers/base_controller.py +++ b/src/fastcs/controllers/base_controller.py @@ -54,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._validate_attribute_type_hints() + + self.__hinted_attributes: dict[str, HintedAttribute] = {} + self.__hinted_sub_controllers: dict[str, type[BaseController]] = {} + self._find_type_hints() self._bind_attrs() @@ -62,14 +65,8 @@ def __init__( self._attribute_ref_io_map = {io.ref_type: io for io in ios} self._validate_io(ios) - def _validate_attribute_type_hints(self) -> dict[str, HintedAttribute]: - """Validate `Attribute` type hints for introspection - - Returns: - A dictionary of `HintedAttribute` from type hints on this controller class - - """ - 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 isinstance(hint, _GenericAlias): # e.g. AttrR[int] args = get_args(hint) @@ -88,9 +85,12 @@ def _validate_attribute_type_hints(self) -> dict[str, HintedAttribute]: f"Invalid type hint for attribute {name}: {hint}" ) - hinted_attributes[name] = HintedAttribute(attr_type=hint, dtype=dtype) + 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. @@ -156,16 +156,19 @@ 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 all `Attribute` type-hints were introspected""" + def _validate_type_hints(self): + """Validate all `Attribute` and `Controller` type-hints were introspected""" for name in self.__hinted_attributes: 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""" @@ -183,6 +186,22 @@ def _validate_hinted_attribute(self, name: str): 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""" for attr in self.__attributes.values(): @@ -263,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/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()