Python-декоратор для агрегирования сигналов PyQt

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_())

Таким образом, один вызов slot выполняется для многих сигналов, отправляемых на Deco_Slot. Как использовать Python-декораторы для замены всего, что находится между START и END?

Answer 1

Я сделал нечто похожее на это, но в менее общем и более детальном масштабе.Я создал менеджер контекста, чтобы временно отключить слоты.Таким образом, становится ясно, что вы отключаете определенные слоты в блоке кода, затем повторно подключаете их и можете выдавать все, что пропустили между ними.

Источник здесь , и я вставилфрагмент ниже.Это может быть не совсем то, что вы хотите, но это было полезно для меня в аналогичном сценарии.Я хотел сделать что-то, что могло бы издавать «промежуточные» сигналы, когда «промежуточные» изменения не имели значения в долгосрочной перспективе.Таким образом, с помощью этого диспетчера контекста я могу отключить шумовые сигналы, а затем просто при необходимости испустить его один раз.Это позволяет избежать многих промежуточных изменений состояния.

Хорошим примером является установка PyQt QCheckboxes в цикле.Вы могли бы вызвать checkbox.setChecked() в цикле, но вы бы испустили кучу сигналов для каждого флажка.Однако вы можете использовать менеджер контекста ниже, чтобы отключить любые слоты для сигнала stateChanged и подать сигнал stateChanged один раз с «окончательным» результатом.

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

Вот что у меня так далеко.Единственная проблема заключается в том, что 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}