f
CI for xtreme_bec / test (push) Successful in 30s

This commit is contained in:
2026-06-23 17:23:12 +02:00
parent 70ce00c7ad
commit 51b9effe31
+5 -6
View File
@@ -16,6 +16,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
@@ -337,7 +338,7 @@ class NormDIODESignals(Device):
self.norm.name = self.name
class X07MAAnalogSignals(Device):
class X07MAAnalogSignals(PSIDeviceBase, Device):
"""
ADC inputs
"""
@@ -355,13 +356,11 @@ class X07MAAnalogSignals(Device):
norm_tey = Cpt(NormTEYSignals, name="norm_tey", kind=Kind.hinted)
norm_diode = Cpt(NormDIODESignals, name="norm_diode", kind=Kind.hinted)
def __init__(self, *args, broadcast_rate_limit_s=0.1, **kwargs):
super().__init__(*args, **kwargs)
self._broadcast_rate_limit_s = broadcast_rate_limit_s
def on_connected(self) -> None:
self._broadcast_lock = threading.Lock()
self._last_broadcast_ts = 0.0
self._last_broadcast_ts = 0
self._broadcast_rate_limit_s = 0.1 # seconds
self._pending_broadcast = None
for signal_name in ("s1", "s2", "s3", "s4", "s5", "s6", "s7"):
getattr(self, signal_name).subscribe(self._schedule_broadcast, run=False)