feat: implement broadcast rate limiting for ADC signals

xtreme_bec/devices/x07ma_devices.py

@@ -2,11 +2,14 @@
 ophyd device classes for X07MA beamline
 """
 
+from __future__ import annotations
+
 import time
 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
@@ -15,6 +18,7 @@ from ophyd import Kind, PVPositioner, Signal
 from ophyd.flyers import FlyerInterface
 from ophyd.pv_positioner import PVPositionerComparator
 from ophyd.status import DeviceStatus, SubscriptionStatus
+from ophyd_devices import PSIDeviceBase
 
 logger = bec_logger.logger
 
@@ -336,27 +340,35 @@ class NormDIODESignals(Device):
         self.norm.name = self.name
 
 
-class X07MAAnalogSignals(Device):
+class X07MAAnalogSignals(PSIDeviceBase, Device):
     """
     ADC inputs
     """
 
-    s1 = Cpt(EpicsSignalRO, "SIGNAL0", kind=Kind.hinted, auto_monitor=True)
-    s2 = Cpt(EpicsSignalRO, "SIGNAL1", kind=Kind.hinted, auto_monitor=True)
-    s3 = Cpt(EpicsSignalRO, "SIGNAL2", kind=Kind.hinted, auto_monitor=True)
-    s4 = Cpt(EpicsSignalRO, "SIGNAL3", kind=Kind.hinted, auto_monitor=True)
-    s5 = Cpt(EpicsSignalRO, "SIGNAL4", kind=Kind.hinted, auto_monitor=True)
-    s6 = Cpt(EpicsSignalRO, "SIGNAL5", kind=Kind.hinted, auto_monitor=True)
-    s7 = Cpt(EpicsSignalRO, "SIGNAL6", kind=Kind.hinted, auto_monitor=True)
+    s1 = Cpt(EpicsSignalRO, "SIGNAL0", kind=Kind.hinted)
+    s2 = Cpt(EpicsSignalRO, "SIGNAL1", kind=Kind.hinted)
+    s3 = Cpt(EpicsSignalRO, "SIGNAL2", kind=Kind.hinted)
+    s4 = Cpt(EpicsSignalRO, "SIGNAL3", kind=Kind.hinted)
+    s5 = Cpt(EpicsSignalRO, "SIGNAL4", kind=Kind.hinted)
+    s6 = Cpt(EpicsSignalRO, "SIGNAL5", kind=Kind.hinted)
+    s7 = Cpt(EpicsSignalRO, "SIGNAL6", kind=Kind.hinted)
     norm_tey = Cpt(NormTEYSignals, name="norm_tey", kind=Kind.hinted)
-    norm_diode = Cpt(NormDIODESignals, name="norm_tey", kind=Kind.hinted)
+    norm_diode = Cpt(NormDIODESignals, name="norm_diode", kind=Kind.hinted)
 
-    # Aliases
-    # tey = s1
-    # i0 = s2
-    # trans = s3
-    # field_z = s4
-    # field_x = s5
+    def on_connected(self) -> None:
+        self._last_broadcast_ts = 0
+        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, 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")
 
 
 class X07MASampleManipulator(Device):