Emit new style PyQt signals with arbitrary signature

Question

I am creating a PyQt GUI for an experimental setup. This will involve computationally heavy operations so I am aiming for an architecture based on the multiprocessing module and inspired from this answer.

The QMainWindow creates

1. child Processes with an individual "task" Queue to get instructions from the main process and a shared "callback" Queue to send back instructions to the main process
2. a QThread to poll the "callback" queue and translate the messages into signals that are connected to slots of the QMainWindow

The example uses old style signals with arbitrary signature self.emit(QtCore.SIGNAL(signature), args). My question is: is it possible to replicate this functionality with new-style signals ?.

I am aware of this question and of this one. However, always emitting a valueChanged signal with a general object does not suit my needs since I would like to connect to slots with different names depending on the signature received from one of the child Processes.

Here is a working code (note there is only one child process and one slot in the MainWindow for simplicity, but there will be several in the finished code):

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

The new-style signal and slot syntax requires that signals are pre-defined as class attributes on a class that inherits from QObject. When the class is instantiated, a bound-signal object is automatically created for the instance. The bound-signal object has connect/disconnect/emit methods, and a __getitem__ syntax which allows different overloads to be selected.

Since bound-signals are objects, it no longer makes sense to allow the dynamic emission of arbitrary signals that was possible with the old-style syntax. This is simply because an arbitrary signal (i.e. one that is not pre-defined) could not have a corresponding bound-signal object for slots to connect to.

The example code in the question can still be ported to the new-style syntax, though:

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)