用于聚合 PyQt 信号的 Python 装饰器

Question

我经常有许多信号，理想情况下可以一次性处理所有信号。例如，触发 OpenGL 窗口更新的信号可以聚合为单个信号。另一个例子是弄脏表中一行的信号。

理想情况下，我希望装饰器生成如下代码(在 START 和 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_())

这样，对于发送到decorated_slot的许多信号，只需调用一次slot即可。如何使用 Python 装饰器替换 START 和 END 之间的所有内容？

Answer 1

这是我到目前为止所拥有的。唯一的问题是 pyqtSignal 装饰器似乎从堆栈跟踪中获取了一些东西，并且据我所知没有办法覆盖它，这是一个明显的 PyQt 设计缺陷。

#!/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_())

输出:

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