Вывести новый сигнал PyQt с произвольной подписи

Question

Я создаю GUI PyQt для экспериментальной установки. Это будет включать в себя вычислительно тяжелые операции, поэтому я нацелен на архитектуру, основанную на модуле многопроцессорности и вдохновленную от this answer.

Функция QMainWindow создает

1. дочерних процессов с помощью отдельной «задачи» очередь, чтобы получить инструкции от основного процесса и общего «обратного вызова» очереди, чтобы послать инструкции к основному процессу
2. QThread к опрашивать «обратный вызов» очередь и переводить сообщения в сигналы, которые подключены к пазам QMainWindow

пример использует старые сигналы стиля с произвольной сигнатурой self.emit(QtCore.SIGNAL(signature), args). Мой вопрос: Можно ли воспроизвести эту функцию с помощью сигналов нового стиля?.

Я знаю this question и 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. Когда экземпляр класса создается, для экземпляра создается автоматически связанный объект. Объект связанного сигнала имеет методы 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)