bec/xtreme_bec
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Automatic backup triggered by new deployment
CI for xtreme_bec / test (push) Successful in 35s
Details

Browse Source
This commit is contained in:
x07ma
2026-06-26 09:54:31 +02:00
parent 36d083a398
commit a0148ea4b9
1 changed files with 20 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xtreme_bec/devices/x07ma_devices.py
+20 -20
View File
@@ -347,30 +347,30 @@ class X07MAAnalogSignals(PSIDeviceBase, Device):
SUB_VALUE = "value"
_default_sub = SUB_VALUE
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)
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)
norm_tey = Cpt(NormTEYSignals, name="norm_tey", kind=Kind.hinted)
norm_diode = Cpt(NormDIODESignals, name="norm_diode", kind=Kind.hinted)
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 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")
# 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