fix: throttle signal update to redis

xtreme_bec/devices/x07ma_devices.py

@@ -8,6 +8,7 @@ import traceback
 from collections import OrderedDict
 from typing import Any
 
+import numpy as np
 from bec_lib import bec_logger
 from ophyd import Component as Cpt
 from ophyd import Device, EpicsMotor, EpicsSignal, EpicsSignalRO
@@ -357,36 +358,19 @@ class X07MAAnalogSignals(PSIDeviceBase, Device):
     norm_diode = Cpt(NormDIODESignals, name="norm_diode", kind=Kind.hinted)
 
     def on_connected(self) -> None:
-        self._broadcast_lock = threading.Lock()
         self._last_broadcast_ts = 0
-        self._broadcast_rate_limit_s = 0.1  # seconds
-        self._pending_broadcast = None
+        self._broadcast_rate_limit_s = 0.01  # seconds
         for signal_name in ("s1", "s2", "s3", "s4", "s5", "s6", "s7"):
             getattr(self, signal_name).subscribe(self._schedule_broadcast, run=False)
 
-    def _schedule_broadcast(self, *args, **kwargs):
-        with self._broadcast_lock:
-            now = time.monotonic()
-            remaining = self._broadcast_rate_limit_s - (now - self._last_broadcast_ts)
-            if remaining <= 0 and self._pending_broadcast is None:
-                self._last_broadcast_ts = now
-                emit_now = True
-            else:
-                emit_now = False
-                if self._pending_broadcast is None:
-                    self._pending_broadcast = threading.Timer(remaining, self._emit_broadcast)
-                    self._pending_broadcast.daemon = True
-                    self._pending_broadcast.start()
-
-        if emit_now:
-            self._run_subs(sub_type=self.SUB_VALUE, value=self.read())
-
-    def _emit_broadcast(self):
-        with self._broadcast_lock:
-            self._pending_broadcast = None
-            self._last_broadcast_ts = time.monotonic()
-
-        self._run_subs(sub_type=self.SUB_VALUE, value=self.read())
+    def _schedule_broadcast(self, *args, old_value, value, **kwargs):
+        if np.isclose(old_value, value):
+            return
+        now = time.monotonic()
+        remaining = self._broadcast_rate_limit_s - (now - self._last_broadcast_ts)
+        if remaining <= 0:
+            self._last_broadcast_ts = now
+            self._run_subs(sub_type="readback")
 
     # Aliases
     # tey = s1