Aluode/PerceptionLabPortable

1import weakref2from time import perf_counter3 4from .functions import SignalBlock5from .Qt import QtCore6from .ThreadsafeTimer import ThreadsafeTimer7 8__all__ = ['SignalProxy']9 10 11class SignalProxy(QtCore.QObject):12    """Object which collects rapid-fire signals and condenses them13    into a single signal or a rate-limited stream of signals.14    Used, for example, to prevent a SpinBox from generating multiple15    signals when the mouse wheel is rolled over it.16 17    Emits sigDelayed after input signals have stopped for a certain period of18    time.19    """20 21    sigDelayed = QtCore.Signal(object)22 23    def __init__(self, signal, delay=0.3, rateLimit=0, slot=None, *, threadSafe=True):24        """Initialization arguments:25        signal - a bound Signal or pyqtSignal instance26        delay - Time (in seconds) to wait for signals to stop before emitting (default 0.3s)27        slot - Optional function to connect sigDelayed to.28        rateLimit - (signals/sec) if greater than 0, this allows signals to stream out at a29                    steady rate while they are being received.30        threadSafe - Specify if thread-safety is required. For backwards compatibility, it31                     defaults to True.32        """33 34        QtCore.QObject.__init__(self)35        self.delay = delay36        self.rateLimit = rateLimit37        self.args = None38        Timer = ThreadsafeTimer if threadSafe else QtCore.QTimer39        self.timer = Timer()40        self.timer.timeout.connect(self.flush)41        self.lastFlushTime = None42        self.signal = signal43        self.signal.connect(self.signalReceived)44        if slot is not None:45            self.blockSignal = False46            self.sigDelayed.connect(slot)47            self.slot = weakref.ref(slot)48        else:49            self.blockSignal = True50            self.slot = None51 52    def setDelay(self, delay):53        self.delay = delay54 55    @QtCore.Slot()56    @QtCore.Slot(object)57    @QtCore.Slot(object, object)58    def signalReceived(self, *args):59        """Received signal. Cancel previous timer and store args to be60        forwarded later."""61        if self.blockSignal:62            return63        self.args = args64        if self.rateLimit == 0:65            self.timer.stop()66            self.timer.start(int(self.delay * 1000) + 1)67        else:68            now = perf_counter()69            if self.lastFlushTime is None:70                leakTime = 071            else:72                lastFlush = self.lastFlushTime73                leakTime = max(0, (lastFlush + (1.0 / self.rateLimit)) - now)74 75            self.timer.stop()76            self.timer.start(int(min(leakTime, self.delay) * 1000) + 1)77 78    @QtCore.Slot()79    def flush(self):80        """If there is a signal queued up, send it now."""81        if self.args is None or self.blockSignal:82            return False83        args, self.args = self.args, None84        self.timer.stop()85        self.lastFlushTime = perf_counter()86        self.sigDelayed.emit(args)87        return True88 89    def disconnect(self):90        self.blockSignal = True91        try:92            self.signal.disconnect(self.signalReceived)93        except:94            pass95        try:96            slot = self.slot()97            if slot is not None:98                self.sigDelayed.disconnect(slot)99        except:100            pass101        finally:102            self.slot = None103 104    def connectSlot(self, slot):105        """Connect the `SignalProxy` to an external slot"""106        assert self.slot is None, "Slot was already connected!"107        self.slot = weakref.ref(slot)108        self.sigDelayed.connect(slot)109        self.blockSignal = False110 111    def block(self):112        """Return a SignalBlocker that temporarily blocks input signals to113        this proxy.114        """115        return SignalBlock(self.signal, self.signalReceived)116