From 637abe50c075191b8d15ca2ba4bce8cc2c2bf64b Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 5 Aug 2026 13:59:03 +0200 Subject: [PATCH 01/16] feat(mo1_bragg): add direct ACS controller communication for scan settings --- debye_bec/devices/mo1_bragg/acs.py | 140 ++++++++++++++++++ debye_bec/devices/mo1_bragg/mo1_bragg.py | 15 +- .../devices/mo1_bragg/mo1_bragg_devices.py | 68 ++++++--- .../devices/mo1_bragg/mo1_bragg_enums.py | 7 + 4 files changed, 206 insertions(+), 24 deletions(-) create mode 100644 debye_bec/devices/mo1_bragg/acs.py diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py new file mode 100644 index 0000000..6eafa61 --- /dev/null +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -0,0 +1,140 @@ +""" +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 + - ophyd_devices.utils.socket.SocketIO -> raw socket helper + - ophyd_devices.utils.socket.SocketSignal -> Signal base talking through it + +Protocol: + read: "?GETVAR(tag)" -> reply is the value + write: "SETVAR(value,tag)" +""" + +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.socket import SocketSignal + +logger = bec_logger.logger + + +class ACSController(Controller): + """ + Shared TCP/IP communicator for one ACS controller. + + Instantiating this class twice with the same (socket_host, socket_port) + returns the same object (see `Controller.__new__`), so every variable + signal below -- across however many devices -- shares one connection. + """ + + _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, + socket_port=socket_port, + device_manager=device_manager, + term="\r", + trail=["\r:\r", ":\r"], + socket_timeout=0.1, + ) + + @threadlocked + def get_var(self, tag: int, prec: int, idx: int | None = None) -> float: + if self.sock is None: + self.on() + + idx = f",{idx:0.0f}" if idx is not None else "" + reply = self.socket_put_and_receive(f"?{{%0.{prec:0.0f}f}}GETVAR({tag}{idx})") + + if reply.startswith("?"): + error = self._query_error(reply) + raise RuntimeError(f"ACS error {reply}: {error}") + + return float(reply) + + @threadlocked + def _query_error(self, reply: str) -> str: + # reply is like "?2002" + return self.socket_put_and_receive(f"?{reply}") + + @threadlocked + def set_var(self, tag: int, value, prec: int, idx: int | None = None) -> None: + 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("?"): + error = self._query_error(reply) + raise RuntimeError(f"ACS error {reply}: {error}") + + +class AcsSignal(SocketSignal): + """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): + self.tag = tag + self.prec = prec + self.num_el = num_el + self.enum = enum + self.last_get = time.time() + 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 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 np.array( + [convert(self.controller.get_var(self.tag, self.prec, i)) for i in range(self.num_el)] + ) + + def _socket_set(self, val): + def convert(v): + if self.enum is None: + return v + if isinstance(v, str): + return self.enum[v].value # e.g. "SI111" -> 0 + return self.enum(v).value # e.g. 0 or Xtal.SI111 -> 0 + + if self.num_el <= 1: + self.controller.set_var(self.tag, convert(val), self.prec) + else: + if len(val) != self.num_el: + raise ValueError( + f"Length of val ({len(val)}) must be equal to specified length of variable ({self.num_el})" + ) + + for i, v in enumerate(val): + self.controller.set_var(self.tag, convert(v), self.prec, i) + + +class AcsSignalRO(AcsSignal): + """Readonly ACS controller variable, identified by its tag number.""" + + def _socket_set(self, val): + return diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg.py b/debye_bec/devices/mo1_bragg/mo1_bragg.py index 54a813f..dea3421 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg.py @@ -13,6 +13,7 @@ from typing import Literal from bec_lib.devicemanager import ScanInfo from bec_lib.logger import bec_logger +from bec_server.device_server.devices.devicemanager import DeviceManagerDS from bec_server.scan_server.scans.scan_base import ScanInfo as ScanServerScanInfo from ophyd import Component as Cpt from ophyd import DeviceStatus, StatusBase @@ -59,7 +60,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): USER_ACCESS = ["set_advanced_xas_settings", "set_xtal", "convert_angle_energy"] - def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, **kwargs): # type: ignore + def __init__(self, name: str, prefix: str = "", scan_info: ScanInfo | None = None, device_manager: DeviceManagerDS | None = None, **kwargs): # type: ignore """ Initialize the PSI Device Base class. @@ -67,7 +68,9 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): name (str) : Name of the device scan_info (ScanInfo): The scan info to use. """ - super().__init__(name=name, scan_info=scan_info, prefix=prefix, **kwargs) + super().__init__( + name=name, scan_info=scan_info, prefix=prefix, device_manager=device_manager, **kwargs + ) self.scan_parameters: ScanServerScanInfo = None self.timeout_for_pvwait = 7.5 self.valid_scan_names = [ @@ -346,7 +349,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): self.cancel_on_stop(status) logger.info(f"Finished calling complete on {self.name} within {time.time()-time_started}s.") return status - + def _status_callback(self, status, **kwargs): logger.info(f"Complete finished on mo1bragg with {status.done} and {status.success}") @@ -380,7 +383,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): if scan_parameters.scan_name in self.valid_scan_names: return True return False - + def _progress_update(self, value, old_value, **kwargs) -> None: """Callback method to update the scan progress, runs a callback to SUB_PROGRESS subscribers, i.e. BEC. @@ -449,13 +452,13 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): in_signal = self.calculator.calc_energy out_signal = self.calculator.calc_angle else: - raise Mo1BraggError(f'Unknown mode {mode}') + raise Mo1BraggError(f"Unknown mode {mode}") in_signal.put(inp) status = CompareStatus(self.calculator.calc_done, 1) self.cancel_on_stop(status) status.wait(self.timeout_for_pvwait) - status = CompareStatus(out_signal, 0, operation_success='>') + status = CompareStatus(out_signal, 0, operation_success=">") self.cancel_on_stop(status) status.wait(self.timeout_for_pvwait) return out_signal.get() diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index 3b6aca7..7ee3051 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -1,11 +1,14 @@ """Module for the Mo1 Bragg positioner""" +from __future__ import annotations + import threading import time import traceback -from typing import Literal +from typing import TYPE_CHECKING, Literal from bec_lib.logger import bec_logger +from bec_server.device_server.devices.devicemanager import DeviceManagerDS from ophyd import Component as Cpt from ophyd import ( Device, @@ -17,8 +20,11 @@ from ophyd import ( Signal, ) from ophyd.utils import LimitError +from ophyd_devices.utils.socket import SocketIO -from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType +# from debye_bec.devices.mo1_bragg.acs_controller import ACSSignal +from debye_bec.devices.mo1_bragg.acs import ACSController, AcsSignal, AcsSignalRO +from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType, Xtal # Initialise logger logger = bec_logger.logger @@ -39,9 +45,9 @@ class MoveTypeSignal(Signal): # pylint: disable=arguments-differ def set(self, value: str | MoveType) -> None: """Returns currently active move method - - Args: - value (str | MoveType) : Can be either 'energy' or 'angle' + auto_monitor=True + Args: + value (str | MoveType) : Can be either 'energy' or 'angle' """ value = MoveType(value.lower()) @@ -110,6 +116,10 @@ class Mo1BraggCrystal(Device): EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True ) + # current_xtal_string = Cpt( + # AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal", auto_monitor=True + # ) + class Mo1BraggScanSettings(Device): """Mo1 Bragg PVs to set the scan setttings""" @@ -128,20 +138,28 @@ class Mo1BraggScanSettings(Device): # XAS simple scan settings s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config") s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config") - s_scan_energy_lo = Cpt( - EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True - ) - s_scan_energy_hi = Cpt( - EpicsSignalWithRBV, suffix="s_scan_energy_hi", kind="config", auto_monitor=True - ) - s_scan_scantime = Cpt( - EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True - ) + # s_scan_energy_lo = Cpt( + # EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True + # ) + # s_scan_energy_hi = Cpt( + # EpicsSignalWithRBV, suffix="s_scan_energy_hi", kind="config", auto_monitor=True + # ) + # s_scan_scantime = Cpt( + # 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) # XAS advanced scan settings - a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=True) - a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=True) - a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=True) + a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False) + a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=False) + a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=False) + + # a_scan_pos = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="omitted") + # a_scan_vel = Cpt(AcsSignal, tag=53501, prec=6, num_el=41, kind="omitted") + # a_scan_time = Cpt(AcsSignal, tag=53502, prec=6, num_el=41, kind="omitted") class Mo1TriggerSettings(Device): @@ -256,6 +274,10 @@ class Mo1BraggPositioner(Device, PositionerBase): angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) + # test = Cpt(AcsSignal, tag=53000, prec=6, kind="normal") # s_scan_angle_low + # test2 = Cpt(AcsSignalRO, tag=12007, prec=6, kind="normal") # scan_msg + # test3 = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="normal") # a_scan_pos + ########## Move Command PVs ########## move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True) @@ -265,7 +287,7 @@ class Mo1BraggPositioner(Device, PositionerBase): _default_sub = SUB_READBACK SUB_PROGRESS = "progress" - def __init__(self, prefix="", *, name: str, **kwargs): + def __init__(self, prefix="", *, name: str, device_manager: DeviceManagerDS, **kwargs): """Initialize the Mo1 Bragg positioner. Args: @@ -273,11 +295,21 @@ class Mo1BraggPositioner(Device, PositionerBase): name (str): Name of the device kwargs: Additional keyword arguments """ + + host = "129.129.123.32" + port = 701 + self.controller = ACSController( + socket_cls=SocketIO, socket_host=host, socket_port=port, device_manager=device_manager + ) + super().__init__(prefix, name=name, **kwargs) self._move_thread = None self._stopped = False self.readback.name = self.name + # self.controller = ACSController(host, port) + # kwargs["controller"] = self.controller + def stop(self, *, success=False) -> None: """Stop any motion on the positioner diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py b/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py index 09602b7..44b6b47 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py @@ -3,6 +3,13 @@ import enum +class Xtal(int, enum.Enum): + """Enum class for the xtal (crystal) of the Bragg positioner""" + + Si111 = 0 + Si311 = 1 + + class TriggerControlSource(int, enum.Enum): """Enum class for the trigger control source of the trigger generator""" -- 2.54.0 From cdd208618ed5cbdc45ecbd7715220592e7d2acbe Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 5 Aug 2026 13:16:11 +0200 Subject: [PATCH 02/16] refactor(devices): vendor term/trail socket controller from ophyd_devices --- .../devices/utils/term_trail_controller.py | 270 ++++++++++++++++++ .../test_term_trail_controller.py | 150 ++++++++++ 2 files changed, 420 insertions(+) create mode 100644 debye_bec/devices/utils/term_trail_controller.py create mode 100644 tests/tests_devices/test_term_trail_controller.py diff --git a/debye_bec/devices/utils/term_trail_controller.py b/debye_bec/devices/utils/term_trail_controller.py new file mode 100644 index 0000000..fcbc50c --- /dev/null +++ b/debye_bec/devices/utils/term_trail_controller.py @@ -0,0 +1,270 @@ +"""Socket controller base with a configurable wire protocol. + +On top of the stock ``Controller`` this adds: + - configurable outgoing line termination (``term``) and reply terminators + (``trail``, a single string or several alternatives), + - a buffered ``socket_get`` that reassembles replies split across TCP reads and + rejects runaway replies (``max_reply_length``), + - a per-controller ``socket_timeout`` for individual send/recv operations, + - reconnect-before-retry on communication errors, discarding stale replies. +""" + +import functools +import time +import traceback + +from bec_lib import bec_logger +from ophyd_devices.utils.controller import Controller, ControllerCommunicationError, threadlocked + +logger = bec_logger.logger + + +def retry_once_reconnected(fcn): + """Decorator to rerun a function once if a communication error was raised. + + Reconnects first to discard any stale/desynced reply left on the wire from the + failed attempt -- without this, a late-arriving reply to the *first* attempt could + be misread as the reply to the retry. + """ + + @functools.wraps(fcn) + def wrapper(self, *args, **kwargs): + try: + val = fcn(self, *args, **kwargs) + except Exception: + content = traceback.format_exc() + logger.warning( + f"Communication error occurred. Reconnecting and retrying the command. Traceback: {content}" + ) + self._reconnect() + val = fcn(self, *args, **kwargs) + return val + + return wrapper + + +class TermTrailController(Controller): + """Socket controller with configurable term/trail strings and buffered replies. + + Args: + term (str, optional): Termination string appended to each outgoing socket request. + Defaults to the class attribute _term ("\\n"). + trail (str | list[str], optional): Termination string(s) marking the end of a + socket reply. Defaults to the class attribute _trail ("\\r\\n"). + max_reply_length (int, optional): Max accepted length of a socket reply in bytes. + Defaults to 1024 bytes. + socket_timeout (int | float, optional): Timeout for each socket operation in + seconds. Defaults to 2 seconds. + + Subclasses that need a different wire protocol should override the _term and _trail + class attributes. The constructor arguments only take effect on the first + instantiation per host:port; the controller is a singleton and later conflicting + values are ignored with a warning. + """ + + _term = "\n" # termination string appended to each outgoing request + _trail = "\r\n" # termination string(s) stripped from the end of each reply + _max_reply_length = 1024 # max accepted length of a socket reply in bytes + _socket_timeout = 2 # timeout for each socket operation in seconds + + def __init__( + self, + *, + socket_cls, + socket_host, + socket_port, + device_manager, + name: str = "", + attr_name="", + parent=None, + labels=None, + kind=None, + term: str | None = None, + trail: str | list[str] | tuple[str, ...] | None = None, + max_reply_length: int | None = None, + socket_timeout: int | float | None = None, + ): + if term is not None and not isinstance(term, str): + raise TypeError(f"term must be a string, got {type(term).__name__}") + if trail is not None: + if isinstance(trail, str): + pass + elif isinstance(trail, (list, tuple)) and all(isinstance(t, str) for t in trail): + if not trail: + raise ValueError("trail must not be an empty list/tuple.") + else: + raise TypeError( + f"trail must be a string or a list/tuple of strings, got {type(trail).__name__}" + ) + if max_reply_length is not None and not isinstance(max_reply_length, int): + raise TypeError( + f"max_reply_length must be an int, got {type(max_reply_length).__name__}" + ) + if socket_timeout is not None and not isinstance(socket_timeout, (int, float)): + raise TypeError(f"socket_timeout must be a number, got {type(socket_timeout).__name__}") + + first_init = not self._initialized + super().__init__( + socket_cls=socket_cls, + socket_host=socket_host, + socket_port=socket_port, + device_manager=device_manager, + name=name, + attr_name=attr_name, + parent=parent, + labels=labels, + kind=kind, + ) + if first_init: + if term is not None: + self._term = term + if trail is not None: + self._trail = trail + self._trail_options: tuple[str, ...] = ( + (self._trail,) if isinstance(self._trail, str) else tuple(self._trail) + ) + if max_reply_length is not None: + self._max_reply_length = max_reply_length + if socket_timeout is not None: + self._socket_timeout = socket_timeout + elif ( + (term is not None and term != self._term) + or (trail is not None and trail != self._trail) + or (max_reply_length is not None and max_reply_length != self._max_reply_length) + or (socket_timeout is not None and socket_timeout != self._socket_timeout) + ): + logger.warning( + f"Controller {self._socket_host}:{self._socket_port} is already initialized with " + f"term={self._term!r}, trail={self._trail!r}, max_reply_length={self._max_reply_length!r}, " + f"socket_timeout={self._socket_timeout!r}; ignoring conflicting values " + f"term={term!r}, trail={trail!r}, max_reply_length={max_reply_length!r}, " + f"socket_timeout={socket_timeout!r}." + ) + + @threadlocked + def _reconnect(self): + """ + Close and reopen the socket connection. + + Required after any communication error, in particular a recv() timeout: + this protocol has no per-message IDs, so there is no way to know whether + bytes that show up on the wire *after* a timeout belong to the request that + timed out or to whatever is sent next. A stale reply arriving late would + otherwise be read as the answer to a new command (or concatenated with it). + Closing and reopening the TCP connection discards any such reply-in-flight, + so the next command starts from a guaranteed-clean slate. + """ + try: + if self.sock is not None: + self.sock.close() + except Exception: + logger.warning("Error closing socket during reconnect.", exc_info=True) + finally: + self.sock = None + self.connected = False + self.on() + + @threadlocked + def socket_put(self, val: str): + """ + Send a command to the controller through the socket. + + Args: + val (str): Command to send + """ + self.command_history.append(f"[PUT]: time:{time.time()}, cmd:{val + self._term}") + self.sock.put(f"{val}{self._term}".encode()) + + @threadlocked + def socket_get(self): + """ + Receive a single, complete reply from the controller. + + Loops on `recv()` until any one of `self._trail_options` is seen, since a + reply can arrive split across multiple TCP reads, and different commands on + the same controller (e.g. ACS SETVAR vs GETVAR) can use different terminators. + `self._max_reply_length` guards against a malformed/runaway reply with no + matching trail. Does not protect against stale replies from a prior + timed-out request; that's handled by reconnecting the socket on + communication errors (see `_reconnect`). + + Returns: + str: The decoded reply, including its trailing terminator. + + Raises: + ControllerCommunicationError: If the connection closes while waiting + for a reply, or the reply exceeds `self._max_reply_length`. + """ + buf = b"" + while True: + for trail in self._trail_options: + if trail.encode() in buf: + response = buf.decode() + self.command_history.append(f"[GET]: time:{time.time()}, rep:{response}") + return response + + chunk = self.sock.receive() + if not chunk: + raise ControllerCommunicationError( + "Socket connection closed by remote host while waiting for a reply." + ) + buf += chunk + if len(buf) > self._max_reply_length: + raise ControllerCommunicationError( + f"Reply exceeded max_reply_length ({self._max_reply_length} bytes): {buf!r}" + ) + + @retry_once_reconnected + @threadlocked + def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str: + """ + Send a command to the controller and receive the response. + Override this method in the derived class if necessary, especially if the response + needs to be parsed differently. + """ + try: + self.socket_put(val) + if remove_trailing_chars: + return self._remove_trailing_characters(self.socket_get()) + return self.socket_get() + except Exception as exc: + logger.error( + f"Error in socket_put_and_receive: {exc}. Command history: {list(self.command_history)}" + ) + raise ControllerCommunicationError( + f"Failed to communicate with the controller. The last {self._command_history_length} commands were: " + f"{list(self.command_history)}" + ) from exc + + def _remove_trailing_characters(self, var) -> str: + """Strip whichever configured trail terminator is present at the end of a + reply; mid-reply occurrences are kept.""" + for trail in self._trail_options: + if var.endswith(trail): + return var.removesuffix(trail) + return var + + def on(self, timeout: int = 10) -> None: + """ + Open a new socket connection to the controller + + Args: + timeout (int): Time in seconds to wait for the connection itself to + be established (passed to `SocketIO.open`). This is separate from + `self._socket_timeout`, which governs how long each individual + send/recv call is allowed to take once connected. + """ + if not self.connected or self.sock is None: + try: + self.sock = self._socket_cls( + host=self._socket_host, + port=self._socket_port, + socket_timeout=self._socket_timeout, + ) + except TypeError: + # socket classes without a socket_timeout parameter (e.g. test mocks) + self.sock = self._socket_cls(host=self._socket_host, port=self._socket_port) + self.sock.open(timeout=timeout) + self.connected = True + else: + logger.info("The connection has already been established.") diff --git a/tests/tests_devices/test_term_trail_controller.py b/tests/tests_devices/test_term_trail_controller.py new file mode 100644 index 0000000..b40def4 --- /dev/null +++ b/tests/tests_devices/test_term_trail_controller.py @@ -0,0 +1,150 @@ +"""Tests for the vendored TermTrailController (term/trail wire-protocol support).""" + +from unittest import mock + +import pytest +from bec_server.device_server.tests.utils import DMMock +from ophyd_devices.tests.utils import SocketMock +from ophyd_devices.utils.controller import Controller, ControllerCommunicationError + +from debye_bec.devices.utils.term_trail_controller import TermTrailController + +# pylint: disable=protected-access + + +@pytest.fixture +def make_controller(): + """Factory fixture to build a fresh controller singleton with custom term/trail settings.""" + + def _make(controller_cls=TermTrailController, **kwargs): + Controller._reset_controller() + controller = controller_cls( + name="controller", + socket_cls=SocketMock, + socket_host="localhost", + socket_port=8080, + device_manager=DMMock(), + **kwargs, + ) + controller.on() + return controller + + yield _make + Controller._reset_controller() + + +def test_socket_put_appends_custom_term(make_controller): + controller = make_controller(term="\r") + controller.socket_put("get") + assert controller.sock.buffer_put == [b"get\r"] + + +@pytest.mark.parametrize( + ["trail", "reply", "expected"], + [ + (None, "value\r\n", "value"), + (None, "\r\n", ""), + (None, "line1\r\nline2\r\n", "line1\r\nline2"), + ("\n", "value\n", "value"), + ("\n", "\n", ""), + ("\n", "line1\nline2\n", "line1\nline2"), + ("", "value\r\n", "value\r\n"), + ], +) +def test_remove_trailing_characters_strips_suffix_only(make_controller, trail, reply, expected): + """The trail is stripped from the end of a reply only; mid-reply occurrences are preserved, + a bare terminator reduces to an empty payload, and an empty trail strips nothing.""" + kwargs = {} if trail is None else {"trail": trail} + controller = make_controller(**kwargs) + assert controller._remove_trailing_characters(reply) == expected + + +def test_multiple_trail_options_strip_whichever_matches(make_controller): + """With several trail alternatives (e.g. ACS GETVAR vs SETVAR replies), the one + that terminates the reply is stripped.""" + controller = make_controller(term="\r", trail=["\r:\r", ":\r"]) + controller.sock.buffer_recv = [b"8000.5\r:\r"] + assert controller.socket_put_and_receive("?GETVAR(1)") == "8000.5" + controller.sock.buffer_recv = [b":\r"] + assert controller.socket_put_and_receive("SETVAR(1,1)") == "" + + +def test_socket_get_reassembles_chunked_reply(make_controller): + controller = make_controller() + controller.sock.buffer_recv = [b"val", b"ue\r", b"\n"] + assert controller.socket_put_and_receive("get") == "value" + + +def test_socket_get_raises_on_runaway_reply(make_controller): + controller = make_controller(max_reply_length=8) + controller.sock.buffer_recv = [b"0123456789abcdef"] + with pytest.raises(ControllerCommunicationError): + controller.socket_put_and_receive("get") + + +def test_put_and_receive_reconnects_and_retries(make_controller): + """A communication error triggers exactly one reconnect, then the retry succeeds.""" + controller = make_controller() + controller.sock.buffer_recv = [b""] # remote closed the connection + + def _restock(): + controller.sock.buffer_recv = [b"value\r\n"] + + with mock.patch.object(controller, "_reconnect", side_effect=_restock) as mock_reconnect: + assert controller.socket_put_and_receive("get") == "value" + mock_reconnect.assert_called_once() + + +@pytest.mark.parametrize( + "kwargs", [{"term": b"\n"}, {"trail": b"\r\n"}, {"term": 13}, {"trail": 0}] +) +def test_term_and_trail_must_be_strings(make_controller, kwargs): + with pytest.raises(TypeError): + make_controller(**kwargs) + + +def test_second_construction_with_conflicting_term_trail_warns(make_controller): + """A second construction for the same host:port keeps the first term/trail and warns.""" + controller = make_controller(term="\r", trail="\n") + with mock.patch("debye_bec.devices.utils.term_trail_controller.logger") as mock_logger: + second = TermTrailController( + name="controller", + socket_cls=SocketMock, + socket_host="localhost", + socket_port=8080, + device_manager=DMMock(), + term="\n", + ) + assert second is controller + mock_logger.warning.assert_called_once() + assert controller._term == "\r" + assert controller._trail == "\n" + + +def test_second_construction_without_conflict_does_not_warn(make_controller): + controller = make_controller(term="\r") + with mock.patch("debye_bec.devices.utils.term_trail_controller.logger") as mock_logger: + for kwargs in ({"term": "\r"}, {}): + second = TermTrailController( + name="controller", + socket_cls=SocketMock, + socket_host="localhost", + socket_port=8080, + device_manager=DMMock(), + **kwargs, + ) + assert second is controller + mock_logger.warning.assert_not_called() + + +def test_subclass_overrides_term_trail_as_class_attributes(make_controller): + """Subclasses can set the wire protocol with class attributes, without any constructor plumbing.""" + + class CRTermController(TermTrailController): + _term = "\r" + _trail = "\n" + + controller = make_controller(controller_cls=CRTermController) + controller.sock.buffer_recv = [b"value\n"] + assert controller.socket_put_and_receive("get") == "value" + assert controller.sock.buffer_put == [b"get\r"] -- 2.54.0 From 5db31fb322dc9772cb10fc8c8b2633b42ab9d549 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 5 Aug 2026 13:16:11 +0200 Subject: [PATCH 03/16] 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): -- 2.54.0 From d92ebcfc18a9fce1aaa6c19be7f2082eab46bd95 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Thu, 6 Aug 2026 10:42:43 +0200 Subject: [PATCH 04/16] fix(mo1_bragg): omit ACS scan settings from the device configuration --- .../devices/mo1_bragg/mo1_bragg_devices.py | 12 +++++---- tests/tests_devices/test_mo1_bragg.py | 26 +++++++++---------- 2 files changed, 20 insertions(+), 18 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index 1da65a2..de68ee5 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -151,16 +151,18 @@ class Mo1BraggScanSettings(Device): # EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True # ) - # cache_ttl: baseline/config reads are served from cache; every put() drops the - # cache, so a fresh hardware read follows each real configuration change + # kind="omitted": these live on the ACS controller and are written by the scan at + # stage time, which also records them in scan_info — so nothing is lost by keeping + # them out of read_configuration(), and no passive device-server read path ever + # touches the socket. cache_ttl still guards explicit reads; put() drops the cache. s_scan_energy_lo = Cpt( - AcsSignal, tag=53003, prec=6, kind="config", auto_monitor=False, cache_ttl=30.0 + AcsSignal, tag=53003, prec=6, kind="omitted", 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 + AcsSignal, tag=53004, prec=6, kind="omitted", 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 + AcsSignal, tag=53002, prec=3, kind="omitted", auto_monitor=False, cache_ttl=30.0 ) # XAS advanced scan settings diff --git a/tests/tests_devices/test_mo1_bragg.py b/tests/tests_devices/test_mo1_bragg.py index aec8e45..869f2c8 100644 --- a/tests/tests_devices/test_mo1_bragg.py +++ b/tests/tests_devices/test_mo1_bragg.py @@ -136,15 +136,17 @@ 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): +def test_read_configuration_never_touches_acs_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.""" + signal updates; neither the ticking status PVs nor the ACS socket signals may be + part of it, so no passive read path ever produces a GETVAR.""" 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 + # ACS scan settings are omitted: config reads must not touch the socket at all + assert "bragg_scan_settings_s_scan_energy_lo" not in config + assert "bragg_scan_settings_s_scan_energy_hi" not in config + assert "bragg_scan_settings_s_scan_scantime" not in config + assert dev.controller.get_var.call_count == 0 # 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 @@ -153,15 +155,13 @@ def test_read_configuration_uses_acs_cache_and_omits_status_pvs(mock_bragg): 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 + # explicit reads still work, are cached, and refresh after a put + assert dev.scan_settings.s_scan_energy_lo.get() == 0.0 + dev.scan_settings.s_scan_energy_lo.get() + assert dev.controller.get_var.call_count == 1 # second read served from cache 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 + assert dev.controller.get_var.call_count == 2 # put dropped the cache def test_set_xas_settings(mock_bragg): -- 2.54.0 From 6e99e4d74e02b239e34b077c91d93e7df1fcaa87 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 5 Aug 2026 14:06:20 +0200 Subject: [PATCH 05/16] chore(mo1_bragg): temporary ACS GETVAR interval debug logging --- debye_bec/devices/mo1_bragg/acs.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index 28ffd23..e1b035a 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -115,6 +115,7 @@ class AcsSignal(SocketSignal): self.enum = enum self.cache_ttl = cache_ttl self._last_hw_read: float | None = None + self.last_get = time.time() # required by the interval debug log in _socket_get super().__init__(*args, **kwargs) @property @@ -170,6 +171,16 @@ class AcsSignal(SocketSignal): return old_value == new_value def _socket_get(self): + now = time.time() + interval = now - self.last_get + self.last_get = now + # "[acs-cache]" marks the NEW cached implementation in the logs; the old acs.py + # logs "Get signal with tag ..." on every read, so the prefix proves which code + # the device server actually imported + logger.info( + f"[acs-cache] hardware GETVAR tag {self.tag}, {interval * 1e3:.1f} ms since last read" + ) + def convert(val): return self.enum(val).name if self.enum is not None else val -- 2.54.0 From 5ae32d58d32bed4a45a46d8393a844b1c0557edd Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 07:49:47 +0200 Subject: [PATCH 06/16] Changing all signals to AcsSignals --- .../devices/mo1_bragg/mo1_bragg_angle.py | 14 +- .../devices/mo1_bragg/mo1_bragg_devices.py | 427 +++++++++++------- .../devices/mo1_bragg/mo1_bragg_enums.py | 14 + 3 files changed, 293 insertions(+), 162 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py b/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py index 46a095a..5e81bf7 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py @@ -3,16 +3,22 @@ from ophyd import Component as Cpt from ophyd import EpicsSignalRO, EpicsSignalWithRBV +from debye_bec.devices.mo1_bragg.acs import AcsSignal, AcsSignalRO from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggPositioner class Mo1BraggAngle(Mo1BraggPositioner): """Positioner implementation with readback angle of the MO1 Bragg positioner.""" - readback = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) - setpoint = Cpt(EpicsSignalWithRBV, suffix="set_abs_pos_angle", kind="normal", auto_monitor=True) - low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_angle_RBV", kind="config", auto_monitor=True) - high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_angle_RBV", kind="config", auto_monitor=True) + # readback = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) + # setpoint = Cpt(EpicsSignalWithRBV, suffix="set_abs_pos_angle", kind="normal", auto_monitor=True) + # low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_angle_RBV", kind="config", auto_monitor=True) + # high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_angle_RBV", kind="config", auto_monitor=True) + + readback = Cpt(AcsSignalRO, tag=51503, prec=6, kind="omitted") + setpoint = Cpt(AcsSignal, tag=51501, prec=6, kind="omitted") + low_lim = Cpt(AcsSignalRO, tag=51505, prec=1, kind="omitted") + high_lim = Cpt(AcsSignalRO, tag=51504, prec=1, kind="omitted") @property def egu(self) -> str: diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index de68ee5..487aa32 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -24,7 +24,17 @@ from ophyd_devices.utils.socket import SocketIO # from debye_bec.devices.mo1_bragg.acs_controller import ACSSignal from debye_bec.devices.mo1_bragg.acs import ACSController, AcsSignal, AcsSignalRO -from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType, Xtal +from debye_bec.devices.mo1_bragg.mo1_bragg_enums import ( + MoveType, + ScanControlLoadMessage, + ScanControlMode, + ScanControlScanStatus, + TriggerControlMode, + TriggerControlSource, + TriggerEnable, + TriggerSelectReference, + Xtal, +) # Initialise logger logger = bec_logger.logger @@ -60,87 +70,112 @@ class MoveTypeSignal(Signal): class Mo1BraggStatus(Device): """Mo1 Bragg PVs for status monitoring""" - error_status = Cpt(EpicsSignalRO, suffix="error_status_RBV", kind="config", auto_monitor=True) - brake_enabled = Cpt(EpicsSignalRO, suffix="brake_enabled_RBV", kind="config", auto_monitor=True) - mot_commutated = Cpt( - EpicsSignalRO, suffix="mot_commutated_RBV", kind="config", auto_monitor=True - ) - axis_enabled = Cpt(EpicsSignalRO, suffix="axis_enabled_RBV", kind="config", auto_monitor=True) - enc_initialized = Cpt( - EpicsSignalRO, suffix="enc_initialized_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) + # error_status = Cpt(EpicsSignalRO, suffix="error_status_RBV", kind="config", auto_monitor=True) + # brake_enabled = Cpt(EpicsSignalRO, suffix="brake_enabled_RBV", kind="config", auto_monitor=True) + # mot_commutated = Cpt( + # EpicsSignalRO, suffix="mot_commutated_RBV", kind="config", auto_monitor=True + # ) + # axis_enabled = Cpt(EpicsSignalRO, suffix="axis_enabled_RBV", kind="config", auto_monitor=True) + # enc_initialized = Cpt( + # EpicsSignalRO, suffix="enc_initialized_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) + + error_status = Cpt(AcsSignalRO, tag=10000, prec=0, kind="config") + brake_enabled = Cpt(AcsSignalRO, tag=10001, prec=0, kind="omitted") + mot_commutated = Cpt(AcsSignalRO, tag=10002, prec=0, kind="omitted") + axis_enabled = Cpt(AcsSignalRO, tag=10003, prec=0, kind="omitted") + enc_initialized = Cpt(AcsSignalRO, tag=10004, prec=0, kind="omitted") + heartbeat = Cpt(AcsSignalRO, tag=10005, prec=0, kind="omitted") class Mo1BraggEncoder(Device): """Mo1 Bragg PVs to communicate with the encoder""" - enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit", kind="config") - enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config") + # enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit", kind="config") + # enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config") + + enc_reinit = Cpt(AcsSignal, tag=11000, prec=0, kind="omitted") + enc_reinit_done = Cpt(AcsSignalRO, tag=11001, prec=0, kind="config") class Mo1BraggCrystal(Device): """Mo1 Bragg PVs to set the crystal parameters""" - bragg_off_si111 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si111", kind="config") - bragg_off_si311 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si311", kind="config") - phi_off_si111 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si111", kind="config") - phi_off_si311 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si311", kind="config") - azm_off_si111 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si111", kind="config") - azm_off_si311 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si311", kind="config") - miscut_si111 = Cpt(EpicsSignalWithRBV, suffix="miscut_si111", kind="config") - miscut_si311 = Cpt(EpicsSignalWithRBV, suffix="miscut_si311", kind="config") - xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config") - d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config") - d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config") - set_offset = Cpt(EpicsSignal, suffix="set_offset", kind="config", put_complete=True) - current_d_spacing = Cpt( - EpicsSignalRO, suffix="current_d_spacing_RBV", kind="normal", auto_monitor=True - ) - current_bragg_off = Cpt( - EpicsSignalRO, suffix="current_bragg_off_RBV", kind="normal", auto_monitor=True - ) - current_phi_off = Cpt( - EpicsSignalRO, suffix="current_phi_off_RBV", kind="normal", auto_monitor=True - ) - current_azm_off = Cpt( - EpicsSignalRO, suffix="current_azm_off_RBV", kind="normal", auto_monitor=True - ) - current_miscut = Cpt( - EpicsSignalRO, suffix="current_miscut_RBV", kind="normal", auto_monitor=True - ) - current_xtal = Cpt( - EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True - ) - current_xtal_string = Cpt( - EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True - ) - - # current_xtal_string = Cpt( - # AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal", auto_monitor=True + # bragg_off_si111 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si111", kind="config") + # bragg_off_si311 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si311", kind="config") + # phi_off_si111 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si111", kind="config") + # phi_off_si311 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si311", kind="config") + # azm_off_si111 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si111", kind="config") + # azm_off_si311 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si311", kind="config") + # miscut_si111 = Cpt(EpicsSignalWithRBV, suffix="miscut_si111", kind="config") + # miscut_si311 = Cpt(EpicsSignalWithRBV, suffix="miscut_si311", kind="config") + # xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config") + # d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config") + # d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config") + # set_offset = Cpt(EpicsSignal, suffix="set_offset", kind="config", put_complete=True) + # current_d_spacing = Cpt( + # EpicsSignalRO, suffix="current_d_spacing_RBV", kind="normal", auto_monitor=True # ) + # current_bragg_off = Cpt( + # EpicsSignalRO, suffix="current_bragg_off_RBV", kind="normal", auto_monitor=True + # ) + # current_phi_off = Cpt( + # EpicsSignalRO, suffix="current_phi_off_RBV", kind="normal", auto_monitor=True + # ) + # current_azm_off = Cpt( + # EpicsSignalRO, suffix="current_azm_off_RBV", kind="normal", auto_monitor=True + # ) + # current_miscut = Cpt( + # EpicsSignalRO, suffix="current_miscut_RBV", kind="normal", auto_monitor=True + # ) + # current_xtal = Cpt( + # EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True + # ) + # current_xtal_string = Cpt( + # EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True + # ) + + bragg_off_si111 = Cpt(AcsSignal, tag=50500, prec=12, kind="config") + bragg_off_si311 = Cpt(AcsSignal, tag=50501, prec=12, kind="config") + phi_off_si111 = Cpt(AcsSignal, tag=50507, prec=12, kind="config") + phi_off_si311 = Cpt(AcsSignal, tag=50508, prec=12, kind="config") + azm_off_si111 = Cpt(AcsSignal, tag=50510, prec=12, kind="config") + azm_off_si311 = Cpt(AcsSignal, tag=50511, prec=12, kind="config") + miscut_si111 = Cpt(AcsSignal, tag=50513, prec=12, kind="config") + miscut_si311 = Cpt(AcsSignal, tag=50514, prec=12, kind="config") + d_spacing_si111 = Cpt(AcsSignal, tag=50000, prec=12, kind="config") + d_spacing_si311 = Cpt(AcsSignal, tag=50001, prec=12, kind="config") + + current_d_spacing = Cpt(AcsSignalRO, tag=50503, prec=12, kind="config") + current_bragg_off = Cpt(AcsSignalRO, tag=50504, prec=12, kind="config") + current_phi_off = Cpt(AcsSignalRO, tag=50509, prec=12, kind="config") + current_azm_off = Cpt(AcsSignalRO, tag=50512, prec=12, kind="config") + current_miscut = Cpt(AcsSignalRO, tag=50515, prec=12, kind="config") + current_xtal = Cpt(AcsSignalRO, tag=10501, prec=0, kind="config") + current_xtal_string = Cpt(AcsSignalRO, tag=10501, prec=0, enum=Xtal, kind="normal") class Mo1BraggScanSettings(Device): """Mo1 Bragg PVs to set the scan setttings""" # TRIG settings - trig_select_ref_enum = Cpt(EpicsSignalWithRBV, suffix="trig_select_ref_ENUM", kind="config") + # trig_select_ref_enum = Cpt(EpicsSignalWithRBV, suffix="trig_select_ref_ENUM", kind="config") - trig_ena_hi_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_hi_ENUM", kind="config") - trig_time_hi = Cpt(EpicsSignalWithRBV, suffix="trig_time_hi", kind="config") - trig_every_n_hi = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_hi", kind="config") + # trig_ena_hi_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_hi_ENUM", kind="config") + # trig_time_hi = Cpt(EpicsSignalWithRBV, suffix="trig_time_hi", kind="config") + # trig_every_n_hi = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_hi", kind="config") - trig_ena_lo_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_lo_ENUM", kind="config") - trig_time_lo = Cpt(EpicsSignalWithRBV, suffix="trig_time_lo", kind="config") - trig_every_n_lo = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_lo", kind="config") + # trig_ena_lo_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_lo_ENUM", kind="config") + # trig_time_lo = Cpt(EpicsSignalWithRBV, suffix="trig_time_lo", kind="config") + # trig_every_n_lo = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_lo", kind="config") - # XAS simple scan settings - s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config") - s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config") + # # XAS simple scan settings + # s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config") + # s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config") # s_scan_energy_lo = Cpt( # EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True # ) @@ -151,110 +186,179 @@ class Mo1BraggScanSettings(Device): # EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True # ) + # XAS advanced scan settings + # a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False) + # a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=False) + # a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=False) + # kind="omitted": these live on the ACS controller and are written by the scan at # stage time, which also records them in scan_info — so nothing is lost by keeping # them out of read_configuration(), and no passive device-server read path ever # touches the socket. cache_ttl still guards explicit reads; put() drops the cache. - s_scan_energy_lo = Cpt( - AcsSignal, tag=53003, prec=6, kind="omitted", auto_monitor=False, cache_ttl=30.0 - ) - s_scan_energy_hi = Cpt( - AcsSignal, tag=53004, prec=6, kind="omitted", auto_monitor=False, cache_ttl=30.0 - ) - s_scan_scantime = Cpt( - AcsSignal, tag=53002, prec=3, kind="omitted", auto_monitor=False, cache_ttl=30.0 + + trig_select_ref_enum = Cpt( + AcsSignal, tag=12504, prec=0, enum=TriggerSelectReference, kind="config" ) - # XAS advanced scan settings - a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False) - a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=False) - a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=False) + trig_ena_hi_enum = Cpt(AcsSignal, tag=12501, prec=0, enum=TriggerEnable, kind="config") + trig_time_hi = Cpt(AcsSignal, tag=52501, prec=3, kind="config") + trig_every_n_hi = Cpt(AcsSignal, tag=12503, prec=0, kind="config") - # a_scan_pos = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="omitted") - # a_scan_vel = Cpt(AcsSignal, tag=53501, prec=6, num_el=41, kind="omitted") - # a_scan_time = Cpt(AcsSignal, tag=53502, prec=6, num_el=41, kind="omitted") + trig_ena_lo_enum = Cpt(AcsSignal, tag=12500, prec=0, enum=TriggerEnable, kind="config") + trig_time_lo = Cpt(AcsSignal, tag=52500, prec=3, kind="config") + trig_every_n_lo = Cpt(AcsSignal, tag=12502, prec=0, kind="config") + + s_scan_angle_hi = Cpt(AcsSignal, tag=53001, prec=6, kind="omitted") + s_scan_angle_lo = Cpt(AcsSignal, tag=53000, prec=6, kind="omitted") + + s_scan_energy_lo = Cpt(AcsSignal, tag=53003, prec=6, kind="config") + s_scan_energy_hi = Cpt(AcsSignal, tag=53004, prec=6, kind="config") + s_scan_scantime = Cpt(AcsSignal, tag=53002, prec=3, kind="config") + + a_scan_pos = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="config") + a_scan_vel = Cpt(AcsSignal, tag=53501, prec=6, num_el=41, kind="config") + a_scan_time = Cpt(AcsSignal, tag=53502, prec=6, num_el=41, kind="config") class Mo1TriggerSettings(Device): """Mo1 Trigger settings""" - settle_time = Cpt(EpicsSignalWithRBV, suffix="settle_time", kind="config") - max_dev = Cpt(EpicsSignalWithRBV, suffix="max_dev", kind="config") + # settle_time = Cpt(EpicsSignalWithRBV, suffix="settle_time", kind="config") + # max_dev = Cpt(EpicsSignalWithRBV, suffix="max_dev", kind="config") - xrd_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_src_ENUM", kind="config") - xrd_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_mode_ENUM", kind="config") - xrd_trig_len = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_len", kind="config") - xrd_trig_period = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_period", kind="config") - xrd_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="xrd_n_of_trig", kind="config") - xrd_trig_req = Cpt(EpicsSignal, suffix="xrd_trig_req", kind="config") + # xrd_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_src_ENUM", kind="config") + # xrd_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_mode_ENUM", kind="config") + # xrd_trig_len = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_len", kind="config") + # xrd_trig_period = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_period", kind="config") + # xrd_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="xrd_n_of_trig", kind="config") + # xrd_trig_req = Cpt(EpicsSignal, suffix="xrd_trig_req", kind="config") - falcon_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_src_ENUM", kind="config") - falcon_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_mode_ENUM", kind="config") - falcon_trig_len = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_len", kind="config") - falcon_trig_period = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_period", kind="config") - falcon_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="falcon_n_of_trig", kind="config") - falcon_trig_req = Cpt(EpicsSignal, suffix="falcon_trig_req", kind="config") + # falcon_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_src_ENUM", kind="config") + # falcon_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_mode_ENUM", kind="config") + # falcon_trig_len = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_len", kind="config") + # falcon_trig_period = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_period", kind="config") + # falcon_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="falcon_n_of_trig", kind="config") + # falcon_trig_req = Cpt(EpicsSignal, suffix="falcon_trig_req", kind="config") - univ1_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_src_ENUM", kind="config") - univ1_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_mode_ENUM", kind="config") - univ1_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_len", kind="config") - univ1_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_period", kind="config") - univ1_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ1_n_of_trig", kind="config") - univ1_trig_req = Cpt(EpicsSignal, suffix="univ1_trig_req", kind="config") + # univ1_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_src_ENUM", kind="config") + # univ1_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_mode_ENUM", kind="config") + # univ1_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_len", kind="config") + # univ1_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_period", kind="config") + # univ1_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ1_n_of_trig", kind="config") + # univ1_trig_req = Cpt(EpicsSignal, suffix="univ1_trig_req", kind="config") - univ2_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_src_ENUM", kind="config") - univ2_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_mode_ENUM", kind="config") - univ2_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_len", kind="config") - univ2_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_period", kind="config") - univ2_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ2_n_of_trig", kind="config") - univ2_trig_req = Cpt(EpicsSignal, suffix="univ2_trig_req", kind="config") + # univ2_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_src_ENUM", kind="config") + # univ2_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_mode_ENUM", kind="config") + # univ2_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_len", kind="config") + # univ2_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_period", kind="config") + # univ2_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ2_n_of_trig", kind="config") + # univ2_trig_req = Cpt(EpicsSignal, suffix="univ2_trig_req", kind="config") + + settle_time = Cpt(AcsSignal, tag=55000, prec=3, kind="config") + max_dev = Cpt(AcsSignal, tag=55001, prec=6, kind="config") + + xrd_trig_src_enum = Cpt(AcsSignal, tag=14501, prec=0, enum=TriggerControlSource, kind="config") + xrd_trig_mode_enum = Cpt(AcsSignal, tag=14502, prec=0, enum=TriggerControlMode, kind="config") + xrd_trig_len = Cpt(AcsSignal, tag=54500, prec=3, kind="config") + xrd_trig_period = Cpt(AcsSignal, tag=54504, prec=3, kind="config") + xrd_n_of_trig = Cpt(AcsSignal, tag=14512, prec=0, kind="config") + xrd_trig_req = Cpt(AcsSignal, tag=14500, prec=0, kind="config") + + falcon_trig_src_enum = Cpt( + AcsSignal, tag=14504, prec=0, enum=TriggerControlSource, kind="config" + ) + falcon_trig_mode_enum = Cpt( + AcsSignal, tag=14505, prec=0, enum=TriggerControlMode, kind="config" + ) + falcon_trig_len = Cpt(AcsSignal, tag=54501, prec=3, kind="config") + falcon_trig_period = Cpt(AcsSignal, tag=54505, prec=3, kind="config") + falcon_n_of_trig = Cpt(AcsSignal, tag=14513, prec=0, kind="config") + falcon_trig_req = Cpt(AcsSignal, tag=14503, prec=0, kind="config") + + univ1_trig_src_enum = Cpt( + AcsSignal, tag=14507, prec=0, enum=TriggerControlSource, kind="config" + ) + univ1_trig_mode_enum = Cpt(AcsSignal, tag=14508, prec=0, enum=TriggerControlMode, kind="config") + univ1_trig_len = Cpt(AcsSignal, tag=54502, prec=3, kind="config") + univ1_trig_period = Cpt(AcsSignal, tag=54506, prec=3, kind="config") + univ1_n_of_trig = Cpt(AcsSignal, tag=14514, prec=0, kind="config") + univ1_trig_req = Cpt(AcsSignal, tag=14506, prec=0, kind="config") + + univ2_trig_src_enum = Cpt( + AcsSignal, tag=14510, prec=0, enum=TriggerControlSource, kind="config" + ) + univ2_trig_mode_enum = Cpt(AcsSignal, tag=14511, prec=0, enum=TriggerControlMode, kind="config") + univ2_trig_len = Cpt(AcsSignal, tag=54503, prec=3, kind="config") + univ2_trig_period = Cpt(AcsSignal, tag=54507, prec=3, kind="config") + univ2_n_of_trig = Cpt(AcsSignal, tag=14515, prec=0, kind="config") + univ2_trig_req = Cpt(AcsSignal, tag=14509, prec=0, kind="config") class Mo1BraggCalculator(Device): """Mo1 Bragg PVs to convert angle to energy or vice-versa.""" - calc_reset = Cpt(EpicsSignalWithRBV, suffix="calc_reset", kind="config", put_complete=True) - calc_done = Cpt(EpicsSignalRO, suffix="calc_done_RBV", kind="config") - calc_energy = Cpt(EpicsSignalWithRBV, suffix="calc_energy", kind="config") - calc_angle = Cpt(EpicsSignalWithRBV, suffix="calc_angle", kind="config") + # calc_reset = Cpt(EpicsSignalWithRBV, suffix="calc_reset", kind="config", put_complete=True) + # calc_done = Cpt(EpicsSignalRO, suffix="calc_done_RBV", kind="config") + # calc_energy = Cpt(EpicsSignalWithRBV, suffix="calc_energy", kind="config") + # calc_angle = Cpt(EpicsSignalWithRBV, suffix="calc_angle", kind="config") + + calc_reset = Cpt(AcsSignal, tag=14000, prec=0, kind="omitted") + calc_done = Cpt(AcsSignalRO, tag=14001, prec=0, kind="omitted") + calc_energy = Cpt(AcsSignal, tag=54000, prec=0, kind="omitted") + calc_angle = Cpt(AcsSignal, tag=54001, prec=0, kind="omitted") class Mo1BraggScanControl(Device): """Mo1 Bragg PVs to control the scan after setting the parameters.""" - scan_mode_enum = Cpt(EpicsSignalWithRBV, suffix="scan_mode_ENUM", kind="config") - scan_duration = Cpt( - EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True - ) - scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=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="omitted", auto_monitor=True - ) - scan_time_left = Cpt( - EpicsSignalRO, suffix="scan_time_left_RBV", kind="omitted", 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="omitted", auto_monitor=True - ) - scan_spectra_done = Cpt( - EpicsSignalRO, suffix="scan_n_osc_RBV", kind="omitted", auto_monitor=True - ) - scan_spectra_left = Cpt( - EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="omitted", auto_monitor=True - ) + # scan_mode_enum = Cpt(EpicsSignalWithRBV, suffix="scan_mode_ENUM", kind="config") + # scan_duration = Cpt( + # EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True + # ) + # scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=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="omitted", auto_monitor=True + # ) + # scan_time_left = Cpt( + # EpicsSignalRO, suffix="scan_time_left_RBV", kind="omitted", 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="omitted", auto_monitor=True + # ) + # scan_spectra_done = Cpt( + # EpicsSignalRO, suffix="scan_n_osc_RBV", kind="omitted", auto_monitor=True + # ) + # scan_spectra_left = Cpt( + # EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="omitted", auto_monitor=True + # ) + + scan_mode_enum = Cpt(AcsSignal, tag=12000, prec=0, enum=ScanControlMode, kind="config") + scan_duration = Cpt(AcsSignal, tag=52000, prec=1, kind="config") + scan_load = Cpt(AcsSignal, tag=12001, prec=0, kind="omitted") + scan_msg = Cpt(AcsSignalRO, tag=12007, prec=0, enum=ScanControlLoadMessage, kind="config") + scan_start_infinite = Cpt(AcsSignal, tag=12003, prec=0, kind="omitted") + scan_start_timer = Cpt(AcsSignal, tag=12006, prec=0, kind="omitted") + scan_stop = Cpt(AcsSignal, tag=12004, prec=0, kind="omitted") + scan_status = Cpt(AcsSignalRO, tag=12002, prec=0, enum=ScanControlScanStatus, kind="config") + scan_time_left = Cpt(AcsSignalRO, tag=52001, prec=1, kind="omitted") + scan_done = Cpt(AcsSignalRO, tag=12005, prec=0, kind="omitted") + scan_val_reset = Cpt(AcsSignal, tag=52000, prec=0, kind="omitted") + scan_progress = Cpt(AcsSignalRO, tag=12011, prec=1, max_poll=0.5, kind="omitted") + scan_spectra_done = Cpt(AcsSignalRO, tag=12009, prec=0, kind="omitted") + scan_spectra_left = Cpt(AcsSignalRO, tag=12010, prec=0, kind="omitted") class Mo1BraggPositioner(Device, PositionerBase): @@ -279,29 +383,36 @@ class Mo1BraggPositioner(Device, PositionerBase): ############# Energy PVs ############# - readback = Cpt( - EpicsSignalRO, suffix="feedback_pos_energy_RBV", kind="hinted", auto_monitor=True - ) - setpoint = Cpt( - EpicsSignalWithRBV, suffix="set_abs_pos_energy", kind="normal", auto_monitor=True - ) - motor_is_moving = Cpt( - EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True - ) - low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_energy_RBV", kind="config", auto_monitor=True) - high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_energy_RBV", kind="config", auto_monitor=True) - velocity = Cpt(EpicsSignalWithRBV, suffix="move_velocity", kind="config", auto_monitor=True) + # readback = Cpt( + # EpicsSignalRO, suffix="feedback_pos_energy_RBV", kind="hinted", auto_monitor=True + # ) + # setpoint = Cpt( + # EpicsSignalWithRBV, suffix="set_abs_pos_energy", kind="normal", auto_monitor=True + # ) + # motor_is_moving = Cpt( + # EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True + # ) + # low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_energy_RBV", kind="config", auto_monitor=True) + # high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_energy_RBV", kind="config", auto_monitor=True) + # velocity = Cpt(EpicsSignalWithRBV, suffix="move_velocity", kind="config", auto_monitor=True) - angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) + # angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) - # test = Cpt(AcsSignal, tag=53000, prec=6, kind="normal") # s_scan_angle_low - # test2 = Cpt(AcsSignalRO, tag=12007, prec=6, kind="normal") # scan_msg - # test3 = Cpt(AcsSignal, tag=53500, prec=6, num_el=41, kind="normal") # a_scan_pos + readback = Cpt(AcsSignalRO, tag=51508, prec=3, kind="hinted") + setpoint = Cpt(AcsSignal, tag=51507, prec=3, kind="normal") + motor_is_moving = Cpt(AcsSignalRO, tag=11504, prec=0, max_poll=0.5, kind="normal") + low_lim = Cpt(AcsSignalRO, tag=51510, prec=3, kind="config") + high_lim = Cpt(AcsSignalRO, tag=51509, prec=3, kind="config") + velocity = Cpt(AcsSignal, tag=51502, prec=3, kind="config") + angle = Cpt(AcsSignalRO, tag=51503, prec=6, kind="normal") ########## Move Command PVs ########## - move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True) - move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config", put_complete=True) + # move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True) + # move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config", put_complete=True) + + move_abs = Cpt(AcsSignal, tag=11503, prec=0, kind="omitted") + move_stop = Cpt(AcsSignal, tag=11509, prec=0, kind="omitted") SUB_READBACK = "readback" _default_sub = SUB_READBACK diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py b/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py index 44b6b47..a89d83c 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_enums.py @@ -10,6 +10,20 @@ class Xtal(int, enum.Enum): Si311 = 1 +class TriggerSelectReference(int, enum.Enum): + """Enum class for the trigger reference selectrion""" + + ANGLE = 0 + ENERGY = 1 + + +class TriggerEnable(int, enum.Enum): + """Enum class to enable/disable trigger control""" + + DISABLED = 0 + ENABLED = 1 + + class TriggerControlSource(int, enum.Enum): """Enum class for the trigger control source of the trigger generator""" -- 2.54.0 From 83cd7e7e3f2b03bb1fd37467c52f29cc3a062805 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 07:50:59 +0200 Subject: [PATCH 07/16] Removing caching. Adding poll thread when subscribed --- debye_bec/devices/mo1_bragg/acs.py | 129 ++++++++++++----------------- 1 file changed, 53 insertions(+), 76 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index e1b035a..0fdb6f9 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -15,7 +15,9 @@ Protocol: from __future__ import annotations +import threading import time +import traceback from enum import Enum import numpy as np @@ -82,22 +84,7 @@ class ACSController(TermTrailController): class AcsSignal(SocketSignal): - """Read/write ACS controller variable, identified by its tag number. - - 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. - """ + """Read/write ACS controller variable, identified by its tag number.""" def __init__( self, @@ -106,86 +93,37 @@ class AcsSignal(SocketSignal): prec: int, num_el: int = 1, enum: Enum = None, - cache_ttl: float | None = None, + max_poll: float = 0.05, **kwargs, ): self.tag = tag self.prec = prec self.num_el = num_el self.enum = enum - self.cache_ttl = cache_ttl - self._last_hw_read: float | None = None - self.last_get = time.time() # required by the interval debug log in _socket_get + self.last_get = time.time() + self._poll_time = max_poll + self._poll_thread = None + self._stop_event = threading.Event() super().__init__(*args, **kwargs) @property def controller(self) -> ACSController: return self.root.controller - 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): now = time.time() interval = now - self.last_get self.last_get = now - # "[acs-cache]" marks the NEW cached implementation in the logs; the old acs.py - # logs "Get signal with tag ..." on every read, so the prefix proves which code - # the device server actually imported - logger.info( - f"[acs-cache] hardware GETVAR tag {self.tag}, {interval * 1e3:.1f} ms since last read" - ) + 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 convert(val): - return self.enum(val).name if self.enum is not None else val + return self.enum(val) if self.enum is not None else val if self.num_el <= 1: - return convert(self.controller.get_var(self.tag, self.prec)) + 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 np.array( [convert(self.controller.get_var(self.tag, self.prec, i)) for i in range(self.num_el)] ) @@ -209,6 +147,45 @@ class AcsSignal(SocketSignal): for i, v in enumerate(val): self.controller.set_var(self.tag, convert(v), self.prec, i) + def subscribe(self, callback, event_type=None, run=True): + logger.info(f"subscribe to signal {self.tag} called from: {traceback.format_stack()}") + self._ensure_polling() + cid = super().subscribe(callback, event_type=event_type, run=run) + return cid + + def clear_sub(self, cb, event_type=None): + super().clear_sub(cb, event_type=event_type) + if not any(self._callbacks.values()): + self._stop_polling() + + def _ensure_polling(self): + if self._poll_thread is None or not self._poll_thread.is_alive(): + self._stop_event.clear() + self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True) + self._poll_thread.start() + + def _stop_polling(self): + self._stop_event.set() + + def _poll_loop(self): + while not self._stop_event.is_set(): + old_value = self._readback + try: + new_value = self._socket_get() + except Exception as e: + logger.warning(f"During polling acs signal, got exception {e}") + time.sleep(self._poll_time) + continue + if new_value != old_value: + self._readback = new_value + self._run_subs( + sub_type=self.SUB_VALUE, + old_value=old_value, + value=new_value, + timestamp=time.time(), + ) + time.sleep(self._poll_time) + class AcsSignalRO(AcsSignal): """Readonly ACS controller variable, identified by its tag number.""" -- 2.54.0 From 60b20b3de6065339e04d77633a608f73b6f81683 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 07:51:24 +0200 Subject: [PATCH 08/16] Remove logging statements --- debye_bec/devices/mo1_bragg/acs.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index 0fdb6f9..aebb128 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -17,7 +17,8 @@ from __future__ import annotations import threading import time -import traceback + +# import traceback from enum import Enum import numpy as np @@ -114,7 +115,7 @@ class AcsSignal(SocketSignal): 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"Get signal with tag {self.tag}, time to last get: {interval*1e3} ms") # logger.info(f"socket_get called from: {traceback.format_stack()}") def convert(val): @@ -148,7 +149,7 @@ class AcsSignal(SocketSignal): self.controller.set_var(self.tag, convert(v), self.prec, i) def subscribe(self, callback, event_type=None, run=True): - logger.info(f"subscribe to signal {self.tag} called from: {traceback.format_stack()}") + # logger.info(f"subscribe to signal {self.tag} called from: {traceback.format_stack()}") self._ensure_polling() cid = super().subscribe(callback, event_type=event_type, run=run) return cid -- 2.54.0 From 72b7ae9c389336c5f833dd7064b9389cfe6edef5 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 07:53:17 +0200 Subject: [PATCH 09/16] Remove old EpicsSignals --- .../devices/mo1_bragg/mo1_bragg_angle.py | 6 - .../devices/mo1_bragg/mo1_bragg_devices.py | 186 +----------------- 2 files changed, 1 insertion(+), 191 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py b/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py index 5e81bf7..9722596 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_angle.py @@ -1,7 +1,6 @@ """Positioner implementation with readback angle of the MO1 Bragg positioner.""" from ophyd import Component as Cpt -from ophyd import EpicsSignalRO, EpicsSignalWithRBV from debye_bec.devices.mo1_bragg.acs import AcsSignal, AcsSignalRO from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggPositioner @@ -10,11 +9,6 @@ from debye_bec.devices.mo1_bragg.mo1_bragg_devices import Mo1BraggPositioner class Mo1BraggAngle(Mo1BraggPositioner): """Positioner implementation with readback angle of the MO1 Bragg positioner.""" - # readback = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) - # setpoint = Cpt(EpicsSignalWithRBV, suffix="set_abs_pos_angle", kind="normal", auto_monitor=True) - # low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_angle_RBV", kind="config", auto_monitor=True) - # high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_angle_RBV", kind="config", auto_monitor=True) - readback = Cpt(AcsSignalRO, tag=51503, prec=6, kind="omitted") setpoint = Cpt(AcsSignal, tag=51501, prec=6, kind="omitted") low_lim = Cpt(AcsSignalRO, tag=51505, prec=1, kind="omitted") diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index 487aa32..a65cc3d 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -10,15 +10,7 @@ from typing import TYPE_CHECKING, Literal from bec_lib.logger import bec_logger from bec_server.device_server.devices.devicemanager import DeviceManagerDS from ophyd import Component as Cpt -from ophyd import ( - Device, - DeviceStatus, - EpicsSignal, - EpicsSignalRO, - EpicsSignalWithRBV, - PositionerBase, - Signal, -) +from ophyd import Device, DeviceStatus, PositionerBase, Signal from ophyd.utils import LimitError from ophyd_devices.utils.socket import SocketIO @@ -70,20 +62,6 @@ class MoveTypeSignal(Signal): class Mo1BraggStatus(Device): """Mo1 Bragg PVs for status monitoring""" - # error_status = Cpt(EpicsSignalRO, suffix="error_status_RBV", kind="config", auto_monitor=True) - # brake_enabled = Cpt(EpicsSignalRO, suffix="brake_enabled_RBV", kind="config", auto_monitor=True) - # mot_commutated = Cpt( - # EpicsSignalRO, suffix="mot_commutated_RBV", kind="config", auto_monitor=True - # ) - # axis_enabled = Cpt(EpicsSignalRO, suffix="axis_enabled_RBV", kind="config", auto_monitor=True) - # enc_initialized = Cpt( - # EpicsSignalRO, suffix="enc_initialized_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) - error_status = Cpt(AcsSignalRO, tag=10000, prec=0, kind="config") brake_enabled = Cpt(AcsSignalRO, tag=10001, prec=0, kind="omitted") mot_commutated = Cpt(AcsSignalRO, tag=10002, prec=0, kind="omitted") @@ -95,9 +73,6 @@ class Mo1BraggStatus(Device): class Mo1BraggEncoder(Device): """Mo1 Bragg PVs to communicate with the encoder""" - # enc_reinit = Cpt(EpicsSignal, suffix="enc_reinit", kind="config") - # enc_reinit_done = Cpt(EpicsSignalRO, suffix="enc_reinit_done_RBV", kind="config") - enc_reinit = Cpt(AcsSignal, tag=11000, prec=0, kind="omitted") enc_reinit_done = Cpt(AcsSignalRO, tag=11001, prec=0, kind="config") @@ -105,40 +80,6 @@ class Mo1BraggEncoder(Device): class Mo1BraggCrystal(Device): """Mo1 Bragg PVs to set the crystal parameters""" - # bragg_off_si111 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si111", kind="config") - # bragg_off_si311 = Cpt(EpicsSignalWithRBV, suffix="bragg_off_si311", kind="config") - # phi_off_si111 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si111", kind="config") - # phi_off_si311 = Cpt(EpicsSignalWithRBV, suffix="phi_off_si311", kind="config") - # azm_off_si111 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si111", kind="config") - # azm_off_si311 = Cpt(EpicsSignalWithRBV, suffix="azm_off_si311", kind="config") - # miscut_si111 = Cpt(EpicsSignalWithRBV, suffix="miscut_si111", kind="config") - # miscut_si311 = Cpt(EpicsSignalWithRBV, suffix="miscut_si311", kind="config") - # xtal_enum = Cpt(EpicsSignalWithRBV, suffix="xtal_ENUM", kind="config") - # d_spacing_si111 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si111", kind="config") - # d_spacing_si311 = Cpt(EpicsSignalWithRBV, suffix="d_spacing_si311", kind="config") - # set_offset = Cpt(EpicsSignal, suffix="set_offset", kind="config", put_complete=True) - # current_d_spacing = Cpt( - # EpicsSignalRO, suffix="current_d_spacing_RBV", kind="normal", auto_monitor=True - # ) - # current_bragg_off = Cpt( - # EpicsSignalRO, suffix="current_bragg_off_RBV", kind="normal", auto_monitor=True - # ) - # current_phi_off = Cpt( - # EpicsSignalRO, suffix="current_phi_off_RBV", kind="normal", auto_monitor=True - # ) - # current_azm_off = Cpt( - # EpicsSignalRO, suffix="current_azm_off_RBV", kind="normal", auto_monitor=True - # ) - # current_miscut = Cpt( - # EpicsSignalRO, suffix="current_miscut_RBV", kind="normal", auto_monitor=True - # ) - # current_xtal = Cpt( - # EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True - # ) - # current_xtal_string = Cpt( - # EpicsSignalRO, suffix="current_xtal_ENUM_RBV", kind="normal", auto_monitor=True, string=True - # ) - bragg_off_si111 = Cpt(AcsSignal, tag=50500, prec=12, kind="config") bragg_off_si311 = Cpt(AcsSignal, tag=50501, prec=12, kind="config") phi_off_si111 = Cpt(AcsSignal, tag=50507, prec=12, kind="config") @@ -162,40 +103,6 @@ class Mo1BraggCrystal(Device): class Mo1BraggScanSettings(Device): """Mo1 Bragg PVs to set the scan setttings""" - # TRIG settings - # trig_select_ref_enum = Cpt(EpicsSignalWithRBV, suffix="trig_select_ref_ENUM", kind="config") - - # trig_ena_hi_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_hi_ENUM", kind="config") - # trig_time_hi = Cpt(EpicsSignalWithRBV, suffix="trig_time_hi", kind="config") - # trig_every_n_hi = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_hi", kind="config") - - # trig_ena_lo_enum = Cpt(EpicsSignalWithRBV, suffix="trig_ena_lo_ENUM", kind="config") - # trig_time_lo = Cpt(EpicsSignalWithRBV, suffix="trig_time_lo", kind="config") - # trig_every_n_lo = Cpt(EpicsSignalWithRBV, suffix="trig_every_n_lo", kind="config") - - # # XAS simple scan settings - # s_scan_angle_hi = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_hi", kind="config") - # s_scan_angle_lo = Cpt(EpicsSignalWithRBV, suffix="s_scan_angle_lo", kind="config") - # s_scan_energy_lo = Cpt( - # EpicsSignalWithRBV, suffix="s_scan_energy_lo", kind="config", auto_monitor=True - # ) - # s_scan_energy_hi = Cpt( - # EpicsSignalWithRBV, suffix="s_scan_energy_hi", kind="config", auto_monitor=True - # ) - # s_scan_scantime = Cpt( - # EpicsSignalWithRBV, suffix="s_scan_scantime", kind="config", auto_monitor=True - # ) - - # XAS advanced scan settings - # a_scan_pos = Cpt(EpicsSignalWithRBV, suffix="a_scan_pos", kind="config", auto_monitor=False) - # a_scan_vel = Cpt(EpicsSignalWithRBV, suffix="a_scan_vel", kind="config", auto_monitor=False) - # a_scan_time = Cpt(EpicsSignalWithRBV, suffix="a_scan_time", kind="config", auto_monitor=False) - - # kind="omitted": these live on the ACS controller and are written by the scan at - # stage time, which also records them in scan_info — so nothing is lost by keeping - # them out of read_configuration(), and no passive device-server read path ever - # touches the socket. cache_ttl still guards explicit reads; put() drops the cache. - trig_select_ref_enum = Cpt( AcsSignal, tag=12504, prec=0, enum=TriggerSelectReference, kind="config" ) @@ -223,37 +130,6 @@ class Mo1BraggScanSettings(Device): class Mo1TriggerSettings(Device): """Mo1 Trigger settings""" - # settle_time = Cpt(EpicsSignalWithRBV, suffix="settle_time", kind="config") - # max_dev = Cpt(EpicsSignalWithRBV, suffix="max_dev", kind="config") - - # xrd_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_src_ENUM", kind="config") - # xrd_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_mode_ENUM", kind="config") - # xrd_trig_len = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_len", kind="config") - # xrd_trig_period = Cpt(EpicsSignalWithRBV, suffix="xrd_trig_period", kind="config") - # xrd_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="xrd_n_of_trig", kind="config") - # xrd_trig_req = Cpt(EpicsSignal, suffix="xrd_trig_req", kind="config") - - # falcon_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_src_ENUM", kind="config") - # falcon_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_mode_ENUM", kind="config") - # falcon_trig_len = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_len", kind="config") - # falcon_trig_period = Cpt(EpicsSignalWithRBV, suffix="falcon_trig_period", kind="config") - # falcon_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="falcon_n_of_trig", kind="config") - # falcon_trig_req = Cpt(EpicsSignal, suffix="falcon_trig_req", kind="config") - - # univ1_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_src_ENUM", kind="config") - # univ1_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_mode_ENUM", kind="config") - # univ1_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_len", kind="config") - # univ1_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ1_trig_period", kind="config") - # univ1_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ1_n_of_trig", kind="config") - # univ1_trig_req = Cpt(EpicsSignal, suffix="univ1_trig_req", kind="config") - - # univ2_trig_src_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_src_ENUM", kind="config") - # univ2_trig_mode_enum = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_mode_ENUM", kind="config") - # univ2_trig_len = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_len", kind="config") - # univ2_trig_period = Cpt(EpicsSignalWithRBV, suffix="univ2_trig_period", kind="config") - # univ2_n_of_trig = Cpt(EpicsSignalWithRBV, suffix="univ2_n_of_trig", kind="config") - # univ2_trig_req = Cpt(EpicsSignal, suffix="univ2_trig_req", kind="config") - settle_time = Cpt(AcsSignal, tag=55000, prec=3, kind="config") max_dev = Cpt(AcsSignal, tag=55001, prec=6, kind="config") @@ -297,11 +173,6 @@ class Mo1TriggerSettings(Device): class Mo1BraggCalculator(Device): """Mo1 Bragg PVs to convert angle to energy or vice-versa.""" - # calc_reset = Cpt(EpicsSignalWithRBV, suffix="calc_reset", kind="config", put_complete=True) - # calc_done = Cpt(EpicsSignalRO, suffix="calc_done_RBV", kind="config") - # calc_energy = Cpt(EpicsSignalWithRBV, suffix="calc_energy", kind="config") - # calc_angle = Cpt(EpicsSignalWithRBV, suffix="calc_angle", kind="config") - calc_reset = Cpt(AcsSignal, tag=14000, prec=0, kind="omitted") calc_done = Cpt(AcsSignalRO, tag=14001, prec=0, kind="omitted") calc_energy = Cpt(AcsSignal, tag=54000, prec=0, kind="omitted") @@ -311,40 +182,6 @@ class Mo1BraggCalculator(Device): class Mo1BraggScanControl(Device): """Mo1 Bragg PVs to control the scan after setting the parameters.""" - # scan_mode_enum = Cpt(EpicsSignalWithRBV, suffix="scan_mode_ENUM", kind="config") - # scan_duration = Cpt( - # EpicsSignalWithRBV, suffix="scan_duration", kind="config", auto_monitor=True - # ) - # scan_load = Cpt(EpicsSignal, suffix="scan_load", kind="config", put_complete=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="omitted", auto_monitor=True - # ) - # scan_time_left = Cpt( - # EpicsSignalRO, suffix="scan_time_left_RBV", kind="omitted", 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="omitted", auto_monitor=True - # ) - # scan_spectra_done = Cpt( - # EpicsSignalRO, suffix="scan_n_osc_RBV", kind="omitted", auto_monitor=True - # ) - # scan_spectra_left = Cpt( - # EpicsSignalRO, suffix="scan_n_osc_left_RBV", kind="omitted", auto_monitor=True - # ) - scan_mode_enum = Cpt(AcsSignal, tag=12000, prec=0, enum=ScanControlMode, kind="config") scan_duration = Cpt(AcsSignal, tag=52000, prec=1, kind="config") scan_load = Cpt(AcsSignal, tag=12001, prec=0, kind="omitted") @@ -383,21 +220,6 @@ class Mo1BraggPositioner(Device, PositionerBase): ############# Energy PVs ############# - # readback = Cpt( - # EpicsSignalRO, suffix="feedback_pos_energy_RBV", kind="hinted", auto_monitor=True - # ) - # setpoint = Cpt( - # EpicsSignalWithRBV, suffix="set_abs_pos_energy", kind="normal", auto_monitor=True - # ) - # motor_is_moving = Cpt( - # EpicsSignalRO, suffix="move_abs_done_RBV", kind="normal", auto_monitor=True - # ) - # low_lim = Cpt(EpicsSignalRO, suffix="lo_lim_pos_energy_RBV", kind="config", auto_monitor=True) - # high_lim = Cpt(EpicsSignalRO, suffix="hi_lim_pos_energy_RBV", kind="config", auto_monitor=True) - # velocity = Cpt(EpicsSignalWithRBV, suffix="move_velocity", kind="config", auto_monitor=True) - - # angle = Cpt(EpicsSignalRO, suffix="feedback_pos_angle_RBV", kind="normal", auto_monitor=True) - readback = Cpt(AcsSignalRO, tag=51508, prec=3, kind="hinted") setpoint = Cpt(AcsSignal, tag=51507, prec=3, kind="normal") motor_is_moving = Cpt(AcsSignalRO, tag=11504, prec=0, max_poll=0.5, kind="normal") @@ -408,9 +230,6 @@ class Mo1BraggPositioner(Device, PositionerBase): ########## Move Command PVs ########## - # move_abs = Cpt(EpicsSignal, suffix="move_abs", kind="config", put_complete=True) - # move_stop = Cpt(EpicsSignal, suffix="move_stop", kind="config", put_complete=True) - move_abs = Cpt(AcsSignal, tag=11503, prec=0, kind="omitted") move_stop = Cpt(AcsSignal, tag=11509, prec=0, kind="omitted") @@ -438,9 +257,6 @@ class Mo1BraggPositioner(Device, PositionerBase): self._stopped = False self.readback.name = self.name - # self.controller = ACSController(host, port) - # kwargs["controller"] = self.controller - def stop(self, *, success=False) -> None: """Stop any motion on the positioner -- 2.54.0 From e1c7476e3762cde86a66a03f03233e12fa7f9e85 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 08:51:32 +0200 Subject: [PATCH 10/16] fix(widgets) due to new AcsSignals --- .../bec_widgets/widgets/digital_twin/digital_twin.py | 4 ++-- .../widgets/digital_twin/panels/input_panel.py | 2 +- .../widgets/scan_control_xas/scan_control_xas.py | 10 ++++++++++ 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py b/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py index f5fcbfe..15304bd 100644 --- a/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py +++ b/debye_bec/bec_widgets/widgets/digital_twin/digital_twin.py @@ -939,11 +939,11 @@ class DigitalTwin(BECWidget, QWidget): Calculates bragg angle in rad """ xtal = self.input.mo1_xtal.currentText() - if xtal == "Si(111)": + if xtal == "Si111": d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=True)[ "mo1_bragg_crystal_d_spacing_si111" ]["value"] - elif xtal == "Si(311)": + elif xtal == "Si311": d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si311.read(cached=True)[ "mo1_bragg_crystal_d_spacing_si311" ]["value"] diff --git a/debye_bec/bec_widgets/widgets/digital_twin/panels/input_panel.py b/debye_bec/bec_widgets/widgets/digital_twin/panels/input_panel.py index 8bebec6..b332f98 100644 --- a/debye_bec/bec_widgets/widgets/digital_twin/panels/input_panel.py +++ b/debye_bec/bec_widgets/widgets/digital_twin/panels/input_panel.py @@ -110,7 +110,7 @@ class InputPanel(QWidget): # Monochromator self.mo1_mode = ComboBox("mo1_mode", "Mode", ["Monochromatic", "Pinkbeam"]) - self.mo1_xtal = ComboBox("mo1_xtal", "Crystal", ["Si(111)", "Si(311)"]) + self.mo1_xtal = ComboBox("mo1_xtal", "Crystal", ["Si111", "Si311"]) self.mo1_bragg_angle = NumberIndicator("Bragg Angle", "deg", decimals=1) self.mo1_eres = NumberIndicator("Energy Resolution", "eV", decimals=2) self.mo1_ass_group = Group( diff --git a/debye_bec/bec_widgets/widgets/scan_control_xas/scan_control_xas.py b/debye_bec/bec_widgets/widgets/scan_control_xas/scan_control_xas.py index 995ea76..b785788 100644 --- a/debye_bec/bec_widgets/widgets/scan_control_xas/scan_control_xas.py +++ b/debye_bec/bec_widgets/widgets/scan_control_xas/scan_control_xas.py @@ -77,6 +77,15 @@ class ScanControlXAS(ScanControl): self._update_d_spacing, MessageEndpoints.device_readback("mo1_bragg") ) + # Read once manually to get a first value + if "mo1_bragg" in self.dev: + self.d_spacing = self.dev.mo1_bragg.crystal.d_spacing_si111.read(cached=True)[ + "mo1_bragg_crystal_d_spacing_si111" + ]["value"] + self.xas_scans_helper_widget.update_plot(d_spacing=self.d_spacing) + else: + logger.warning("mo1_bragg not in config, widget will not plot anything!") + @SafeSlot(dict, dict) def _update_d_spacing(self, msg: dict, _: dict): d_spacing = msg["signals"].get("mo1_bragg_crystal_current_d_spacing")["value"] @@ -293,6 +302,7 @@ class XASScansHelper(QWidget): if self.d_spacing is None: return if d_spacing == 0: + logger.warning("self.d_spacing is 0") return x_time = np.linspace(0, self.scan_parameters["scan_time"], PLOT_RESOLUTION) -- 2.54.0 From 92c57967db7ee5bc69b173eff7dfba7c98ed809d Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 08:51:54 +0200 Subject: [PATCH 11/16] fix(AcsSignals) for enums --- debye_bec/devices/mo1_bragg/acs.py | 4 +- .../devices/mo1_bragg/mo1_bragg_devices.py | 52 ++++++------------- 2 files changed, 18 insertions(+), 38 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index aebb128..f91195d 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -119,11 +119,11 @@ class AcsSignal(SocketSignal): # logger.info(f"socket_get called from: {traceback.format_stack()}") def convert(val): - return self.enum(val) if self.enum is not None else 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") + # logger.info(f"Got signal with tag {self.tag} and value {val}, time to last get: {interval*1e3} ms") return val 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 a65cc3d..4e02e50 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -16,17 +16,7 @@ from ophyd_devices.utils.socket import SocketIO # from debye_bec.devices.mo1_bragg.acs_controller import ACSSignal from debye_bec.devices.mo1_bragg.acs import ACSController, AcsSignal, AcsSignalRO -from debye_bec.devices.mo1_bragg.mo1_bragg_enums import ( - MoveType, - ScanControlLoadMessage, - ScanControlMode, - ScanControlScanStatus, - TriggerControlMode, - TriggerControlSource, - TriggerEnable, - TriggerSelectReference, - Xtal, -) +from debye_bec.devices.mo1_bragg.mo1_bragg_enums import MoveType, Xtal # Initialise logger logger = bec_logger.logger @@ -91,7 +81,7 @@ class Mo1BraggCrystal(Device): d_spacing_si111 = Cpt(AcsSignal, tag=50000, prec=12, kind="config") d_spacing_si311 = Cpt(AcsSignal, tag=50001, prec=12, kind="config") - current_d_spacing = Cpt(AcsSignalRO, tag=50503, prec=12, kind="config") + current_d_spacing = Cpt(AcsSignalRO, tag=50503, prec=12, kind="normal") current_bragg_off = Cpt(AcsSignalRO, tag=50504, prec=12, kind="config") current_phi_off = Cpt(AcsSignalRO, tag=50509, prec=12, kind="config") current_azm_off = Cpt(AcsSignalRO, tag=50512, prec=12, kind="config") @@ -103,15 +93,13 @@ class Mo1BraggCrystal(Device): class Mo1BraggScanSettings(Device): """Mo1 Bragg PVs to set the scan setttings""" - trig_select_ref_enum = Cpt( - AcsSignal, tag=12504, prec=0, enum=TriggerSelectReference, kind="config" - ) + trig_select_ref_enum = Cpt(AcsSignal, tag=12504, prec=0, kind="config") - trig_ena_hi_enum = Cpt(AcsSignal, tag=12501, prec=0, enum=TriggerEnable, kind="config") + trig_ena_hi_enum = Cpt(AcsSignal, tag=12501, prec=0, kind="config") trig_time_hi = Cpt(AcsSignal, tag=52501, prec=3, kind="config") trig_every_n_hi = Cpt(AcsSignal, tag=12503, prec=0, kind="config") - trig_ena_lo_enum = Cpt(AcsSignal, tag=12500, prec=0, enum=TriggerEnable, kind="config") + trig_ena_lo_enum = Cpt(AcsSignal, tag=12500, prec=0, kind="config") trig_time_lo = Cpt(AcsSignal, tag=52500, prec=3, kind="config") trig_every_n_lo = Cpt(AcsSignal, tag=12502, prec=0, kind="config") @@ -133,37 +121,29 @@ class Mo1TriggerSettings(Device): settle_time = Cpt(AcsSignal, tag=55000, prec=3, kind="config") max_dev = Cpt(AcsSignal, tag=55001, prec=6, kind="config") - xrd_trig_src_enum = Cpt(AcsSignal, tag=14501, prec=0, enum=TriggerControlSource, kind="config") - xrd_trig_mode_enum = Cpt(AcsSignal, tag=14502, prec=0, enum=TriggerControlMode, kind="config") + xrd_trig_src_enum = Cpt(AcsSignal, tag=14501, prec=0, kind="config") + xrd_trig_mode_enum = Cpt(AcsSignal, tag=14502, prec=0, kind="config") xrd_trig_len = Cpt(AcsSignal, tag=54500, prec=3, kind="config") xrd_trig_period = Cpt(AcsSignal, tag=54504, prec=3, kind="config") xrd_n_of_trig = Cpt(AcsSignal, tag=14512, prec=0, kind="config") xrd_trig_req = Cpt(AcsSignal, tag=14500, prec=0, kind="config") - falcon_trig_src_enum = Cpt( - AcsSignal, tag=14504, prec=0, enum=TriggerControlSource, kind="config" - ) - falcon_trig_mode_enum = Cpt( - AcsSignal, tag=14505, prec=0, enum=TriggerControlMode, kind="config" - ) + falcon_trig_src_enum = Cpt(AcsSignal, tag=14504, prec=0, kind="config") + falcon_trig_mode_enum = Cpt(AcsSignal, tag=14505, prec=0, kind="config") falcon_trig_len = Cpt(AcsSignal, tag=54501, prec=3, kind="config") falcon_trig_period = Cpt(AcsSignal, tag=54505, prec=3, kind="config") falcon_n_of_trig = Cpt(AcsSignal, tag=14513, prec=0, kind="config") falcon_trig_req = Cpt(AcsSignal, tag=14503, prec=0, kind="config") - univ1_trig_src_enum = Cpt( - AcsSignal, tag=14507, prec=0, enum=TriggerControlSource, kind="config" - ) - univ1_trig_mode_enum = Cpt(AcsSignal, tag=14508, prec=0, enum=TriggerControlMode, kind="config") + univ1_trig_src_enum = Cpt(AcsSignal, tag=14507, prec=0, kind="config") + univ1_trig_mode_enum = Cpt(AcsSignal, tag=14508, prec=0, kind="config") univ1_trig_len = Cpt(AcsSignal, tag=54502, prec=3, kind="config") univ1_trig_period = Cpt(AcsSignal, tag=54506, prec=3, kind="config") univ1_n_of_trig = Cpt(AcsSignal, tag=14514, prec=0, kind="config") univ1_trig_req = Cpt(AcsSignal, tag=14506, prec=0, kind="config") - univ2_trig_src_enum = Cpt( - AcsSignal, tag=14510, prec=0, enum=TriggerControlSource, kind="config" - ) - univ2_trig_mode_enum = Cpt(AcsSignal, tag=14511, prec=0, enum=TriggerControlMode, kind="config") + univ2_trig_src_enum = Cpt(AcsSignal, tag=14510, prec=0, kind="config") + univ2_trig_mode_enum = Cpt(AcsSignal, tag=14511, prec=0, kind="config") univ2_trig_len = Cpt(AcsSignal, tag=54503, prec=3, kind="config") univ2_trig_period = Cpt(AcsSignal, tag=54507, prec=3, kind="config") univ2_n_of_trig = Cpt(AcsSignal, tag=14515, prec=0, kind="config") @@ -182,14 +162,14 @@ class Mo1BraggCalculator(Device): class Mo1BraggScanControl(Device): """Mo1 Bragg PVs to control the scan after setting the parameters.""" - scan_mode_enum = Cpt(AcsSignal, tag=12000, prec=0, enum=ScanControlMode, kind="config") + scan_mode_enum = Cpt(AcsSignal, tag=12000, prec=0, kind="config") scan_duration = Cpt(AcsSignal, tag=52000, prec=1, kind="config") scan_load = Cpt(AcsSignal, tag=12001, prec=0, kind="omitted") - scan_msg = Cpt(AcsSignalRO, tag=12007, prec=0, enum=ScanControlLoadMessage, kind="config") + scan_msg = Cpt(AcsSignalRO, tag=12007, prec=0, kind="config") scan_start_infinite = Cpt(AcsSignal, tag=12003, prec=0, kind="omitted") scan_start_timer = Cpt(AcsSignal, tag=12006, prec=0, kind="omitted") scan_stop = Cpt(AcsSignal, tag=12004, prec=0, kind="omitted") - scan_status = Cpt(AcsSignalRO, tag=12002, prec=0, enum=ScanControlScanStatus, kind="config") + scan_status = Cpt(AcsSignalRO, tag=12002, prec=0, kind="config") scan_time_left = Cpt(AcsSignalRO, tag=52001, prec=1, kind="omitted") scan_done = Cpt(AcsSignalRO, tag=12005, prec=0, kind="omitted") scan_val_reset = Cpt(AcsSignal, tag=52000, prec=0, kind="omitted") -- 2.54.0 From 76d4de133678ee615d49cc361767f689c0dd64c2 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 09:44:33 +0200 Subject: [PATCH 12/16] improve stage speed --- debye_bec/devices/mo1_bragg/mo1_bragg.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg.py b/debye_bec/devices/mo1_bragg/mo1_bragg.py index dea3421..dd451d0 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg.py @@ -81,6 +81,8 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): "nidaq_continuous_scan", ] + self.stage_start = None + ######################################## # Beamline Specific Implementations # ######################################## @@ -106,6 +108,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): Information about the upcoming scan can be accessed from the scan_info (self.scan_info.msg) object. """ + self.stage_start = time.time() self.scan_parameters = fetch_scan_info(self.scan_info) if self.scan_control.scan_msg.get() != ScanControlLoadMessage.PENDING: status = CompareStatus(self.scan_control.scan_msg, ScanControlLoadMessage.PENDING) @@ -290,9 +293,9 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): else: return # Setting scan duration seems to lag behind slightly in the backend, include small sleep - logger.info(f"Sleeping for one second") - time.sleep(1) - logger.info(f"Device {self.name}, done sleeping") + # logger.info(f"Sleeping for one second") + # time.sleep(1) + # logger.info(f"Device {self.name}, done sleeping") # Load the scan parameters to the controller status = CompareStatus( self.scan_control.scan_msg, @@ -303,6 +306,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): self.scan_control.scan_load.put(1) # Wait for params to be checked from controller status.wait(self.timeout_for_pvwait) + # logger.info(f"Starting scan took {time.time() - self.stage_start} s") return None def on_unstage(self) -> DeviceStatus | StatusBase | None: -- 2.54.0 From 6771d338d011c5a8d36e207888c1fd02a40571ee Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 09:45:02 +0200 Subject: [PATCH 13/16] fix(AcsSignal) force update before subscribing --- debye_bec/devices/mo1_bragg/acs.py | 33 ++++++++++++++++++------------ 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index f91195d..c8ac44a 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -149,10 +149,22 @@ class AcsSignal(SocketSignal): self.controller.set_var(self.tag, convert(v), self.prec, i) def subscribe(self, callback, event_type=None, run=True): - # logger.info(f"subscribe to signal {self.tag} called from: {traceback.format_stack()}") self._ensure_polling() - cid = super().subscribe(callback, event_type=event_type, run=run) - return cid + if run: + self._force_fresh_read() + return super().subscribe(callback, event_type=event_type, run=run) + + def _force_fresh_read(self): + old_value = self._readback + try: + new_value = self._socket_get() + except Exception as e: + logger.warning(f"Fresh read failed for {self.name} during subscribe() with {e}") + return + self._readback = new_value + self._run_subs( + sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time() + ) def clear_sub(self, cb, event_type=None): super().clear_sub(cb, event_type=event_type) @@ -173,18 +185,13 @@ class AcsSignal(SocketSignal): old_value = self._readback try: new_value = self._socket_get() - except Exception as e: - logger.warning(f"During polling acs signal, got exception {e}") + except Exception: time.sleep(self._poll_time) continue - if new_value != old_value: - self._readback = new_value - self._run_subs( - sub_type=self.SUB_VALUE, - old_value=old_value, - value=new_value, - timestamp=time.time(), - ) + self._readback = new_value + self._run_subs( + sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time() + ) time.sleep(self._poll_time) -- 2.54.0 From 1b168c088cce48df039e34fc85ac52fb2fedf263 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 14:55:23 +0200 Subject: [PATCH 14/16] fix(mo1_bragg): Set scan angles in advanced scan to prevent bug in NIDAQ --- debye_bec/devices/mo1_bragg/mo1_bragg.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg.py b/debye_bec/devices/mo1_bragg/mo1_bragg.py index dd451d0..b31f65f 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg.py @@ -465,6 +465,7 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): status = CompareStatus(out_signal, 0, operation_success=">") self.cancel_on_stop(status) status.wait(self.timeout_for_pvwait) + # logger.info(f"Converted energy {inp} to angle {out_signal.get()}") return out_signal.get() def set_advanced_xas_settings( @@ -494,6 +495,15 @@ class Mo1Bragg(PSIDeviceBase, Mo1BraggPositioner): status_list = [] + # Even though s_scan_angle_lo and -_hi are not used by the mono + # for the advanced scan, we set it because the NIDAQ will read + # those signals to calculate the expected number of encoder steps + status_list.append(self.scan_settings.s_scan_angle_lo.set(pos[0])) + self.cancel_on_stop(status_list[-1]) + + status_list.append(self.scan_settings.s_scan_angle_hi.set(pos[-1])) + self.cancel_on_stop(status_list[-1]) + status_list.append(self.scan_settings.a_scan_pos.set(pos)) self.cancel_on_stop(status_list[-1]) -- 2.54.0 From b076d446c6eb488c6fdd50d0b597b6a272ef99d4 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 14:55:44 +0200 Subject: [PATCH 15/16] fix(mo1_bragg): Change prec for calc signals --- debye_bec/devices/mo1_bragg/mo1_bragg_devices.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py index 4e02e50..c62fdb3 100644 --- a/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py +++ b/debye_bec/devices/mo1_bragg/mo1_bragg_devices.py @@ -155,8 +155,8 @@ class Mo1BraggCalculator(Device): calc_reset = Cpt(AcsSignal, tag=14000, prec=0, kind="omitted") calc_done = Cpt(AcsSignalRO, tag=14001, prec=0, kind="omitted") - calc_energy = Cpt(AcsSignal, tag=54000, prec=0, kind="omitted") - calc_angle = Cpt(AcsSignal, tag=54001, prec=0, kind="omitted") + calc_energy = Cpt(AcsSignal, tag=54000, prec=6, kind="omitted") + calc_angle = Cpt(AcsSignal, tag=54001, prec=6, kind="omitted") class Mo1BraggScanControl(Device): -- 2.54.0 From 9a98e3fdf344cdccaa4b0b9625143341bd15e9fe Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 14:57:45 +0200 Subject: [PATCH 16/16] fix(AcsSignals): Various bugfixes for edge cases --- debye_bec/devices/mo1_bragg/acs.py | 110 ++++++++++++++++++++++------- 1 file changed, 84 insertions(+), 26 deletions(-) diff --git a/debye_bec/devices/mo1_bragg/acs.py b/debye_bec/devices/mo1_bragg/acs.py index c8ac44a..7575d6c 100644 --- a/debye_bec/devices/mo1_bragg/acs.py +++ b/debye_bec/devices/mo1_bragg/acs.py @@ -24,6 +24,7 @@ from enum import Enum import numpy as np from bec_lib.logger import bec_logger from ophyd_devices.utils.controller import threadlocked +from ophyd_devices.utils.psi_device_base_utils import CompareStatus, SubscriptionStatus from ophyd_devices.utils.socket import SocketSignal from debye_bec.devices.utils.term_trail_controller import TermTrailController @@ -104,7 +105,9 @@ class AcsSignal(SocketSignal): self.last_get = time.time() self._poll_time = max_poll self._poll_thread = None - self._stop_event = threading.Event() + self._poll_stop_event = None + self._lifecycle_lock = threading.Lock() + self._sub_count = 0 super().__init__(*args, **kwargs) @property @@ -123,7 +126,9 @@ class AcsSignal(SocketSignal): if self.num_el <= 1: val = convert(self.controller.get_var(self.tag, self.prec)) - # logger.info(f"Got signal with tag {self.tag} and value {val}, time to last get: {interval*1e3} ms") + # logger.info( + # f"Got signal with tag {self.tag} and value {val}, time to last get: {interval*1e3} ms" + # ) return val return np.array( [convert(self.controller.get_var(self.tag, self.prec, i)) for i in range(self.num_el)] @@ -148,40 +153,81 @@ class AcsSignal(SocketSignal): for i, v in enumerate(val): self.controller.set_var(self.tag, convert(v), self.prec, i) + def set(self, value, *, timeout=None, settle_time=None, **kwargs): + """ + Write value to the controller and return a Status that finishes once + the readback confirms the new value (via CompareStatus + polling). + """ + if self.num_el > 1: + status = self.array_compare_status(value, atol=10 ** (-self.prec), timeout=timeout) + else: + status = CompareStatus(self, value, timeout=timeout, settle_time=settle_time or 0) + try: + self.put(value, **kwargs) + except Exception as exc: + status.set_exception(exc) + return status + + def array_compare_status(self, value, *, atol=None, rtol=None, timeout=None, settle_time=0): + """CompareStatus equivalent for array-valued (num_el > 1) signals.""" + target = np.asarray(value) + + def _compare(value, **kwargs): + current = np.asarray(value) + if atol is not None or rtol is not None: + return bool(np.allclose(current, target, atol=atol or 0, rtol=rtol or 0)) + return bool(np.array_equal(current, target)) + + return SubscriptionStatus(self, _compare, timeout=timeout, settle_time=settle_time) + def subscribe(self, callback, event_type=None, run=True): - self._ensure_polling() + with self._lifecycle_lock: + self._sub_count += 1 + self._ensure_polling_locked() if run: self._force_fresh_read() return super().subscribe(callback, event_type=event_type, run=run) - def _force_fresh_read(self): - old_value = self._readback - try: - new_value = self._socket_get() - except Exception as e: - logger.warning(f"Fresh read failed for {self.name} during subscribe() with {e}") - return - self._readback = new_value - self._run_subs( - sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time() - ) - def clear_sub(self, cb, event_type=None): + before = sum(len(d) for d in self._callbacks.values()) super().clear_sub(cb, event_type=event_type) - if not any(self._callbacks.values()): - self._stop_polling() + removed = before - sum(len(d) for d in self._callbacks.values()) + if not removed: + return + with self._lifecycle_lock: + self._sub_count = max(0, self._sub_count - removed) + if self._sub_count == 0: + self._stop_polling_locked() - def _ensure_polling(self): - if self._poll_thread is None or not self._poll_thread.is_alive(): - self._stop_event.clear() - self._poll_thread = threading.Thread(target=self._poll_loop, daemon=True) - self._poll_thread.start() + def _ensure_polling_locked(self): + # Called with _lifecycle_lock held. + # A thread only "covers us" if it's alive AND its own stop + # event hasn't already been set — an alive-but-doomed thread + # (mid-sleep, about to notice a stop request) must NOT block a + # fresh thread from starting. + if ( + self._poll_thread is not None + and self._poll_thread.is_alive() + and self._poll_stop_event is not None + and not self._poll_stop_event.is_set() + ): + return + stop_event = threading.Event() + self._poll_stop_event = stop_event + self._poll_thread = threading.Thread( + target=self._poll_loop, args=(stop_event,), daemon=True + ) + # logger.info(f"Start poll thread for tag {self.tag}") + self._poll_thread.start() - def _stop_polling(self): - self._stop_event.set() + def _stop_polling_locked(self): + # Called with _lifecycle_lock held. + if self._poll_stop_event is not None: + self._poll_stop_event.set() + # logger.info(f"Stop poll thread for tag {self.tag}") - def _poll_loop(self): - while not self._stop_event.is_set(): + def _poll_loop(self, stop_event): + while not stop_event.is_set(): old_value = self._readback try: new_value = self._socket_get() @@ -194,6 +240,18 @@ class AcsSignal(SocketSignal): ) time.sleep(self._poll_time) + def _force_fresh_read(self): + old_value = self._readback + try: + new_value = self._socket_get() + except Exception: + logger.exception(f"Fresh read failed for {self.name} during subscribe()") + return + self._readback = new_value + self._run_subs( + sub_type=self.SUB_VALUE, old_value=old_value, value=new_value, timestamp=time.time() + ) + class AcsSignalRO(AcsSignal): """Readonly ACS controller variable, identified by its tag number.""" -- 2.54.0