fix(mo1_bragg): stop constant polling of ACS config signals

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

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
     )
 
 

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

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)

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