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
2 changes: 2 additions & 0 deletions src/fastcs/backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from .controller import BaseController, Controller
from .controller_api import ControllerAPI
from .exceptions import FastCSException
from .util import validate_hinted_attributes


class Backend:
Expand All @@ -28,6 +29,7 @@ def __init__(
# Initialise controller and then build its APIs
loop.run_until_complete(controller.initialise())
loop.run_until_complete(controller.attribute_initialise())
validate_hinted_attributes(controller)
self.controller_api = build_controller_api(controller)
self._link_process_tasks()

Expand Down
35 changes: 35 additions & 0 deletions src/fastcs/util.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
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


Expand All @@ -26,3 +29,35 @@ def numpy_to_fastcs_datatype(np_type) -> DataType:
return Bool()
else:
return String()


def validate_hinted_attributes(controller: BaseController):
"""Validates that type-hinted attributes exist in the controller, and are accessible
via the dot accessor, from the attributes dictionary and with the right datatype.
"""
hints = get_type_hints(type(controller))
alias_hints = {k: v for k, v in hints.items() if isinstance(v, _GenericAlias)}
for name, hint in alias_hints.items():
attr_class = get_origin(hint)
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 type(attr) is not attr_class:
raise RuntimeError(
f"Controller '{controller.__class__.__name__}' introspection of hinted "
f"attribute '{name}' does not match defined access mode. "
f"Expected '{attr_class.__name__}', got '{type(attr).__name__}'."
)
(attr_dtype,) = get_args(hint)
if attr.datatype.dtype != attr_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__}'."
)
70 changes: 69 additions & 1 deletion tests/test_util.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,15 @@
import asyncio
import enum

import numpy as np
import pytest
from pvi.device import SignalR
from pydantic import ValidationError

from fastcs.datatypes import Bool, Float, Int, String
from fastcs.attributes import AttrR, AttrRW
from fastcs.backend import Backend
from fastcs.controller import Controller
from fastcs.datatypes import Bool, Enum, Float, Int, String
from fastcs.util import numpy_to_fastcs_datatype, snake_to_pascal


Expand Down Expand Up @@ -56,3 +62,65 @@ def test_pvi_validation_error():
)
def test_numpy_to_fastcs_datatype(numpy_type, fastcs_datatype):
assert fastcs_datatype == numpy_to_fastcs_datatype(numpy_type)


def test_hinted_attributes_verified():
loop = asyncio.get_event_loop()

class ControllerWithWrongType(Controller):
hinted_wrong_type: AttrR[int]

async def initialise(self):
self.hinted_wrong_type = AttrR(Float()) # type: ignore
self.attributes["hinted_wrong_type"] = self.hinted_wrong_type

with pytest.raises(RuntimeError) as excinfo:
Backend(ControllerWithWrongType(), loop)
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]

with pytest.raises(RuntimeError) as excinfo:
Backend(ControllerWithMissingAttr(), loop)
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

with pytest.raises(RuntimeError) as excinfo:
Backend(ControllerAttrWrongAccessMode(), loop)
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))

with pytest.raises(RuntimeError) as excinfo:
Backend(ControllerWrongEnumClass(), loop)
assert str(excinfo.value) == (
"Controller 'ControllerWrongEnumClass' introspection of hinted attribute "
"'hinted_enum' does not match defined datatype. "
"Expected 'MyEnum', got 'MyEnum2'."
)
Loading