Python decorator to aggregate PyQt signals

Question

I often have many signals that would ideally be processed all at once. For example, the signal that triggers an update to an OpenGL window can be aggregated into a single signal. Another example is the signal that dirties a row in a table.

Ideally, I'd like the decorator to generate code something like what's below (between START and END):

#!/usr/bin/env python

from PyQt4.QtCore import *
from PyQt4.QtGui import *
from functools import wraps
import signal
import sys

signal.signal(signal.SIGINT, signal.SIG_DFL)

class AggregateManager:
    def __init__(self):
        self.clear()

    def clear(self):
        self.sent = False
        self.value = 0

    def aggregate(self, other):
        send = not self.sent
        self.sent = True
        self.value += other
        return send


class A(QObject):
    def __init__(self):
        super().__init__()
    # START
        self.generated_signal.connect(self.slot, Qt.QueuedConnection)
        self.slot_manager = AggregateManager()

    @pyqtSlot(int)
    def decorated_slot(self, *args):
        me = self.slot_manager
        if me.aggregate(*args):
            print("Sending")
            self.generated_signal.emit()

    generated_signal = pyqtSignal()

    @pyqtSlot()
    def slot(self):
        me = self.slot_manager
        print("Received", me.value)
        me.clear()
    # END


class B(QObject):
    signal = pyqtSignal(int)


a = A()
b = B()

b.signal.connect(a.decorated_slot)

for i in range(10):
    b.signal.emit(i)

app = QApplication(sys.argv)
sys.exit(app.exec_())

This way, a single call to slot is made for many signals sent to decorated_slot. How do I use Python decorators to replace everything between START and END?

Answer 1

I've done something similar to this but on a less generic and more granular scale. I created a context manager to temporarily disconnect slots. This way it's clear that you disable certain slots within a block of code then you re-connect them and can emit anything that you missed in between.

The source is here and I pasted the snippet below. This might not be quite what you want, but it was useful for me in a similar scenario. I wanted to do something that could emit of 'intermediate' signals where the 'intermediate' changes didn't matter in the long run. So, with this context manager I can disable the noisy signals and then just emit it once afterwards if needed. This allows you to avoid a lot of intermediate state changes.

A good example is setting PyQt QCheckboxes in a loop. You could call checkbox.setChecked() in a loop but you would emit a bunch of signals for each checkbox. However, you could use the context manager below to disable any slots for the stateChanged signal and the emit the stateChanged signal once yourself with the 'final' result.

from contextlib import contextmanager

@contextmanager
def slot_disconnected(signal, slot):
    """
    Create context to perform operations with given slot disconnected from
    given signal and automatically connected afterwards.

    usage:
        with slot_disconnected(chkbox.stateChanged, self._stateChanged):
            foo()
            bar()
    """

    signal.disconnect(slot)
    yield
    signal.connect(slot)

Answer 2

Here's what I have so far. The only problem is that the pyqtSignal decorator seems to be getting something from the stack trace and there's no way that I know of to override that, which is a clear PyQt design flaw.

#!/usr/bin/env python

from PyQt4.QtCore import *
from PyQt4.QtGui import *
from functools import wraps
import signal
import sys

signal.signal(signal.SIGINT, signal.SIG_DFL)

class SetAdder:
    def __init__(self):
        self.clear()

    def clear(self):
        self.value = set()

    def aggregate(self, other):
        send = not self.value
        self.sent = True
        self.value.add(other)
        return send


# This class decorator adds nameSlot, nameAuxSignal, nameAuxSlot, and
# name_manager.  Signals should be connected to nameSlot.  They will cause
# the function 'name' to be called with aggregated values.
def aggregated_slot_class_decorator(list_):
    def class_decorator(cls):
        for manager_type, name, *args in list_:
            signal_name = name + "AuxSignal"
            slot_a_name = name + "Slot"
            slot_b_name = name + "AuxSlot"
            manager_name = name + "_manager"

            def slot_a(self, *args_):
                manager = getattr(self, manager_name)
                if manager.aggregate(*args_):
                    print("Sending")
                    getattr(self, signal_name).emit()

            def slot_b(self):
                manager = getattr(self, manager_name)
                getattr(self, name)(manager.value)
                manager.clear()

            setattr(cls, slot_a_name,
                    pyqtSlot(cls, *args, name=slot_a_name)(slot_a))
            setattr(cls, slot_b_name,
                    pyqtSlot(cls, name=slot_b_name)(slot_b))

            orig_init = cls.__init__
            def new_init(self, *args_, **kwargs):
                orig_init(self, *args_, **kwargs)
                getattr(self, signal_name).connect(getattr(self, slot_b_name),
                                                   Qt.QueuedConnection)
                setattr(self, manager_name, manager_type())
            cls.__init__ = new_init
            #setattr(cls, signal_name, pyqtSignal())
        return cls
    return class_decorator


@aggregated_slot_class_decorator([(SetAdder, 'test', int)])
class A(QObject):
    def __init__(self):
        super().__init__()

    testAuxSignal = pyqtSignal()

    def test(self, value):
        print("Received", value)


class B(QObject):
    signal = pyqtSignal(int)


a = A()
b = B()

b.signal.connect(a.testSlot)

for i in range(10):
    b.signal.emit(i % 5)

app = QApplication(sys.argv)
sys.exit(app.exec_())

Outputs:

Sending
Received {0, 1, 2, 3, 4}