Removing caching. Adding poll thread when subscribed
This commit is contained in:
x01da
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user