Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/fastcs/attributes/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
7 changes: 3 additions & 4 deletions src/fastcs/attributes/attribute.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand Down
18 changes: 18 additions & 0 deletions src/fastcs/attributes/hinted_attribute.py
Original file line number Diff line number Diff line change
@@ -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`"""
125 changes: 94 additions & 31 deletions src/fastcs/controllers/base_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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"""
Expand Down Expand Up @@ -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():
Expand All @@ -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}. "
Expand Down
2 changes: 1 addition & 1 deletion src/fastcs/datatypes/__init__.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down
15 changes: 10 additions & 5 deletions src/fastcs/datatypes/datatype.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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]):
Expand Down
25 changes: 21 additions & 4 deletions tests/test_controllers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]

Expand All @@ -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

Expand All @@ -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()