From 5db31fb322dc9772cb10fc8c8b2633b42ab9d549 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 5 Aug 2026 13:16:11 +0200 Subject: [PATCH] fix(mo1_bragg): stop constant polling of ACS config signals --- debye_bec/devices/mo1_bragg/acs.py | 102 ++++++++++++++--- .../devices/mo1_bragg/mo1_bragg_devices.py | 40 +++++-- tests/tests_devices/test_acs_signal.py | 108 ++++++++++++++++++ tests/tests_devices/test_mo1_bragg.py | 51 ++++++++- tests/tests_devices/test_mo1_bragg_angle.py | 6 +- 5 files changed, 274 insertions(+), 33 deletions(-) create mode 100644 tests/tests_devices/test_acs_signal.py diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index 6eafa61..28ffd23 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -2,7 +2,9 @@ ACS controller device exposing plain read/write variables (no motion). Uses the same BEC building blocks as before: - - ophyd_devices.utils.controller.Controller -> shared TCP/IP communicator + - debye_bec.devices.utils.term_trail_controller.TermTrailController + -> shared TCP/IP communicator (vendored term/trail extension of + ophyd_devices.utils.controller.Controller) - ophyd_devices.utils.socket.SocketIO -> raw socket helper - ophyd_devices.utils.socket.SocketSignal -> Signal base talking through it @@ -14,18 +16,19 @@ Protocol: from __future__ import annotations import time -import traceback from enum import Enum import numpy as np from bec_lib.logger import bec_logger -from ophyd_devices.utils.controller import Controller, threadlocked +from ophyd_devices.utils.controller import threadlocked from ophyd_devices.utils.socket import SocketSignal +from debye_bec.devices.utils.term_trail_controller import TermTrailController + logger = bec_logger.logger -class ACSController(Controller): +class ACSController(TermTrailController): """ Shared TCP/IP communicator for one ACS controller. @@ -37,7 +40,6 @@ class ACSController(Controller): _axes_per_controller = 0 # not used for plain variables, no motion axes def __init__(self, *, socket_cls, socket_host, socket_port, device_manager): - socket_cls.socket_timeout = 5 super().__init__( socket_cls=socket_cls, socket_host=socket_host, @@ -72,7 +74,6 @@ class ACSController(Controller): if self.sock is None: self.on() idx = f",{idx:0.0f}" if idx is not None else "" - # logger.info(f"Send request: SETVAR({np.round(value, prec)},{tag}{idx})") reply = self.socket_put_and_receive(f"SETVAR({np.round(value, prec)},{tag}{idx})") if reply.startswith("?"): @@ -81,34 +82,99 @@ class ACSController(Controller): class AcsSignal(SocketSignal): - """Read/write ACS controller variable, identified by its tag number.""" + """Read/write ACS controller variable, identified by its tag number. - def __init__(self, *args, tag: int, prec: int, num_el: int = 1, enum: Enum = None, **kwargs): + Args: + tag (int): ACS variable tag number. + prec (int): Decimal precision used for reads and writes. + num_el (int): Number of elements for array variables. + enum (Enum): Optional enum mapping raw values to names. + cache_ttl (float | None): If set, `get()` serves the cached value for up to + this many seconds instead of querying the controller. The cache is + dropped on every `put()` (and via `invalidate_cache()`), so a fresh + hardware read follows each real configuration change. Use this for + config-kind signals: the BEC device server re-reads the full device + configuration whenever any auto-monitored signal of the device updates, + and without a cache each of those reads is a blocking GETVAR round trip. + None (default) disables caching. + """ + + def __init__( + self, + *args, + tag: int, + prec: int, + num_el: int = 1, + enum: Enum = None, + cache_ttl: float | None = None, + **kwargs, + ): self.tag = tag self.prec = prec self.num_el = num_el self.enum = enum - self.last_get = time.time() + self.cache_ttl = cache_ttl + self._last_hw_read: float | None = None super().__init__(*args, **kwargs) @property def controller(self) -> ACSController: return self.root.controller - def _socket_get(self): - now = time.time() - interval = now - self.last_get - self.last_get = now - logger.info(f"Get signal with tag {self.tag}, time to last get: {interval*1e3} ms") - # logger.info(f"socket_get called from: {traceback.format_stack()}") + def get(self, **kwargs): + """Return the cached value while it is fresh; otherwise query the controller. + Deviates from `SocketSignal.get` in two ways: reads within `cache_ttl` of the + last hardware read are served from `self._readback` without touching the + socket, and the value callbacks only run when the value actually changed. + Unconditional callbacks would re-trigger the device server's auto-monitor + machinery on every read and turn polling into a self-sustaining loop. + """ + if self.SUB_VALUE in self._active_socket_callbacks: + return self._readback + if self._cache_is_fresh(): + return self._readback + old_value = self._readback + self._readback = self._socket_get() + timestamp = time.time() + self._metadata["timestamp"] = timestamp + self._last_hw_read = timestamp + if not self._values_equal(old_value, self._readback): + self._run_subs( + sub_type=self.SUB_VALUE, + old_value=old_value, + value=self._readback, + timestamp=timestamp, + ) + return self._readback + + def put(self, value, connection_timeout=1, **kwargs): + super().put(value, connection_timeout=connection_timeout, **kwargs) + # the controller may quantize the written value, so drop the cache and let + # the next read report what the hardware actually stored + self.invalidate_cache() + + def invalidate_cache(self) -> None: + """Force the next `get()` to read from the controller.""" + self._last_hw_read = None + + def _cache_is_fresh(self) -> bool: + if self.cache_ttl is None or self._last_hw_read is None: + return False + return (time.time() - self._last_hw_read) < self.cache_ttl + + @staticmethod + def _values_equal(old_value, new_value) -> bool: + if isinstance(old_value, np.ndarray) or isinstance(new_value, np.ndarray): + return np.array_equal(old_value, new_value) + return old_value == new_value + + def _socket_get(self): def convert(val): return self.enum(val).name if self.enum is not None else val if self.num_el <= 1: - val = convert(self.controller.get_var(self.tag, self.prec)) - logger.info(f"Get signal with tag {self.tag}, time to last get: {interval*1e3} ms") - return val + return convert(self.controller.get_var(self.tag, self.prec)) return np.array( [convert(self.controller.get_var(self.tag, self.prec, i)) for i in range(self.num_el)] ) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index 7ee3051..1da65a2 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -69,7 +69,10 @@ class Mo1BraggStatus(Device): enc_initialized = Cpt( EpicsSignalRO, suffix="enc_initialized_RBV", kind="config", auto_monitor=True ) - heartbeat = Cpt(EpicsSignalRO, suffix="heartbeat_RBV", kind="config", auto_monitor=True) + # kind="omitted": the heartbeat ticks continuously, and any auto-monitored + # config-kind signal update makes the device server re-read the full device + # configuration (incl. the ACS socket signals) on every tick + heartbeat = Cpt(EpicsSignalRO, suffix="heartbeat_RBV", kind="omitted", auto_monitor=True) class Mo1BraggEncoder(Device): @@ -148,9 +151,17 @@ class Mo1BraggScanSettings(Device): # EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True # ) - s_scan_energy_lo = Cpt(AcsSignal, tag=53003, prec=6, kind="config", auto_monitor=False) - s_scan_energy_hi = Cpt(AcsSignal, tag=53004, prec=6, kind="config", auto_monitor=False) - s_scan_scantime = Cpt(AcsSignal, tag=53002, prec=3, kind="config", auto_monitor=False) + # cache_ttl: baseline/config reads are served from cache; every put() drops the + # cache, so a fresh hardware read follows each real configuration change + s_scan_energy_lo = Cpt( + AcsSignal, tag=53003, prec=6, kind="config", auto_monitor=False, cache_ttl=30.0 + ) + s_scan_energy_hi = Cpt( + AcsSignal, tag=53004, prec=6, kind="config", auto_monitor=False, cache_ttl=30.0 + ) + s_scan_scantime = Cpt( + AcsSignal, tag=53002, prec=3, kind="config", auto_monitor=False, cache_ttl=30.0 + ) # XAS advanced scan settings a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False) @@ -214,26 +225,33 @@ class Mo1BraggScanControl(Device): EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True ) scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=True) - scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="config", auto_monitor=True) + # The live scan-status PVs below are kind="omitted": they change during every scan + # (some, like the progress/time counters, every second), and each update of an + # auto-monitored config-kind signal makes the device server re-read and republish + # the full device configuration. They are still fully usable via get()/subscribe() + # (CompareStatus/TransitionStatus waits, progress forwarding), which ignore kind. + scan_msg = Cpt(EpicsSignalRO, suffix="scan_msg_ENUM_RBV", kind="omitted", auto_monitor=True) scan_start_infinite = Cpt( EpicsSignal, suffix="scan_start_infinite", kind="config", put_complete=True ) scan_start_timer = Cpt(EpicsSignal, suffix="scan_start_timer", kind="config", put_complete=True) scan_stop = Cpt(EpicsSignal, suffix="scan_stop", kind="config", put_complete=True) scan_status = Cpt( - EpicsSignalRO, suffix="scan_status_ENUM_RBV", kind="config", auto_monitor=True + EpicsSignalRO, suffix="scan_status_ENUM_RBV", kind="omitted", auto_monitor=True ) scan_time_left = Cpt( - EpicsSignalRO, suffix="scan_time_left_RBV", kind="config", auto_monitor=True + EpicsSignalRO, suffix="scan_time_left_RBV", kind="omitted", auto_monitor=True ) - scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="config", auto_monitor=True) + scan_done = Cpt(EpicsSignalRO, suffix="scan_done_RBV", kind="omitted", auto_monitor=True) scan_val_reset = Cpt(EpicsSignal, suffix="scan_val_reset", kind="config", put_complete=True) - scan_progress = Cpt(EpicsSignalRO, suffix="scan_progress_RBV", kind="config", auto_monitor=True) + scan_progress = Cpt( + EpicsSignalRO, suffix="scan_progress_RBV", kind="omitted", auto_monitor=True + ) scan_spectra_done = Cpt( - EpicsSignalRO, suffix="scan_n_osc_RBV", kind="config", auto_monitor=True + EpicsSignalRO, suffix="scan_n_osc_RBV", kind="omitted", auto_monitor=True ) scan_spectra_left = Cpt( - EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="config", auto_monitor=True + EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="omitted", auto_monitor=True ) diff --git a/tests/tests_devices/test_acs_signal.py b/tests/tests_devices/test_acs_signal.py new file mode 100644 index 0000000..f497749 --- /dev/null +++ b/tests/tests_devices/test_acs_signal.py @@ -0,0 +1,108 @@ +"""Tests for the AcsSignal config-read cache and change-only value callbacks.""" + +import time +from unittest import mock + +import numpy as np +import pytest +from ophyd import Component as Cpt +from ophyd import Device + +from debye_bec.devices.mo1_bragg.acs import AcsSignal + +# pylint: disable=protected-access + + +class _AcsDevice(Device): + """Minimal host device providing the controller attribute AcsSignal expects.""" + + cached = Cpt(AcsSignal, tag=100, prec=3, kind="config", cache_ttl=30.0) + uncached = Cpt(AcsSignal, tag=200, prec=3, kind="config") + array = Cpt(AcsSignal, tag=300, prec=3, num_el=3, kind="config", cache_ttl=30.0) + + def __init__(self, *args, controller=None, **kwargs): + self.controller = controller + super().__init__(*args, **kwargs) + + +@pytest.fixture +def acs_device(): + controller = mock.MagicMock() + controller.get_var.return_value = 1.5 + yield _AcsDevice(name="acs", controller=controller) + + +def test_cached_get_reads_hardware_once(acs_device): + assert acs_device.cached.get() == 1.5 + assert acs_device.cached.get() == 1.5 + acs_device.controller.get_var.assert_called_once_with(100, 3) + + +def test_uncached_get_reads_hardware_every_time(acs_device): + acs_device.uncached.get() + acs_device.uncached.get() + assert acs_device.controller.get_var.call_count == 2 + + +def test_cache_expires_after_ttl(acs_device): + acs_device.cached.cache_ttl = 0.05 + acs_device.cached.get() + time.sleep(0.06) + acs_device.cached.get() + assert acs_device.controller.get_var.call_count == 2 + + +def test_put_invalidates_cache(acs_device): + acs_device.cached.get() + acs_device.cached.put(2.0) + acs_device.controller.set_var.assert_called_once_with(100, 2.0, 3) + acs_device.controller.get_var.return_value = 2.0 + assert acs_device.cached.get() == 2.0 + assert acs_device.controller.get_var.call_count == 2 + + +def test_invalidate_cache_forces_fresh_read(acs_device): + acs_device.cached.get() + acs_device.cached.invalidate_cache() + acs_device.cached.get() + assert acs_device.controller.get_var.call_count == 2 + + +def test_value_callbacks_fire_only_on_change(acs_device): + events = [] + acs_device.cached.subscribe(lambda **kwargs: events.append(kwargs), run=False) + + acs_device.cached.get() # None -> 1.5: fires + assert len(events) == 1 + + acs_device.cached.invalidate_cache() + acs_device.cached.get() # 1.5 -> 1.5: hardware read, but no callback + assert acs_device.controller.get_var.call_count == 2 + assert len(events) == 1 + + acs_device.controller.get_var.return_value = 2.5 + acs_device.cached.invalidate_cache() + acs_device.cached.get() # 1.5 -> 2.5: fires + assert len(events) == 2 + assert events[-1]["value"] == 2.5 + + +def test_array_signal_caches_and_compares_by_content(acs_device): + acs_device.controller.get_var.side_effect = lambda tag, prec, idx=None: float(idx) + + events = [] + acs_device.array.subscribe(lambda **kwargs: events.append(kwargs), run=False) + + assert np.array_equal(acs_device.array.get(), np.array([0.0, 1.0, 2.0])) + assert acs_device.controller.get_var.call_count == 3 # one call per element + assert len(events) == 1 + + # cache hit: no additional hardware reads + acs_device.array.get() + assert acs_device.controller.get_var.call_count == 3 + + # fresh read with identical content: no callback + acs_device.array.invalidate_cache() + acs_device.array.get() + assert acs_device.controller.get_var.call_count == 6 + assert len(events) == 1 diff --git a/tests/tests_devices/test_mo1_bragg.py b/tests/tests_devices/test_mo1_bragg.py index 89fd6e2..aec8e45 100644 --- a/tests/tests_devices/test_mo1_bragg.py +++ b/tests/tests_devices/test_mo1_bragg.py @@ -8,6 +8,7 @@ from unittest import mock import ophyd import pytest from bec_lib.messages import ScanQueueMessage, ScanStatusMessage +from bec_server.device_server.tests.utils import DMMock from bec_server.scan_server.scan_assembler import ScanAssembler from bec_server.scan_server.scan_queue import RequestBlock from bec_server.scan_server.scan_worker import ScanWorker @@ -15,7 +16,7 @@ from bec_server.scan_server.tests.fixtures import scan_server_mock from ophyd.utils import LimitError from ophyd_devices.tests.utils import MockPV -# from bec_server.device_server.tests.utils import DMMock +from debye_bec.devices.mo1_bragg.acs import ACSController from debye_bec.devices.mo1_bragg.mo1_bragg import ( Mo1Bragg, Mo1BraggError, @@ -40,12 +41,28 @@ def scan_worker_mock(scan_server_mock): def mock_bragg(): name = "bragg" prefix = "X01DA-OP-MO1:BRAGG:" - with mock.patch.object(ophyd, "cl") as mock_cl: + # dict-backed stand-in for the ACS controller so AcsSignal set/get round-trips + # work without a socket + acs_store = {} + + def _set_var(tag, value, prec, idx=None): + acs_store[(tag, idx)] = value + + def _get_var(tag, prec, idx=None): + return acs_store.get((tag, idx), 0.0) + + ACSController._reset_controller() + with ( + mock.patch.object(ACSController, "get_var", side_effect=_get_var), + mock.patch.object(ACSController, "set_var", side_effect=_set_var), + mock.patch.object(ophyd, "cl") as mock_cl, + ): mock_cl.get_pv = MockPV mock_cl.thread_class = threading.Thread - dev = Mo1Bragg(name=name, prefix=prefix) + dev = Mo1Bragg(name=name, prefix=prefix, device_manager=DMMock()) patch_dual_pvs(dev) yield dev + ACSController._reset_controller() def test_init(mock_bragg): @@ -119,6 +136,34 @@ def test_set_xtal(mock_bragg): assert dev.crystal.xtal_enum.get() == 1 +def test_read_configuration_uses_acs_cache_and_omits_status_pvs(mock_bragg): + """The device server re-reads the full configuration whenever any auto-monitored + signal updates; the ticking status PVs must not be part of it, and repeated + config reads must not hit the ACS controller again while the cache is fresh.""" + dev = mock_bragg + config = dev.read_configuration() + assert "bragg_scan_settings_s_scan_energy_lo" in config + assert "bragg_scan_settings_s_scan_energy_hi" in config + assert "bragg_scan_settings_s_scan_scantime" in config + # live status/progress PVs are omitted from the configuration + assert "bragg_status_heartbeat" not in config + assert "bragg_scan_control_scan_msg" not in config + assert "bragg_scan_control_scan_status" not in config + assert "bragg_scan_control_scan_progress" not in config + assert "bragg_scan_control_scan_time_left" not in config + assert "bragg_scan_control_scan_done" not in config + + acs_reads = dev.controller.get_var.call_count + dev.read_configuration() + assert dev.controller.get_var.call_count == acs_reads # served from cache + + # a config change drops the cache for exactly that signal + dev.scan_settings.s_scan_energy_lo.put(7000.0) + dev.read_configuration() + assert dev.controller.get_var.call_count == acs_reads + 1 + assert dev.scan_settings.s_scan_energy_lo.get() == 7000.0 + + def test_set_xas_settings(mock_bragg): dev = mock_bragg dev.set_xas_settings(low=0.5, high=1, scan_time=0.1) diff --git a/tests/tests_devices/test_mo1_bragg_angle.py b/tests/tests_devices/test_mo1_bragg_angle.py index c09c055..5e9ea6e 100644 --- a/tests/tests_devices/test_mo1_bragg_angle.py +++ b/tests/tests_devices/test_mo1_bragg_angle.py @@ -5,8 +5,10 @@ from unittest import mock import ophyd import pytest +from bec_server.device_server.tests.utils import DMMock from ophyd_devices.tests.utils import MockPV, patch_dual_pvs +from debye_bec.devices.mo1_bragg.acs import ACSController from debye_bec.devices.mo1_bragg.mo1_bragg_angle import Mo1BraggAngle from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggStoppedError @@ -18,12 +20,14 @@ def mock_bragg() -> Mo1BraggAngle: """Fixture for the Mo1BraggAngle device.""" name = "bragg" prefix = "X01DA-OP-MO1:BRAGG:" + ACSController._reset_controller() with mock.patch.object(ophyd, "cl") as mock_cl: mock_cl.get_pv = MockPV mock_cl.thread_class = threading.Thread - dev = Mo1BraggAngle(name=name, prefix=prefix) + dev = Mo1BraggAngle(name=name, prefix=prefix, device_manager=DMMock()) patch_dual_pvs(dev) yield dev + ACSController._reset_controller() def test_mo1_bragg_angle_init(mock_bragg):