发出具有任意签名的新型 PyQt 信号

Question

我正在为实验设置创建 PyQt GUI。这将涉及计算繁重的操作，因此我的目标是基于多处理模块的架构，并受到 this answer 的启发。

QMainWindow 创建

具有单独的子进程"task"队列以从主进程获取指令和共享"callback"队列以将指令发送回主进程
一个 QThread 轮询 "callback" 队列并将消息转换为连接到 QMainWindow

该示例使用具有任意签名的旧式信号 self.emit(QtCore.SIGNAL(signature), args)。 我的问题是： 是否可以使用新型信号复制此功能？。

我知道 this question and of this one。但是，总是用一般对象发出 valueChanged 信号不符合我的需要，因为我想根据从其中一个子进程收到的签名连接到具有不同名称的插槽。

这是一个工作代码（请注意，为简单起见，MainWindow 中只有一个子进程和一个插槽，但在完成的代码中会有多个）：

from multiprocessing import Process, Queue
import sys
from PyQt4 import QtGui, QtCore


class CallbackQueueToSignal(QtCore.QThread):

    def __init__(self, queue, parent=None):
        super(CallbackQueueToSignal, self).__init__(parent)
        self.queue = queue

    def _emit(self, signature, args=None):
        if args:
            self.emit(QtCore.SIGNAL(signature), args)
        else:
            self.emit(QtCore.SIGNAL(signature))

    def run(self):
        while True:
            signature = self.queue.get()
            self._emit(*signature)


class WorkerProcess(Process):

    def __init__(self, callback_queue, task_queue, daemon=True):
        super(WorkerProcess, self).__init__()
        self.daemon = daemon
        self.callback_queue = callback_queue
        self.task_queue = task_queue

    def _process_call(self, func_name, args=None):
        func = getattr(self, func_name)
        if args:
            func(args)
        else:
            func()

    def emit_to_mother(self, signature, args=None):
        signature = (signature, )
        if args:
            signature += (args, )
        self.callback_queue.put(signature)

    def run(self):
        while True:
            call = self.task_queue.get()
            # print("received: {}".format(call))
            self._process_call(*call)

    def text_upper(self, text):
        self.emit_to_mother('data(PyQt_PyObject)', (text.upper(),))


class MainWin(QtGui.QMainWindow):

    def __init__(self, parent=None):
        super(MainWin, self).__init__(parent)

        self.data_to_child = Queue()
        self.callback_queue = Queue()

        self.callback_queue_watcher = CallbackQueueToSignal(self.callback_queue)
        self.callback_queue_watcher.start()

        self.child = WorkerProcess(self.callback_queue, self.data_to_child)
        self.child.start()

        self.browser = QtGui.QTextBrowser()
        self.lineedit = QtGui.QLineEdit('Type text and press <Enter>')
        self.lineedit.selectAll()
        layout = QtGui.QVBoxLayout()
        layout.addWidget(self.browser)
        layout.addWidget(self.lineedit)
        self.layout_widget = QtGui.QWidget()
        self.layout_widget.setLayout(layout)
        self.setCentralWidget(self.layout_widget)
        self.lineedit.setFocus()
        self.setWindowTitle('Upper')
        self.connect(self.lineedit, QtCore.SIGNAL('returnPressed()'), self.to_child)
        self.connect(self.callback_queue_watcher, QtCore.SIGNAL('data(PyQt_PyObject)'), self.updateUI)

    def to_child(self):
        self.data_to_child.put(("text_upper", ) + (self.lineedit.text(), ))
        self.lineedit.clear()

    def updateUI(self, text):
        text = text[0]
        self.browser.append(text)

    def closeEvent(self, event):
        result = QtGui.QMessageBox.question(
            self,
            "Confirm Exit...",
            "Are you sure you want to exit ?",
            QtGui.QMessageBox.Yes | QtGui.QMessageBox.No)
        event.ignore()

        if result == QtGui.QMessageBox.Yes:
            # self.pipeWatcher.exit()
            self.child.terminate()
            event.accept()

if __name__ == '__main__':

    app = QtGui.QApplication(sys.argv)

    form = MainWin()
    form.show()

    app.aboutToQuit.connect(app.deleteLater)
    sys.exit(app.exec_())

Answer 1

new-style signal and slot syntax 要求将信号预定义为继承自 QObject 的 class 上的 class 属性。当实例化 class 时，会自动为该实例创建一个绑定信号对象。绑定信号对象具有 connect/disconnect/emit 方法和 __getitem__ 允许选择不同重载的语法。

由于绑定信号是对象，允许动态发射任意信号不再有意义，而这在旧式语法中是可能的。这仅仅是因为任意信号（即不是预定义的信号）不能有相应的绑定信号对象供插槽连接。

问题中的示例代码仍然可以移植到新式语法中，但是：

class CallbackQueueToSignal(QtCore.QThread):
    dataSignal = QtCore.pyqtSignal([], [object], [object, object])   
    ...

    def _emit(self, signal, *args):
        getattr(self, signal)[(object,) * len(args)].emit(*args)

    def run(self):
        while True:
            args = self.queue.get()
            self._emit(*args)


class WorkerProcess(Process):
    ...

    def emit_to_mother(self, *args):
        self.callback_queue.put(args)

    def text_upper(self, text):
        self.emit_to_mother('dataSignal', (text.upper(),))


class MainWin(QtGui.QMainWindow):
    def __init__(self, parent=None):
        ...

        self.lineedit.returnPressed.connect(self.to_child)
        self.callback_queue_watcher.dataSignal[object].connect(self.updateUI)

发出具有任意签名的新型 PyQt 信号

Emit new style PyQt signals with arbitrary signature

python

pyqt4

python-3.x

qt-signals

python-multiprocessing