diff --git a/xtreme_bec/devices/x07ma_devices.py b/xtreme_bec/devices/x07ma_devices.py index efafac1..c488b3a 100644 --- a/xtreme_bec/devices/x07ma_devices.py +++ b/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):