diff --git a/src/fastcs/attributes/attr_r.py b/src/fastcs/attributes/attr_r.py index 28d66c07b..ccf044f22 100644 --- a/src/fastcs/attributes/attr_r.py +++ b/src/fastcs/attributes/attr_r.py @@ -46,6 +46,13 @@ def get(self) -> DType_T: """Get the cached value of the attribute.""" return self._value + def update_datatype(self, datatype: DataType[DType_T]) -> None: + value = self._value + if not isinstance(value, datatype.dtype): # e.g. a member of a replaced enum + value = datatype.validate(value) + super().update_datatype(datatype) + self._value = value + @property def access_mode(self) -> AttributeAccessMode: return "r" diff --git a/src/fastcs/controllers/base_controller.py b/src/fastcs/controllers/base_controller.py index 8b29ed6d2..a5989d95f 100755 --- a/src/fastcs/controllers/base_controller.py +++ b/src/fastcs/controllers/base_controller.py @@ -1,5 +1,6 @@ from __future__ import annotations +import enum from collections import Counter from collections.abc import Sequence from copy import deepcopy @@ -13,6 +14,7 @@ from fastcs.attributes import AnyAttributeIO, Attribute, AttrR, AttrW, HintedAttribute from fastcs.controllers.controller_api import ControllerAPI +from fastcs.datatypes import Enum, EnumTypes, StrictEnum, validate_enum_choices from fastcs.logging import logger from fastcs.methods import Command, Method, Scan, UnboundCommand, UnboundScan from fastcs.tracer import Tracer @@ -323,7 +325,13 @@ def add_attribute(self, name, attr: Attribute): f"hinted attribute '{name}' does not match defined access mode. " f"Expected '{hint.attr_type.__name__}' got '{type(attr).__name__}'." ) - if hint.dtype is not None and hint.dtype != attr.datatype.dtype: + if ( + hint.dtype is not None + and isinstance(attr.datatype, Enum) + and issubclass(hint.dtype, EnumTypes) + ): + self._validate_hinted_enum(name, hint.dtype, attr) + elif 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. " @@ -336,6 +344,23 @@ def add_attribute(self, name, attr: Attribute): self.__attributes[name] = attr super().__setattr__(name, attr) + def _validate_hinted_enum( + self, name: str, hinted: type[enum.Enum], attr: Attribute + ) -> None: + """Validate introspected enum choices and adopt the hinted enum""" + introspected = attr.datatype.dtype + try: + validate_enum_choices(hinted, introspected) + except TypeError as exc: + raise RuntimeError( + f"Controller '{self.__class__.__name__}' introspection of " + f"hinted attribute '{name}' does not match defined enum choices. {exc}" + ) from None + + # Superset and Subset serve only the choices the device has + allowed = None if issubclass(hinted, StrictEnum) else introspected + attr.update_datatype(Enum(hinted, allowed=allowed)) + @property def attributes(self) -> dict[str, Attribute]: return self.__attributes diff --git a/src/fastcs/datatypes/__init__.py b/src/fastcs/datatypes/__init__.py index fc108c9d5..c6c1d5377 100644 --- a/src/fastcs/datatypes/__init__.py +++ b/src/fastcs/datatypes/__init__.py @@ -4,6 +4,11 @@ from .datatype import DType as DType from .datatype import DType_T as DType_T from .enum import Enum as Enum +from .enum import EnumTypes as EnumTypes +from .enum import StrictEnum as StrictEnum +from .enum import SubsetEnum as SubsetEnum +from .enum import SupersetEnum as SupersetEnum +from .enum import validate_enum_choices as validate_enum_choices from .float import Float as Float from .int import Int as Int from .string import String as String diff --git a/src/fastcs/datatypes/enum.py b/src/fastcs/datatypes/enum.py index e490f5c76..8fe47be95 100644 --- a/src/fastcs/datatypes/enum.py +++ b/src/fastcs/datatypes/enum.py @@ -1,7 +1,7 @@ import enum from dataclasses import dataclass from functools import cached_property -from typing import Generic, TypeVar +from typing import Any, Generic, TypeVar from fastcs.datatypes.datatype import DataType @@ -12,6 +12,9 @@ @dataclass(frozen=True) class Enum(Generic[Enum_T], DataType[Enum_T]): enum_cls: type[Enum_T] + allowed: type[enum.Enum] | None = None + """Introspected device enum restricting (for SupersetEnum) + or extending (for SubsetEnum) the choices of ``enum_cls``, mapped by value""" def __post_init__(self): if not issubclass(self.enum_cls, enum.Enum): @@ -22,11 +25,29 @@ def index_of(self, value: Enum_T) -> int: @cached_property def members(self) -> list[Enum_T]: - return list(self.enum_cls) + if self.allowed is None: + return list(self.enum_cls) + # SubsetEnum returns the raw value for device choices it doesn't define + return [self.enum_cls(device_member.value) for device_member in self.allowed] @cached_property def names(self) -> list[str]: - return [member.name for member in self.members] + if self.allowed is None: + return [member.name for member in self.members] + return [ + choice.name if isinstance(choice, self.enum_cls) else device_member.name + for choice, device_member in zip(self.members, self.allowed, strict=True) + ] + + def validate(self, value: Any) -> Enum_T: + # Map members of another enum, e.g. an introspected one, by value + if isinstance(value, enum.Enum) and not isinstance(value, self.enum_cls): + value = value.value + + value = super().validate(value) + if value not in self.members: + raise ValueError(f"{value!r} is not one of the choices {self.names}") + return value @property def dtype(self) -> type[Enum_T]: @@ -35,3 +56,77 @@ def dtype(self) -> type[Enum_T]: @property def initial_value(self) -> Enum_T: return self.members[0] + + +class StrictEnum(enum.Enum): + """All members should exist on the device, and there will be no extras.""" + + +class _SubsetEnumMeta(enum.EnumMeta): + _member_type_: type + """The mixed in value type, e.g. ``str``, set by ``EnumMeta``""" + + def __new__(mcls, *args, **kwargs): + cls = super().__new__(mcls, *args, **kwargs) + if cls._member_map_ and cls._member_type_ is object: + raise TypeError( + f"{cls.__name__} needs a value type, e.g. (str, SubsetEnum)" + ) + return cls + + def __call__(cls, value, names=None, *args, **kwargs): # type: ignore + """Return the member for ``value``, or ``value`` itself if it is of the + member type but not one of the members.""" + if names is not None: # functional API + return super().__call__(value, names, *args, **kwargs) + try: + return super().__call__(value) + except ValueError: + if isinstance(value, cls._member_type_): + return value + raise + + +class SubsetEnum(enum.Enum, metaclass=_SubsetEnumMeta): + """All members should exist on the device, but the device may have extras. + + Subclasses must mix in a value type, e.g. ``class Trigger(str, SubsetEnum)``, as + values for device choices outside the enum are kept as that raw type. + """ + + +class SupersetEnum(enum.Enum): + """Some members should exist on the device, and there should be no extras.""" + + +EnumTypes = (StrictEnum, SubsetEnum, SupersetEnum) + + +def validate_enum_choices( + hinted: type[enum.Enum], introspected: type[enum.Enum] +) -> None: + """Check introspected enum choices against a hinted enum, compared by value. + + Raises: + TypeError: If the choices are not compatible with the hinted enum type + + """ + choices = [member.value for member in hinted] + device_choices = [member.value for member in introspected] + error_msg = ( + f"Device has choices {device_choices}, but {hinted.__name__} requested " + f"{choices} " + ) + if issubclass(hinted, StrictEnum): + if set(choices) != set(device_choices): + raise TypeError(error_msg + "to be strictly equal to them.") + elif issubclass(hinted, SubsetEnum): + if not set(choices).issubset(device_choices): + raise TypeError(error_msg + "to be a subset of them.") + elif issubclass(hinted, SupersetEnum): + if not set(device_choices).issubset(choices): + raise TypeError(error_msg + "to be a superset of them.") + else: + raise TypeError( + f"{hinted.__name__} is not a StrictEnum, SubsetEnum, or SupersetEnum" + ) diff --git a/src/fastcs/transports/epics/ca/util.py b/src/fastcs/transports/epics/ca/util.py index 668039518..adcdfe38c 100644 --- a/src/fastcs/transports/epics/ca/util.py +++ b/src/fastcs/transports/epics/ca/util.py @@ -1,4 +1,3 @@ -import enum import re from collections.abc import Callable from dataclasses import asdict @@ -246,10 +245,8 @@ def cast_from_epics_type(datatype: DataType[DType_T], value: object) -> DType_T: return datatype.validate(datatype.members[value]) else: # enum backed by string record assert isinstance(value, str), "Got non-string value for long Enum" - # python typing can't narrow the nested generic enum_cls - assert issubclass(datatype.enum_cls, enum.Enum), "Invalid Enum.enum_cls" - enum_member = datatype.enum_cls[value] - return datatype.validate(enum_member) + # Map by position, as SubsetEnum extras are raw values, not members + return datatype.validate(datatype.members[datatype.names.index(value)]) case datatype if issubclass(type(datatype), EPICS_ALLOWED_DATATYPES): return datatype.validate(value) # type: ignore case _: @@ -263,7 +260,8 @@ def cast_to_epics_type(datatype: DataType[DType_T], value: DType_T) -> Any: if len(datatype.members) <= MBB_MAX_CHOICES: return datatype.index_of(datatype.validate(value)) else: # enum backed by string record - return datatype.validate(value).name + # Map by position, as SubsetEnum extras are raw values, not members + return datatype.names[datatype.index_of(datatype.validate(value))] case String() as string: if string.length is not None: return value[: string.length] diff --git a/src/fastcs/transports/tango/util.py b/src/fastcs/transports/tango/util.py index 9a82f264c..0d31363f8 100644 --- a/src/fastcs/transports/tango/util.py +++ b/src/fastcs/transports/tango/util.py @@ -2,7 +2,7 @@ from dataclasses import asdict from typing import Any -from tango import AttrDataFormat +from tango import AttrDataFormat, CmdArgType from fastcs.attributes import Attribute from fastcs.datatypes import ( @@ -100,6 +100,10 @@ def get_server_metadata_from_datatype(datatype: DataType[DType]) -> dict[str, st ) case Float(): arguments["format"] = f"%.{datatype.prec}" + case Enum(): + # Labels must match the served choices that indices refer to + dtype = CmdArgType.DevEnum + arguments["enum_labels"] = datatype.names arguments["dtype"] = dtype for argument, value in arguments.items(): diff --git a/tests/example_p4p_ioc.py b/tests/example_p4p_ioc.py index 95cc8e70b..378568834 100644 --- a/tests/example_p4p_ioc.py +++ b/tests/example_p4p_ioc.py @@ -10,6 +10,7 @@ from fastcs.launch import FastCS from fastcs.methods import command, scan from fastcs.transports.epics.pva import EpicsPVATransport +from tests.util import DeviceGain, DeviceMode, DeviceTrigger, Gain, Mode, Trigger @dataclass @@ -42,10 +43,22 @@ class ParentController(Controller): Table([("A", np.int32), ("B", "i"), ("C", "?"), ("D", np.float64)]), io_ref=SimpleAttributeIORef(), ) + mode: AttrRW[Mode] + gain: AttrRW[Gain] + trigger: AttrRW[Trigger] def __init__(self, description=None, ios=None): super().__init__(description, ios) + async def initialise(self): + # Enums as introspected from a device, validated against the hints above + for name, device_enum in ( + ("mode", DeviceMode), + ("gain", DeviceGain), + ("trigger", DeviceTrigger), + ): + self.add_attribute(name, AttrRW(Enum(device_enum))) + class ChildController(Controller): fail_on_next_e = True diff --git a/tests/example_softioc.py b/tests/example_softioc.py index 3688ff4fb..ffc370f00 100644 --- a/tests/example_softioc.py +++ b/tests/example_softioc.py @@ -1,21 +1,57 @@ import asyncio +import enum from pathlib import Path +from typing import cast from fastcs.attributes import AttrR, AttrRW, AttrW from fastcs.control_system import FastCS from fastcs.controllers import Controller, ControllerVector -from fastcs.datatypes import Int +from fastcs.datatypes import Enum, Int, SubsetEnum from fastcs.methods import command from fastcs.transports.epics.ca.transport import ( EpicsCAOptions, EpicsCATransport, EpicsGUIOptions, ) +from tests.util import DeviceGain, DeviceMode, DeviceTrigger, Gain, Mode, Trigger + + +class LongTrigger(str, SubsetEnum): + INTERNAL = "Internal" + EXTERNAL = "External" + + +# More than 16 choices, so CA serves it as a string record of names +DeviceLongTrigger = cast( + type[enum.Enum], + enum.Enum( + "DeviceLongTrigger", + { + "INTERNAL": "Internal", + "EXTERNAL": "External", + **{f"LINE{i}": f"Line{i}" for i in range(16)}, + }, + ), +) class ParentController(Controller): a: AttrR = AttrR(Int()) b: AttrRW = AttrRW(Int()) + mode: AttrRW[Mode] + gain: AttrRW[Gain] + trigger: AttrRW[Trigger] + long_trigger: AttrRW[LongTrigger] + + async def initialise(self): + # Enums as introspected from a device, validated against the hints above + for name, device_enum in ( + ("mode", DeviceMode), + ("gain", DeviceGain), + ("trigger", DeviceTrigger), + ("long_trigger", DeviceLongTrigger), + ): + self.add_attribute(name, AttrRW(Enum(device_enum))) class ChildController(Controller): diff --git a/tests/test_attributes.py b/tests/test_attributes.py index 41bd8547e..0b33fa977 100644 --- a/tests/test_attributes.py +++ b/tests/test_attributes.py @@ -1,4 +1,5 @@ import asyncio +import enum from dataclasses import dataclass from functools import partial from typing import Generic, TypeVar @@ -8,7 +9,8 @@ from fastcs.attributes import AttributeIO, AttributeIORef, AttrR, AttrRW, AttrW from fastcs.controllers import Controller -from fastcs.datatypes import Float, Int, String +from fastcs.datatypes import Enum, Float, Int, String +from tests.util import DeviceGain, Gain def test_attribute_access_mode(): @@ -50,6 +52,40 @@ def test_attr_r(): assert attr.get() == "" +def test_update_datatype_converts_enum_value(): + attr: AttrR = AttrR(Enum(DeviceGain)) + assert attr.get() is DeviceGain.LOW + + attr.update_datatype(Enum(Gain, allowed=DeviceGain)) + + # Converted by value to the new enum class + assert attr.get() is Gain.LOW + + +def test_update_datatype_failed_conversion_changes_nothing(mocker: MockerFixture): + attr: AttrR = AttrR(Enum(DeviceGain), initial_value=DeviceGain.HIGH) + original = attr.datatype + callback = mocker.Mock() + attr.add_update_datatype_callback(callback) + + # HIGH is not one of the new datatype's choices + restricted = enum.Enum("Restricted", {"LOW": "Low"}) + with pytest.raises(ValueError): + attr.update_datatype(Enum(Gain, allowed=restricted)) + + assert attr.datatype is original + assert attr.get() is DeviceGain.HIGH + callback.assert_not_called() + + +def test_update_datatype_same_type_keeps_value(): + # TODO: New metadata may result in stale _value, so possibly this should re-validate + # Metadata changes don't re-validate, so narrowing limits doesn't raise + attr = AttrR(Int(), initial_value=0) + attr.update_datatype(Int(min=5)) + assert attr.get() == 0 + + @pytest.mark.asyncio async def test_attr_update(mocker: MockerFixture): attr = AttrRW(Int()) diff --git a/tests/test_controllers.py b/tests/test_controllers.py index 5d5dfb4ed..1b4f969e2 100644 --- a/tests/test_controllers.py +++ b/tests/test_controllers.py @@ -7,6 +7,7 @@ from fastcs.controllers import Controller, ControllerVector from fastcs.datatypes import Enum, Float, Int from fastcs.methods import Command, Scan, command, scan +from tests.util import DeviceGain, DeviceMode, DeviceTrigger, Gain, Mode, Trigger def test_controller_nesting(): @@ -196,6 +197,53 @@ class HintedController(Controller): controller.add_attribute("enum", AttrRW(Enum(GoodEnum))) +@pytest.mark.parametrize( + "hinted, device_enum, allowed, initial", + [ + pytest.param(Mode, DeviceMode, False, Mode.OFF, id="strict"), + pytest.param(Gain, DeviceGain, True, Gain.LOW, id="superset"), + pytest.param(Trigger, DeviceTrigger, True, Trigger.INTERNAL, id="subset"), + ], +) +def test_enum_hint_adopts_hinted_enum(hinted, device_enum, allowed, initial): + class HintedController(Controller): + enum: AttrRW[hinted] # type: ignore + + controller = HintedController() + # Adds attribute using introspected enum + controller.add_attribute("enum", AttrRW(Enum(device_enum))) + + attr = controller.attributes["enum"] + assert isinstance(attr, AttrRW) + assert isinstance(attr.datatype, Enum) + # Hinted enum is used instead of introspected enum + assert attr.datatype.dtype is hinted + # Strict serves the whole hint, Superset and Subset only the device's choices + assert attr.datatype.allowed is (device_enum if allowed else None) + # The stored value is converted from the device enum to the hinted one + assert attr.get() is initial + + +@pytest.mark.parametrize( + "hinted, device_choices", + [ + pytest.param(Mode, {"OFF": "Off"}, id="strict-missing"), + pytest.param(Mode, {"OFF": "Off", "ON": "On", "X": "x"}, id="strict-extra"), + pytest.param(Gain, {"LOW": "Low", "TURBO": "Turbo"}, id="superset-extra"), + pytest.param(Trigger, {"INTERNAL": "Internal"}, id="subset-missing"), + ], +) +def test_enum_hint_rejects_incompatible_choices(hinted, device_choices): + class HintedController(Controller): + enum: AttrRW[hinted] # type: ignore + + controller = HintedController() + device_enum = enum.Enum("DeviceEnum", device_choices) + + with pytest.raises(RuntimeError, match="does not match defined enum choices"): + controller.add_attribute("enum", AttrRW(Enum(device_enum))) + + @pytest.mark.asyncio async def test_sub_controller_hint_validation(): class HintedController(Controller): diff --git a/tests/test_datatypes.py b/tests/test_datatypes.py index 221fa1de6..db353f01b 100644 --- a/tests/test_datatypes.py +++ b/tests/test_datatypes.py @@ -1,10 +1,22 @@ +import enum from enum import IntEnum import numpy as np import pytest -from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, Table, Waveform +from fastcs.datatypes import ( + Bool, + DataType, + Enum, + Float, + Int, + String, + SubsetEnum, + Table, + Waveform, +) from fastcs.datatypes._util import numpy_to_fastcs_datatype +from tests.util import DeviceGain, DeviceMode, DeviceTrigger, Gain, Mode, Trigger def test_base_validate(): @@ -129,3 +141,95 @@ def test_string_length(): with pytest.raises(ValueError): String(length=0) + + +@pytest.mark.parametrize( + "hinted_enum, introspected_enum, allowed_members, names", + [ + pytest.param(Mode, None, [Mode.OFF, Mode.ON], ["OFF", "ON"], id="strict"), + pytest.param( + Mode, + DeviceMode, + [Mode.OFF, Mode.ON], + ["OFF", "ON"], + id="strict-with-allowed", + ), + pytest.param( + Gain, DeviceGain, [Gain.LOW, Gain.HIGH], ["LOW", "HIGH"], id="superset" + ), + pytest.param( + Trigger, + DeviceTrigger, + [Trigger.INTERNAL, Trigger.EXTERNAL, "Software"], + ["INTERNAL", "EXTERNAL", "SOFTWARE"], + id="subset", + ), + ], +) +def test_enum_types_correctly_allow_options_based_on_introspected_info( + hinted_enum, introspected_enum, allowed_members, names +): + test_enum = Enum(hinted_enum, allowed=introspected_enum) + assert test_enum.members == allowed_members + # == alone can't tell str-mixin members from raw strings, so + # Check that members are of expected type + assert [type(m) for m in test_enum.members] == [type(m) for m in allowed_members] + # Check identical names + assert test_enum.names == names + + +def test_enum_validate_allowed(): + subset_enum = Enum(Trigger, allowed=DeviceTrigger) + superset_enum = Enum(Gain, allowed=DeviceGain) + with pytest.raises(ValueError): + superset_enum.validate(Gain.AUTO) # hint-only member, not on device + with pytest.raises(ValueError): + subset_enum.validate("Typo") # raw value not on device + + assert superset_enum.validate(DeviceGain.LOW) is Gain.LOW + assert superset_enum.validate(DeviceGain.HIGH) is Gain.HIGH + + assert subset_enum.validate(DeviceTrigger.EXTERNAL) is Trigger.EXTERNAL + assert subset_enum.validate(DeviceTrigger.INTERNAL) is Trigger.INTERNAL + assert subset_enum.validate(DeviceTrigger.SOFTWARE) == "Software" + + +def test_subset_enum_requires_value_type(): + with pytest.raises(TypeError, match="needs a value type"): + + class NoMixin(SubsetEnum): + A = "a" + + +def test_subset_enum_call(): + # Known values are members, unknown values of the mixin type are kept raw + class IntSubsetEnum(int, SubsetEnum): + ONE = 1 + TWO = 2 + + class IntDetectorEnum(enum.IntEnum): + ONE = 1 + TWO = 2 + THREE = 3 + + assert IntSubsetEnum(1) is IntSubsetEnum.ONE + assert IntSubsetEnum(IntDetectorEnum.TWO) is IntSubsetEnum.TWO + raw = IntSubsetEnum(IntDetectorEnum.THREE) + assert raw == 3 + assert not isinstance(raw, IntSubsetEnum) + assert isinstance(raw, IntDetectorEnum) + raw = IntSubsetEnum(4) + assert raw == 4 + assert not isinstance(raw, (IntSubsetEnum, IntDetectorEnum)) + + # Values of another type are still rejected + with pytest.raises(ValueError): + IntSubsetEnum("Five") + + +def test_subset_enum_functional_api(): + # Creating an enum with names must not be mistaken for a raw value lookup + dynamic = SubsetEnum("Dynamic", {"A": "a"}, type=str) + assert isinstance(dynamic, type) + assert issubclass(dynamic, SubsetEnum) + assert dynamic("a") is dynamic.A diff --git a/tests/transports/epics/ca/test_ca_util.py b/tests/transports/epics/ca/test_ca_util.py index fd59dc35e..e503af3b9 100644 --- a/tests/transports/epics/ca/test_ca_util.py +++ b/tests/transports/epics/ca/test_ca_util.py @@ -1,9 +1,10 @@ import enum +from typing import cast import pytest from fastcs.controllers import ControllerAPI -from fastcs.datatypes import Bool, Enum, Float, Int, String +from fastcs.datatypes import Bool, Enum, Float, Int, String, SubsetEnum from fastcs.transports.epics.ca.util import ( cast_from_epics_type, cast_to_epics_type, @@ -135,6 +136,43 @@ def test_cast_from_epics_validations(datatype, input): cast_from_epics_type(datatype, input) +class LongTrigger(str, SubsetEnum): + INTERNAL = "Internal" + EXTERNAL = "External" + + +# Device has more than 16 choices, so CA serves it as a string record of names +DeviceLongTrigger = cast( + type[enum.Enum], + enum.Enum( + "DeviceLongTrigger", + { + "INTERNAL": "Internal", + "EXTERNAL": "External", + **{f"LINE{i}": f"Line{i}" for i in range(16)}, + }, + ), +) + + +@pytest.mark.parametrize( + "value, epics_value", + [ + pytest.param(LongTrigger.INTERNAL, "INTERNAL", id="member"), + pytest.param("Line3", "LINE3", id="device-extra"), + ], +) +def test_long_subset_enum_round_trip(value, epics_value): + datatype = Enum(LongTrigger, allowed=DeviceLongTrigger) + + assert cast_to_epics_type(datatype, value) == epics_value + + result = cast_from_epics_type(datatype, epics_value) + assert result == value + # Members stay members and device extras stay raw values + assert type(result) is type(value) + + @pytest.mark.parametrize("id", ["DEVICE", "my-id", "name_1", "ABC-123_xyz"]) def test_validate_ca_id_accepts_valid(id): validate_ca_id(ControllerAPI(path=[id])) diff --git a/tests/transports/epics/ca/test_softioc_system.py b/tests/transports/epics/ca/test_softioc_system.py index 68d44a08d..faf6a177b 100644 --- a/tests/transports/epics/ca/test_softioc_system.py +++ b/tests/transports/epics/ca/test_softioc_system.py @@ -2,7 +2,7 @@ from multiprocessing import Queue import pytest -from aioca import FORMAT_TIME, caget, camonitor, caput +from aioca import DBR_CHAR_STR, FORMAT_CTRL, FORMAT_TIME, caget, camonitor, caput from softioc import alarm @@ -36,3 +36,26 @@ async def test_ioc(softioc_subprocess: tuple[str, Queue]): assert d_value.severity == alarm.NO_ALARM # Second real call succeeds finally: subscription.close() + + +@pytest.mark.asyncio +async def test_hinted_enums(softioc_subprocess: tuple[str, Queue]): + pv_prefix, _ = softioc_subprocess + + async def choices(pv: str) -> list[str]: + return list((await caget(f"{pv_prefix}:{pv}", format=FORMAT_CTRL)).enums) + + # Served choices are the device's: Superset drops AUTO, Subset adds SOFTWARE + assert await choices("Mode_RBV") == ["OFF", "ON"] + assert await choices("Gain_RBV") == ["LOW", "HIGH"] + assert await choices("Trigger_RBV") == ["INTERNAL", "EXTERNAL", "SOFTWARE"] + + # A choice only the device has can be set and read back + await caput(f"{pv_prefix}:Trigger", "SOFTWARE", wait=True) + assert await caget(f"{pv_prefix}:Trigger_RBV", datatype=str) == "SOFTWARE" + + # >16 choices are served as a long string record of names + long_rbv = f"{pv_prefix}:LongTrigger_RBV" + assert await caget(long_rbv, datatype=DBR_CHAR_STR) == "INTERNAL" + await caput(f"{pv_prefix}:LongTrigger", "LINE3", datatype=DBR_CHAR_STR, wait=True) + assert await caget(long_rbv, datatype=DBR_CHAR_STR) == "LINE3" diff --git a/tests/transports/epics/pva/test_p4p.py b/tests/transports/epics/pva/test_p4p.py index e50f74695..a104432cc 100644 --- a/tests/transports/epics/pva/test_p4p.py +++ b/tests/transports/epics/pva/test_p4p.py @@ -40,6 +40,9 @@ async def test_ioc(p4p_subprocess: tuple[str, Queue]): "r": f"{pv_prefix}:Table_RBV", "w": f"{pv_prefix}:Table", }, + "mode": {"w": f"{pv_prefix}:Mode", "r": f"{pv_prefix}:Mode_RBV"}, + "gain": {"w": f"{pv_prefix}:Gain", "r": f"{pv_prefix}:Gain_RBV"}, + "trigger": {"w": f"{pv_prefix}:Trigger", "r": f"{pv_prefix}:Trigger_RBV"}, } child_vector_pvi_pv = parent_pvi["value"]["child"]["d"] @@ -76,6 +79,31 @@ async def test_ioc(p4p_subprocess: tuple[str, Queue]): } +@pytest.mark.asyncio +async def test_hinted_enums(p4p_subprocess: tuple[str, Queue]): + pv_prefix, _ = p4p_subprocess + ctxt = Context("pva", nt=False) + + async def get_enum(pv: str) -> Value: + value = await ctxt.get(f"{pv_prefix}:{pv}") + assert isinstance(value, Value) + return value.value + + # Served choices are the device's: Superset drops AUTO, Subset adds SOFTWARE + assert list((await get_enum("Mode_RBV")).choices) == ["OFF", "ON"] + assert list((await get_enum("Gain_RBV")).choices) == ["LOW", "HIGH"] + assert list((await get_enum("Trigger_RBV")).choices) == [ + "INTERNAL", + "EXTERNAL", + "SOFTWARE", + ] + + # A choice only the device has can be set and read back + await ctxt.put(f"{pv_prefix}:Trigger", {"value.index": 2}) + trigger = await get_enum("Trigger_RBV") + assert trigger.choices[trigger.index] == "SOFTWARE" + + @pytest.mark.asyncio async def test_scan_method(p4p_subprocess: tuple[str, Queue]): pv_prefix, _ = p4p_subprocess diff --git a/tests/transports/tango/test_dsr.py b/tests/transports/tango/test_dsr.py index 61a8f73ef..ff819f735 100644 --- a/tests/transports/tango/test_dsr.py +++ b/tests/transports/tango/test_dsr.py @@ -12,7 +12,16 @@ ) from fastcs.attributes import AttrR, AttrRW, AttrW -from fastcs.datatypes import Bool, Enum, Float, Int, String, Waveform +from fastcs.datatypes import ( + Bool, + Enum, + Float, + Int, + String, + SubsetEnum, + SupersetEnum, + Waveform, +) from fastcs.transports.tango.transport import TangoTransport @@ -29,6 +38,27 @@ def mock_run_threadsafe_blocking(module_mocker: MockerFixture): yield m +class Gain(SupersetEnum): + AUTO = 0 + LOW = 1 + HIGH = 2 + + +# Device lacks AUTO, as introspection would report +DeviceGain = enum.Enum("DeviceGain", {"LOW": 1, "HIGH": 2}) + + +class Trigger(int, SubsetEnum): + INTERNAL = 0 + EXTERNAL = 1 + + +# Device has an extra SOFTWARE choice that Trigger doesn't define +DeviceTrigger = enum.Enum( + "DeviceTrigger", {"INTERNAL": 0, "EXTERNAL": 1, "SOFTWARE": 2} +) + + class TangoController(MyTestController): read_int = AttrR(Int()) read_write_int = AttrRW(Int()) @@ -37,6 +67,8 @@ class TangoController(MyTestController): write_bool = AttrW(Bool()) read_string = AttrRW(String()) enum = AttrRW(Enum(enum.IntEnum("Enum", {"RED": 0, "GREEN": 1, "BLUE": 2}))) + superset_enum = AttrRW(Enum(Gain, allowed=DeviceGain)) + subset_enum = AttrRW(Enum(Trigger, allowed=DeviceTrigger)) one_d_waveform = AttrRW(Waveform(np.int32, (10,))) two_d_waveform = AttrRW(Waveform(np.int32, (10, 10))) @@ -76,6 +108,8 @@ def test_list_attributes(self, tango_context: DeviceProxy): "WriteBool", "ReadString", "Enum", + "SupersetEnum", + "SubsetEnum", "OneDWaveform", "TwoDWaveform", "SubController01_ReadInt", @@ -162,6 +196,41 @@ def test_enum(self, tango_controller_api: AssertableControllerAPI, tango_context assert isinstance(enum_attr.get(), enum_cls) assert enum_attr.get() == enum_cls(1) + def test_superset_enum( + self, tango_controller_api: AssertableControllerAPI, tango_context + ): + attr = tango_controller_api.attributes["superset_enum"] + assert isinstance(attr, AttrRW) + labels = list(tango_context.get_attribute_config("SupersetEnum").enum_labels) + assert labels == ["LOW", "HIGH"] # no AUTO, the device doesn't have it + + with tango_controller_api.assert_read_here(["superset_enum"]): + index = tango_context.read_attribute("SupersetEnum").value + assert labels[index] == attr.get().name + + with tango_controller_api.assert_write_here(["superset_enum"]): + tango_context.write_attribute("SupersetEnum", labels.index("HIGH")) + assert attr.get() is Gain.HIGH + + def test_subset_enum( + self, tango_controller_api: AssertableControllerAPI, tango_context + ): + attr = tango_controller_api.attributes["subset_enum"] + assert isinstance(attr, AttrRW) + labels = list(tango_context.get_attribute_config("SubsetEnum").enum_labels) + assert labels == ["INTERNAL", "EXTERNAL", "SOFTWARE"] # includes device extra + + # A device extra is stored as its raw value, not a Trigger member + with tango_controller_api.assert_write_here(["subset_enum"]): + tango_context.write_attribute("SubsetEnum", labels.index("SOFTWARE")) + assert attr.get() == 2 + assert not isinstance(attr.get(), Trigger) + assert labels[tango_context.read_attribute("SubsetEnum").value] == "SOFTWARE" + + # Known choices are still Trigger members (put can only be spied on once) + tango_context.write_attribute("SubsetEnum", labels.index("EXTERNAL")) + assert attr.get() is Trigger.EXTERNAL + def test_1d_waveform( self, tango_controller_api: AssertableControllerAPI, tango_context ): diff --git a/tests/util.py b/tests/util.py index e08f606a5..8ecc61870 100644 --- a/tests/util.py +++ b/tests/util.py @@ -1,7 +1,40 @@ import enum +from fastcs.datatypes import StrictEnum, SubsetEnum, SupersetEnum + class ColourEnum(enum.IntEnum): RED = 0 GREEN = 1 BLUE = 2 + + +class Mode(StrictEnum): + """Device choices must exactly match.""" + + OFF = "Off" + ON = "On" + + +class Gain(SupersetEnum): + """Device lacks AUTO, which only exists on the setpoint.""" + + AUTO = "Auto" + LOW = "Low" + HIGH = "High" + + +class Trigger(str, SubsetEnum): + """Device has extra trigger sources we don't care about.""" + + INTERNAL = "Internal" + EXTERNAL = "External" + + +# Choices as a device would report them during introspection +DeviceMode = enum.Enum("DeviceMode", {"OFF": "Off", "ON": "On"}) +DeviceGain = enum.Enum("DeviceGain", {"LOW": "Low", "HIGH": "High"}) +DeviceTrigger = enum.Enum( + "DeviceTrigger", + {"INTERNAL": "Internal", "EXTERNAL": "External", "SOFTWARE": "Software"}, +)