PYQT - 在 QThread 之间共享信号

Question

这是一个解释我的问题的小例子。我想接收 gui 类中的所有线程信号，来自线程 A 和线程 B 的信号应打印在线程 C 中。下一步我想将信号写入 Excel 表。

1. 问题：如何共享信号
2. 问题：是否可以设置像 x=signal 这样的变量

感谢您的回复。

import sys,time
from PyQt4 import  QtGui, uic
from PyQt4.QtCore import QThread, SIGNAL

qtCreatorFile = 'ThreadUi.ui'

Ui_MainWindow, QtBaseClass = uic.loadUiType(qtCreatorFile)

class MyApp(QtGui.QMainWindow, Ui_MainWindow):
    def __init__(self,parent=None):
        super(MyApp, self).__init__(parent)
        self.setupUi(self)
        self.Thread()
        self.Ui()


    def Ui(self):
        self.state= False
        self.pushButtonA.pressed.connect(self.BtnA)
        self.pushButtonB.pressed.connect(self.BtnB)
        self.pushButtonC.pressed.connect(self.BtnC)

    def Thread(self):
        self.threadA = WorkThreadA(self)
        self.threadB = WorkThreadB(self)
        self.threadC = WorkThreadC

    def BtnA(self):
        self.state = not self.state
        if self.state:
            self.threadA.start()
            self.pushButtonA.setChecked(False)
        else:
            self.threadA.stop()
            self.pushButtonA.setChecked(True)

    def BtnB(self):
        self.state = not self.state
        if self.state:
            self.threadB.start()
            self.pushButtonB.setChecked(False)
        else:
            self.threadB.stop()
            self.pushButtonB.setChecked(True)

    def BtnC(self):
        A=self.connect(self.threadA,SIGNAL('Signal_A'))
        B=self.connect(self.threadA,SIGNAL('Signal_B'))
        self.threadC(self,A,B)

        self.state = not self.state
        if self.state:
            self.threadC.start()
            self.pushButtonC.setChecked(False)
        else:
            self.threadC.stop()
            self.pushButtonC.setChecked(True)

if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    window = MyApp()
    window.show()
    sys.exit(app.exec_())


class WorkThreadA(QThread):
     def __init__(self ,parent=None):
         super(WorkThreadA, self).__init__(parent)
         self.host=host_window
         self.exiting = False

     def run(self):
         self.exiting = False
         i=0
         while not self.exiting:
             self.emit(SIGNAL('Signal_A'),i)
             print("Thread A:",i)
             i=i+1
             time.sleep(1)

     def stop(self):
         self.exiting = True

class WorkThreadB(QThread):
    def __init__(self,host_window ,parent=None):
        super(WorkThreadB, self).__init__(parent)
        self.host=host_window
        self.exiting = False

    def run(self):
        self.exiting = False
        i=0
        while not self.exiting:
            self.emit(SIGNAL('Signal_B'),i)
            print("Thread B:",i)
            i=i+1
            time.sleep(1)

    def stop(self):
        self.exiting = True

class WorkThreadC(QThread):
    def __init__(self,host_window ,SignalA,SignalB,parent=None):
        super(WorkThreadC, self).__init__(parent)
        self.host=host_window
        self.exiting = False
        self.signalA=SignalA
        self.signalB=SignalB

    def run(self):
        self.exiting = False
        while not self.exiting:
            print("Signal A: ",self.signalA)
            print("Signal B: ",self.signalB)    

    def stop(self):
        self.exiting = True

UI 文件在这里：

<?xml version="1.0" encoding="UTF-8"?>
   <ui version="4.0">
    <class>Form</class>
    <widget class="QWidget" name="Form">
     <property name="geometry">
      <rect>
       <x>0</x>
       <y>0</y>
       <width>400</width>
       <height>300</height>
      </rect>
     </property>
     <property name="windowTitle">
      <string>TreadUi</string>
     </property>
     <widget class="QWidget" name="verticalLayoutWidget">
      <property name="geometry">
       <rect>
        <x>-1</x>
        <y>-1</y>
        <width>401</width>
        <height>301</height>
       </rect>
      </property>
      <layout class="QVBoxLayout" name="verticalLayout">
       <item>
        <widget class="QPushButton" name="pushButtonA">
         <property name="sizePolicy">
          <sizepolicy hsizetype="Minimum" vsizetype="MinimumExpanding">
           <horstretch>0</horstretch>
           <verstretch>0</verstretch>
          </sizepolicy>
         </property>
         <property name="text">
          <string>Thread A</string>
         </property>
         <property name="checkable">
          <bool>true</bool>
         </property>
        </widget>
       </item>
       <item>
        <widget class="QPushButton" name="pushButtonB">
         <property name="sizePolicy">
          <sizepolicy hsizetype="Minimum" vsizetype="MinimumExpanding">
           <horstretch>0</horstretch>
           <verstretch>0</verstretch>
          </sizepolicy>
         </property>
         <property name="text">
          <string>Thread B</string>
         </property>
         <property name="checkable">
          <bool>true</bool>
         </property>
        </widget>
       </item>
       <item>
        <widget class="QPushButton" name="pushButtonC">
         <property name="sizePolicy">
          <sizepolicy hsizetype="Minimum" vsizetype="MinimumExpanding">
           <horstretch>0</horstretch>
           <verstretch>0</verstretch>
          </sizepolicy>
         </property>
         <property name="text">
          <string>Thread C</string>
         </property>
         <property name="checkable">
          <bool>true</bool>
         </property>
        </widget>
       </item>
      </layout>
     </widget>
    </widget>
    <resources/>
    <connections/>
   </ui>

Answer 1

您不能在类实例上动态定义信号。它们必须被定义为类属性。您应该使用new-style signals and slot syntax。在使用 pyqtSlot 装饰器跨线程连接信号和槽时，您还需要声明您的槽。

class WorkThreadA(QThread):
    signal_a = QtCore.pyqtSignal(int)

    def run(self):
        ...
        self.signal_a.emit(10)


class MyApp(...)

    def thread(self):
        self.threadA = WorkThreadA(self)
        self.threadA.signal_a.connect(self.handle_signal)

    @QtCore.pyqtSlot(int)
    def handle_signal(self, value):
        print 'Thread Value', value

Answer 2

现在它可以工作了，但不确定这是否是最好的方法。 感谢您的帮助。

import sys,time
from PyQt4 import  QtGui, uic
from PyQt4.QtCore import QThread , pyqtSignal,pyqtSlot

qtCreatorFile = 'ThreadUi.ui'

Ui_MainWindow, QtBaseClass = uic.loadUiType(qtCreatorFile)

class MyApp(QtGui.QMainWindow, Ui_MainWindow):

    def __init__(self,parent=None):
        super(MyApp, self).__init__(parent)
        self.setupUi(self)
        self.Thread()
        self.Ui()
        self.Signals()


    def Ui(self):
        self.state= False
        self.pushButtonA.pressed.connect(self.BtnA)
        self.pushButtonB.pressed.connect(self.BtnB)
        self.pushButtonC.pressed.connect(self.BtnC)

    def Thread(self):
        self.threadA = WorkThreadA(self)
        self.threadB = WorkThreadB(self)
        self.threadC = WorkThreadC(self)

    def Signals(self):
        self.threadA.signal_a.connect(self.threadC.handle_signalA)
        self.threadB.signal_b.connect(self.threadC.handle_signalB)

    def BtnA(self):
        self.state = not self.state
        if self.state:
            self.threadA.start()
            self.pushButtonA.setChecked(False)
        else:
            self.threadA.stop()
            self.pushButtonA.setChecked(True)

    def BtnB(self):
        self.state = not self.state
        if self.state:
            self.threadB.start()
            self.pushButtonB.setChecked(False)
        else:
            self.threadB.stop()
            self.pushButtonB.setChecked(True)

    def BtnC(self):

        self.state = not self.state
        if self.state:
            self.threadC.start()
            self.pushButtonC.setChecked(False)
        else:
            self.threadC.stop()
            self.pushButtonC.setChecked(True)



class WorkThreadA(QThread):
     signal_a = pyqtSignal(int)

     def __init__(self,host_window,parent=None):
         super(WorkThreadA, self).__init__(parent)
         self.host=host_window
         self.exiting = False


     def run(self):
         self.exiting = False
         i=0
         while not self.exiting:
            self.signal_a.emit(i)
            print("Thread A:",i)
            i=i+1
            time.sleep(1)

     def stop(self):
         self.exiting = True

class WorkThreadB(QThread):
     signal_b = pyqtSignal(int)

     def __init__(self,host_window,parent=None):
         super(WorkThreadB, self).__init__(parent)
         self.host=host_window
         self.exiting = False


     def run(self):
         self.exiting = False
         i=0
         while not self.exiting:
            self.signal_b.emit(i)
            print("Thread B:",i)
            i=i+1
            time.sleep(1)

     def stop(self):
         self.exiting = True



class WorkThreadC(QThread):
    def __init__(self,host_window,parent=None):
        super(WorkThreadC, self).__init__(parent)
        self.host=host_window
        self.exiting = False


    def run(self):
        self.exiting = False
        while not self.exiting:
            print("Signal A: ",self.signalA)
            print("Signal B: ",self.signalB)
            time.sleep(1)


    def stop(self):
        self.exiting = True

    @pyqtSlot(int)
    def handle_signalA(self, value):
        self.signalA=value

    @pyqtSlot(int)
    def handle_signalB(self, value):
        self.signalB=value

if __name__ == "__main__":
    app = QtGui.QApplication(sys.argv)
    window = MyApp()
    window.show()
    sys.exit(app.exec_())