如何解决Python装饰器聚合PyQt信号
| 我经常会有很多信号，理想情况下可以一次全部处理。例如，可以将触发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_())
解决方法
这是我到目前为止的内容。唯一的问题是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,name=slot_b_name)(slot_b))
orig_init = cls.__init__
def new_init(self,*args_,**kwargs):
orig_init(self,**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}
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)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。