Add access_mode property to Attribute classes (#260)

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

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
 

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
     ):

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
 

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:

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

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")