diff --git a/pyaml_cs_oa/scalar_aggregator.py b/pyaml_cs_oa/aggregator.py similarity index 95% rename from pyaml_cs_oa/scalar_aggregator.py rename to pyaml_cs_oa/aggregator.py index 48c2716..7e959d6 100644 --- a/pyaml_cs_oa/scalar_aggregator.py +++ b/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]: diff --git a/pyaml_cs_oa/controlsystem.py b/pyaml_cs_oa/controlsystem.py index ea81413..361d03c 100644 --- a/pyaml_cs_oa/controlsystem.py +++ b/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__) diff --git a/pyaml_cs_oa/dynamic_catalog.py b/pyaml_cs_oa/dynamic_catalog.py index b595016..0779089 100644 --- a/pyaml_cs_oa/dynamic_catalog.py +++ b/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) diff --git a/tests/test_scalar_aggregator.py b/tests/test_scalar_aggregator.py index c934cd0..d11e22c 100644 --- a/tests/test_scalar_aggregator.py +++ b/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,7 +100,7 @@ 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"): @@ -108,7 +108,7 @@ def test_set_rejects_length_mismatch() -> None: 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]))