PyQt5 Gui 线程到工作线程信号/插槽 [英] PyQt5 Gui Thread to worker Thread signal/Slot

问题描述

我试图在工作线程之间进行简单的通信，在这种情况下，它通过 PyQt5 信号和插槽机制被称为 WorkToDo.我可以通过这种机制可靠地将数据从 Worker 发送到 Gui 线程，但我不能对 gui 线程做同样的事情.从我的研究中，我发现这是由于我用自己的逻辑覆盖了 run 函数.我的问题是，有没有办法手动处理工作线程中信号的执行?有没有更好的方法来实现这一点?

I am trying to make a simple communication between a worker thread, in this case it is called WorkToDo via the PyQt5 Signals and Slots mechanism. I can reliably send data from the Worker to the Gui thread via this mechanism, but I cannot do the same to the gui thread. From my research I have found that this is due to the fact that I have overridden the run function with my own logic. My question, is there any way to manually handle the execution of signals in the worker thread? Is there a better way to accomplish this?

编辑:
我实际上并没有覆盖 run，因为我没有看到 QObject 文档中列出的 run.我错了，因为这是 QThread 对象的默认函数.这对我来说更令人困惑.除非我进一步误解了这一点.

EDIT:
I am actually not overriding run as I do not see run listed in the documentation for QObject. I am mistaken as this is a default function of a QThread object. This is more perplexing to me. Unless I am just further misunderstanding this.

import sys
import time
import qdarkstyle

from PyQt5.QtWidgets import (
                            QWidget, QLineEdit, QMessageBox, QPushButton, QVBoxLayout, QHBoxLayout, QTabWidget,
                            QComboBox, QProgressBar, QApplication, QLabel, QGroupBox, QFileDialog, QTextEdit,
                            QStyleFactory, QFormLayout
                            )

from PyQt5 import (Qt, QtCore)

from PyQt5.QtCore import *


class Command:
    def __init__(self, **kwargs):
        self.cmd = kwargs.get('cmd', None)
        self.args = kwargs.get('args')

class CommandSignals(QObject):
    command = pyqtSignal(Command)

class WorkToDo(QObject):
    def __init__(self):
        super().__init__()
        self.counter = 0
        self.signals = CommandSignals()

    @QtCore.pyqtSlot(Command)
    def process_command(self, command):
        """
        @brief       process update from gui thread
        @param       self     self
        @param       command  command
        @return      none
        """
        print(f'Update from GUI: {command.__dict__}')
        if command.cmd == 'add_to_counter':
            self.counter = self.counter + command.args.get('addition', 0)

    @pyqtSlot()
    def run(self):
        """
        @brief       actual thread function to run,
                     started via thread.started signal (why its decorated with a pyqtSlot)
        @param       self  obj ref
        @return      none
        """
        while True:
            QThread.sleep(1)
            # emit text change signal to main gui thread
            cmd = Command(
                cmd = 'update_text',
                args = {
                    'text': f'Hello update {self.counter}'
                }
            )
            print(f'Update from worker: {cmd.__dict__}')
            self.signals.command.emit(cmd)
            self.counter += 1


class Gui(QWidget):
    def __init__(self):
        super().__init__()
        """ make gui elements """
        layout = QFormLayout()
        self.text_line = QLineEdit()
        self.add_button = QPushButton('Add 10 To Counter')
        layout.addRow(self.text_line)
        layout.addRow(self.add_button)

        self.add_button.clicked.connect(self.issue_button_update)
        self.signals = CommandSignals()
        self.MakeThreadWorker()

        """ finalize gui """
        self.setLayout(layout)
        self.setWindowTitle('Sync Thread Command/Response test')
        self.show()


    def MakeThreadWorker(self):
        self.worker_thread = QThread()
        self.worker = WorkToDo()

        """ incoming gui update command, works """
        self.worker.signals.command.connect(self.process_command)
        """ outgoing update to worker thread, does not work """
        self.signals.command.connect(self.worker.process_command)

        """ signal to start the thread function, works """
        self.worker_thread.started.connect(self.worker.run)

        self.worker_thread.start()
        self.worker.moveToThread(self.worker_thread)


    @pyqtSlot(Command)
    def process_command(self, command):
        """
        @brief       process update from work thread
        @param       self     self
        @param       command  command object
        @return      none
        """
        if command.cmd == 'update_text':
            text_update = command.args.get('text', '')
            self.text_line.setText(text_update)

    def issue_button_update(self):
        cmd = Command(
            cmd = 'add_to_counter',
            args = {
                'addition': 10
            }
        )
        print('Button Clicked!')
        self.signals.command.emit(cmd)


if __name__ == '__main__':
    APPLICATION = QApplication([])
    APPLICATION.setStyleSheet(qdarkstyle.load_stylesheet_pyqt5())
    gui = Gui()
    sys.exit(APPLICATION.exec_())

推荐答案

问题是当你阻塞辅助线程的事件循环阻止接收信号时，如果你想做一个周期性的任务，那么问题是使用 QTimer.

The problem is that with True while you are blocking the event loop of the secondary thread preventing the signals from being received, if you want to do a periodic task then use a QTimer.

import sys
import time
import qdarkstyle

from PyQt5.QtCore import pyqtSignal, pyqtSlot, QObject, QThread, QTimer
from PyQt5.QtWidgets import QApplication, QFormLayout, QLineEdit, QPushButton, QWidget


class Command:
    def __init__(self, **kwargs):
        self.cmd = kwargs.get("cmd", None)
        self.args = kwargs.get("args")


class CommandSignals(QObject):
    command = pyqtSignal(Command)


class WorkToDo(QObject):
    def __init__(self):
        super().__init__()
        self.counter = 0
        self.signals = CommandSignals()
        self.timer = QTimer(self, timeout=self.run, interval=1000)

    @pyqtSlot()
    def start(self):
        self.timer.start()

    @pyqtSlot()
    def stop(self):
        self.timer.stop()

    @pyqtSlot(Command)
    def process_command(self, command):
        """
        @brief       process update from gui thread
        @param       self     self
        @param       command  command
        @return      none
        """
        print(f"Update from GUI: {command.__dict__}")
        if command.cmd == "add_to_counter":
            self.counter = self.counter + command.args.get("addition", 0)

    @pyqtSlot()
    def run(self):
        print(self.thread(), self.timer.thread())
        cmd = Command(cmd="update_text", args={"text": f"Hello update {self.counter}"})
        print(f"Update from worker: {cmd.__dict__}")
        self.signals.command.emit(cmd)
        self.counter += 1


class Gui(QWidget):
    def __init__(self):
        super().__init__()
        """ make gui elements """
        layout = QFormLayout()
        self.text_line = QLineEdit()
        self.add_button = QPushButton("Add 10 To Counter")
        layout.addRow(self.text_line)
        layout.addRow(self.add_button)

        self.add_button.clicked.connect(self.issue_button_update)
        self.signals = CommandSignals()
        self.MakeThreadWorker()

        """ finalize gui """
        self.setLayout(layout)
        self.setWindowTitle("Sync Thread Command/Response test")
        self.show()

    def MakeThreadWorker(self):
        self.worker_thread = QThread()
        self.worker = WorkToDo()

        """ incoming gui update command, works """
        self.worker.signals.command.connect(self.process_command)
        """ outgoing update to worker thread, does not work """
        self.signals.command.connect(self.worker.process_command)

        """ signal to start the thread function, works """
        self.worker_thread.started.connect(self.worker.start)

        self.worker_thread.start()
        self.worker.moveToThread(self.worker_thread)

    @pyqtSlot(Command)
    def process_command(self, command):
        """
        @brief       process update from work thread
        @param       self     self
        @param       command  command object
        @return      none
        """
        if command.cmd == "update_text":
            text_update = command.args.get("text", "")
            self.text_line.setText(text_update)

    def issue_button_update(self):
        cmd = Command(cmd="add_to_counter", args={"addition": 10})
        print("Button Clicked!")
        self.signals.command.emit(cmd)


if __name__ == "__main__":
    APPLICATION = QApplication([])
    APPLICATION.setStyleSheet(qdarkstyle.load_stylesheet_pyqt5())
    gui = Gui()
    sys.exit(APPLICATION.exec_())

这篇关于PyQt5 Gui 线程到工作线程信号/插槽的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！