diff --git a/src/fastcs/attributes/__init__.py b/src/fastcs/attributes/__init__.py index b25e66f7..d0f5e59f 100644 --- a/src/fastcs/attributes/__init__.py +++ b/src/fastcs/attributes/__init__.py @@ -2,6 +2,7 @@ from .attr_rw import AttrRW as AttrRW from .attr_w import AttrW as AttrW from .attribute import Attribute as Attribute +from .attribute import AttributeAccessMode as AttributeAccessMode from .attribute_io import AnyAttributeIO as AnyAttributeIO from .attribute_io import AttributeIO as AttributeIO from .attribute_io_ref import AttributeIORef as AttributeIORef diff --git a/src/fastcs/attributes/attr_r.py b/src/fastcs/attributes/attr_r.py index aec7f522..ae67c6d1 100644 --- a/src/fastcs/attributes/attr_r.py +++ b/src/fastcs/attributes/attr_r.py @@ -4,7 +4,7 @@ from collections.abc import Awaitable, Callable from typing import Any -from fastcs.attributes.attribute import Attribute +from fastcs.attributes.attribute import Attribute, AttributeAccessMode from fastcs.attributes.attribute_io_ref import AttributeIORefT from fastcs.attributes.util import AttrValuePredicate, PredicateEvent from fastcs.datatypes import DataType, DType_T @@ -47,6 +47,10 @@ def get(self) -> DType_T: """Get the cached value of the attribute.""" return self._value + @property + def access_mode(self) -> AttributeAccessMode: + return "r" + async def update(self, value: Any) -> None: """Update the value of the attibute diff --git a/src/fastcs/attributes/attr_rw.py b/src/fastcs/attributes/attr_rw.py index c8365373..af6491a0 100644 --- a/src/fastcs/attributes/attr_rw.py +++ b/src/fastcs/attributes/attr_rw.py @@ -1,5 +1,6 @@ from fastcs.attributes.attr_r import AttrR from fastcs.attributes.attr_w import AttrW +from fastcs.attributes.attribute import AttributeAccessMode from fastcs.attributes.attribute_io_ref import AttributeIORefT from fastcs.datatypes import DataType, DType_T from fastcs.logging import bind_logger @@ -25,6 +26,10 @@ def __init__( if io_ref is None: self.set_on_put_callback(self._internal_update) + @property + def access_mode(self) -> AttributeAccessMode: + return "rw" + async def _internal_update( self, attr: AttrW[DType_T, AttributeIORefT], value: DType_T ): diff --git a/src/fastcs/attributes/attr_w.py b/src/fastcs/attributes/attr_w.py index a9b2cff7..4532364a 100644 --- a/src/fastcs/attributes/attr_w.py +++ b/src/fastcs/attributes/attr_w.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Callable from typing import Any -from fastcs.attributes.attribute import Attribute +from fastcs.attributes.attribute import Attribute, AttributeAccessMode from fastcs.attributes.attribute_io_ref import AttributeIORefT from fastcs.datatypes import DataType, DType_T from fastcs.logging import bind_logger @@ -37,6 +37,10 @@ def __init__( self._sync_setpoint_callbacks: list[AttrSyncSetpointCallback[DType_T]] = [] """Callbacks to publish changes to the setpoint of the attribute""" + @property + def access_mode(self) -> AttributeAccessMode: + return "w" + async def put(self, setpoint: DType_T, sync_setpoint: bool = False) -> None: """Set the setpoint of the attribute diff --git a/src/fastcs/attributes/attribute.py b/src/fastcs/attributes/attribute.py index c737bd28..ae6e3852 100644 --- a/src/fastcs/attributes/attribute.py +++ b/src/fastcs/attributes/attribute.py @@ -1,5 +1,6 @@ +from abc import ABC, abstractmethod from collections.abc import Callable -from typing import Generic +from typing import Generic, Literal from fastcs.attributes.attribute_io_ref import AttributeIORefT from fastcs.datatypes import DataType, DType, DType_T @@ -8,8 +9,10 @@ logger = bind_logger(logger_name=__name__) +AttributeAccessMode = Literal["r", "w", "rw"] -class Attribute(Generic[DType_T, AttributeIORefT], Tracer): + +class Attribute(Generic[DType_T, AttributeIORefT], Tracer, ABC): """Base FastCS attribute. Instances of this class added to a ``Controller`` will be used by the FastCS class. @@ -74,6 +77,12 @@ def path(self) -> list[str]: def full_name(self) -> str: return ".".join(self._path + [self._name]) + @property + @abstractmethod + def access_mode(self) -> AttributeAccessMode: + """The access mode of this attribute.""" + ... + def add_update_datatype_callback( self, callback: Callable[[DataType[DType_T]], None] ) -> None: diff --git a/src/fastcs/transports/epics/pva/pvi.py b/src/fastcs/transports/epics/pva/pvi.py index 2823fe38..5c381c05 100644 --- a/src/fastcs/transports/epics/pva/pvi.py +++ b/src/fastcs/transports/epics/pva/pvi.py @@ -6,7 +6,6 @@ from p4p.server import StaticProvider from p4p.server.asyncio import SharedPV -from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW from fastcs.transports.controller_api import ControllerAPI from fastcs.util import snake_to_pascal @@ -15,19 +14,6 @@ AccessModeType = Literal["r", "w", "rw", "d", "x"] -# TODO: This should be removed after https://github.com/DiamondLightSource/FastCS/issues/260 -def _attribute_to_access(attribute: Attribute) -> AccessModeType: - match attribute: - case AttrRW(): - return "rw" - case AttrR(): - return "r" - case AttrW(): - return "w" - case _: - raise ValueError(f"Unknown attribute type {type(attribute)}") - - def add_pvi_info( provider: StaticProvider, pv_prefix: str, @@ -79,7 +65,7 @@ def _make_p4p_raw_value(pv_prefix: str, controller_api: ControllerAPI) -> dict: for pv_leaf, attribute in controller_api.attributes.items(): # Add attribute entry pv = f"{pv_prefix}:{snake_to_pascal(pv_leaf)}" - p4p_raw_value[pv_leaf][_attribute_to_access(attribute)] = pv + p4p_raw_value[pv_leaf][attribute.access_mode] = pv for pv_leaf, _ in controller_api.command_methods.items(): pv = f"{pv_prefix}:{snake_to_pascal(pv_leaf)}" p4p_raw_value[pv_leaf]["x"] = pv diff --git a/tests/test_attributes.py b/tests/test_attributes.py index 41de437a..e9fb5e66 100644 --- a/tests/test_attributes.py +++ b/tests/test_attributes.py @@ -6,11 +6,38 @@ import pytest from pytest_mock import MockerFixture -from fastcs.attributes import AttributeIO, AttributeIORef, AttrR, AttrRW, AttrW +from fastcs.attributes import ( + AttributeAccessMode, + AttributeIO, + AttributeIORef, + AttrR, + AttrRW, + AttrW, +) from fastcs.controllers import Controller from fastcs.datatypes import Float, Int, String +def test_attribute_access_mode(): + """Test that attributes have the correct access_mode property.""" + attr_r = AttrR(String()) + attr_w = AttrW(String()) + attr_rw = AttrRW(String()) + + assert attr_r.access_mode == "r" + assert attr_w.access_mode == "w" + assert attr_rw.access_mode == "rw" + + # Verify type compatibility + mode_r: AttributeAccessMode = attr_r.access_mode + mode_w: AttributeAccessMode = attr_w.access_mode + mode_rw: AttributeAccessMode = attr_rw.access_mode + + assert isinstance(mode_r, str) + assert isinstance(mode_w, str) + assert isinstance(mode_rw, str) + + def test_attr_r(): attr = AttrR(String(), group="test group")