Simplify aggregator

pyaml_cs_oa/scalar_aggregator.py → pyaml_cs_oa/aggregator.py

@@ -10,22 +10,15 @@
 from . import arun
 from .float_signal import FloatSignalContainer
 
-PYAMLCLASS: str = "OAScalarAggregator"
 
-
-class ConfigModel(BaseModel):
-    pass
-
-
-class OAScalarAggregator(DeviceAccessList):
-    def __init__(self, cfg: ConfigModel = None):
+class OAAggregator(DeviceAccessList):
+    def __init__(self):
         super().__init__()
         self._r_signal_list = {}  # List of signal to read
         self._w_signal_list = {}  # List of signal to read/write
         self._writable = None
 
     def _add_to_dev_list(self, d: FloatSignalContainer):
-
         # Check type and read/write
         if not isinstance(d, FloatSignalContainer):
             raise PyAMLException("All devices must be instances of FloatSignalContainer.")
@@ -65,7 +58,6 @@ def get_devices(self) -> DeviceAccess | list[DeviceAccess]:
             return self
 
     def set(self, value: npt.NDArray[np.float64]):
-
         if len(value) != len(self):
             raise PyAMLException(f"Size of value ({len(value)} do not match the number of managed devices ({len(self)})")
 
@@ -79,7 +71,6 @@ def set_and_wait(self, value: npt.NDArray[np.float64]):
         raise NotImplementedError("Not implemented yet.")
 
     def _read(self, signal_list: dict) -> npt.NDArray[np.float64]:
-
         requests = []  # list of status to await
         for _d, dc in signal_list.items():
             requests.append(dc["source"].async_get())
@@ -97,14 +88,12 @@ def _read(self, signal_list: dict) -> npt.NDArray[np.float64]:
         return rvalues
 
     def get(self) -> npt.NDArray[np.float64]:
-
         if self._writable:
             return self._read(self._w_signal_list)
         else:
             return self._read(self._r_signal_list)
 
     def readback(self) -> np.array:
-
         return self._read(self._r_signal_list)
 
     def get_range(self) -> list[float]:

pyaml_cs_oa/controlsystem.py

@@ -3,9 +3,11 @@
 from pyaml.common.exception import PyAMLException
 from pyaml.control.controlsystem import ControlSystem
 from pyaml.control.deviceaccess import DeviceAccess
+from pyaml.control.deviceaccesslist import DeviceAccessList
 from pydantic import BaseModel, ConfigDict
 
 from . import __version__
+from .aggregator import OAAggregator
 from .catalog import Catalog
 from .epicsR import EpicsR
 from .epicsRW import EpicsRW
@@ -35,12 +37,6 @@ class ConfigModel(BaseModel):
         If None specified a dynamic catalog is used.
     debug_level : str
         Debug verbosity level.
-    scalar_aggregator : str
-        Aggregator module for scalar values. If none specified, writings and
-        readings of sclar value are serialized.
-    vector_aggregator : str
-        Aggregator module for vecrors. If none specified, writings and readings
-        of vector are serialized,
     """
 
     model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
@@ -49,8 +45,6 @@ class ConfigModel(BaseModel):
     prefix: str = ""
     catalog: Catalog | None = None
     debug_level: str | None = None
-    scalar_aggregator: str | None = "pyaml_cs_oa.scalar_aggregator"
-    vector_aggregator: str | None = None
 
 
 class OphydAsyncControlSystem(ControlSystem):
@@ -155,27 +149,9 @@ def name(self) -> str:
         """
         return self._cfg.name
 
-    def scalar_aggregator(self) -> str | None:
-        """
-        Returns the module name used for handling aggregator of DeviceAccess
-
-        Returns
-        -------
-        str
-            Aggregator module name
-        """
-        return self._cfg.scalar_aggregator
-
-    def vector_aggregator(self) -> str | None:
-        """
-        Returns the module name used for handling aggregator of DeviceVectorAccess
-
-        Returns
-        -------
-        str
-            Aggregator module name
-        """
-        return self._cfg.vector_aggregator
+    def get_aggregator(self) -> DeviceAccessList | None:
+        """Returns a new empty DeviceAccessList. If None is returned serialized readings/writtings are performed"""
+        return OAAggregator()
 
     def __repr__(self):
         return repr(self._cfg).replace("ConfigModel", self.__class__.__name__)

pyaml_cs_oa/dynamic_catalog.py

@@ -20,8 +20,9 @@ class ConfigModel(BaseModel):
 
     For EPCIS::
 
-    - ``READ_PV[unit]``                 → scalar read-only
-    - ``READ_PV@index[unit]``           → array PV with element index, read-only
+    - ``READ_PV[unit]``                   → scalar read-only
+    - ``(WRITE_PV)[unit]``                → scalar write-only
+    - ``READ_PV@index[unit]``             → array PV with element index, read-only
     - ``(READ_PV, WRITE_PV)[unit]``       → scalar read-write
     - ``(READ_PV, WRITE_PV)@index[unit]`` → array PV with element index read-write
 
@@ -59,7 +60,6 @@ def resolve(self, key: str) -> BaseModel:
 
 
 def _extract_unit(token: str) -> Tuple[str, str]:
-
     # Extract unit block
     try:
         start_idx = token.index("[") + 1
@@ -72,8 +72,7 @@ def _extract_unit(token: str) -> Tuple[str, str]:
 # ── PV spec parser ────────────────────────────────────────────────────────────
 
 
-def _parse_pv(token: str) -> tuple[list[str], int | None, str]:
-
+def _parse_pv(token: str) -> tuple[list[str], int | None, str, bool]:
     token = token.strip()
 
     unit, token = _extract_unit(token)
@@ -90,21 +89,27 @@ def _parse_pv(token: str) -> tuple[list[str], int | None, str]:
 
     # Parenthesized keys describe one read PV and one write PV.
     if token.startswith("(") and token.endswith(")"):
+        hasw = True
         names = token[1:-1]
         name_list = [name.strip() for name in names.split(",")]
     else:
+        hasw = False
         name_list = [token.strip()]
 
-    return name_list, index, unit
+    return name_list, index, unit, hasw
 
 
 def _build_epics_config(pv_str: str, timeout_ms: int) -> DeviceAccess:
     from .epicsR import ConfigModel as EpicsRConfig
     from .epicsRW import ConfigModel as EpicsRWConfig
+    from .epicsW import ConfigModel as EpicsWConfig
 
-    pv_names, index, unit = _parse_pv(pv_str)
+    pv_names, index, unit, hasw = _parse_pv(pv_str)
     if len(pv_names) == 1:
-        return EpicsRConfig(read_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
+        if hasw:
+            return EpicsWConfig(write_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
+        else:
+            return EpicsRConfig(read_pvname=pv_names[0], timeout_ms=timeout_ms, index=index, unit=unit)
     if len(pv_names) == 2:
         return EpicsRWConfig(read_pvname=pv_names[0], write_pvname=pv_names[1], timeout_ms=timeout_ms, index=index, unit=unit)
     raise PyAMLException(f"Too many comma-separated tokens in key '{pv_str}' (max 2)")
@@ -114,7 +119,6 @@ def _build_epics_config(pv_str: str, timeout_ms: int) -> DeviceAccess:
 
 
 def _parse_attribute(token: str) -> tuple[list[str], int | None, str]:
-
     token = token.strip()
 
     unit, token = _extract_unit(token)
@@ -133,7 +137,6 @@ def _parse_attribute(token: str) -> tuple[list[str], int | None, str]:
 
 
 def _build_tango_config(att_name: str, timeout_ms: int) -> BaseModel:
-
     from .tangoAtt import ConfigModel as TangoAtt
 
     att_name, index, unit = _parse_attribute(att_name)

tests/test_scalar_aggregator.py

@@ -3,8 +3,8 @@
 from pyaml import PyAMLException
 from pyaml.control.deviceaccess import DeviceAccess
 
+from pyaml_cs_oa.aggregator import OAAggregator
 from pyaml_cs_oa.float_signal import FloatSignalContainer
-from pyaml_cs_oa.scalar_aggregator import OAScalarAggregator
 from pyaml_cs_oa.types import EpicsConfigR
 
 
@@ -67,21 +67,21 @@ def check_device_availability(self) -> bool:
 
 
 def test_add_devices_rejects_single_wrong_device() -> None:
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
 
     with pytest.raises(PyAMLException, match="All devices must be instances"):
         aggregator.add_devices(WrongDevice())
 
 
 def test_add_devices_rejects_wrong_device_in_list() -> None:
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
 
     with pytest.raises(PyAMLException, match="All devices must be instances"):
         aggregator.add_devices([FakeScalarSignal(1.0), WrongDevice()])
 
 
 def test_get_returns_values_in_device_order() -> None:
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
     aggregator.add_devices([FakeScalarSignal(1.0), FakeScalarSignal(2.0)])
 
     np.testing.assert_array_equal(aggregator.get(), np.array([1.0, 2.0]))
@@ -90,7 +90,7 @@ def test_get_returns_values_in_device_order() -> None:
 def test_set_writes_values_in_device_order() -> None:
     first = FakeScalarSignal(1.0)
     second = FakeScalarSignal(2.0)
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
     aggregator.add_devices([first, second])
 
     aggregator.set(np.array([10.0, 20.0]))
@@ -100,15 +100,15 @@ def test_set_writes_values_in_device_order() -> None:
 
 
 def test_set_rejects_length_mismatch() -> None:
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
     aggregator.add_devices([FakeScalarSignal(1.0), FakeScalarSignal(2.0)])
 
     with pytest.raises(PyAMLException, match="do not match"):
         aggregator.set(np.array([10.0]))
 
 
 def test_readback_returns_readback_values_in_device_order() -> None:
-    aggregator = OAScalarAggregator()
+    aggregator = OAAggregator()
     aggregator.add_devices([FakeScalarSignal(3.0), FakeScalarSignal(4.0)])
 
     np.testing.assert_array_equal(aggregator.readback(), np.array([3.0, 4.0]))