From 83cd7e7e3f2b03bb1fd37467c52f29cc3a062805 Mon Sep 17 00:00:00 2001 From: x01da Date: Tue, 18 Aug 2026 07:50:59 +0200 Subject: [PATCH] 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."""