qthreaddemo.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from PyQt4 import QtGui, QtCore
import time
import sys
import random

class WorkerThread(QtCore.QThread):
  # Worker Thread
  # We define some signals to connect to. I choose to implement them in the
  # worker thread, but we could also implement them in the main thread
  # I use one signal to realy the progression, another to notify failures.
  advanced = QtCore.pyqtSignal(int)
  error_raised = QtCore.pyqtSignal(str)
  
  def __init__(self, app, steps, raise_error=False):
    # app holds a reference to the parent, so the thread is destroyed if
    # application closes
    # steps is some value for our demo task
    # raise_error is also some value to test raising an error in a thread
    super().__init__(app)
    # Do init steps here
    self.steps = steps
    self.raise_error = raise_error
  
  def run(self):
    # A stupid function to demonstrate something that takes time.
    # The handling of the progress bar is also a little but rough, as
    # we could better define minimum and maximum and not convert into
    # percentages. But as long operations do not always want to report
    # at each iteration (this takes computation time), there may be an algorithm
    # here anyway.
    try:
      for i in range(self.steps):
        time.sleep(1)
        # Signal advancement to connected slots
        self.advanced.emit(i*100/self.steps)
      self.advanced.emit(100)
      if self.raise_error:
        raise RuntimeError("Hey you! Don't")
    # Demonstrate how one could handle failures in a thread
    except RuntimeError as e:
      self.error_raised.emit(str(e))

class SampleApp(QtGui.QApplication):
  # Main application
  def __init__(self):
    super().__init__(sys.argv)
    # Define widgets
    self.mainWindow = QtGui.QWidget()
    self.prProgress1 = QtGui.QProgressBar()
    self.prProgress2 = QtGui.QProgressBar()
    self.vbProgress = QtGui.QVBoxLayout()
    self.btnQuit = QtGui.QPushButton("Close")
    self.btnStart = QtGui.QPushButton("Start")
    self.btnSurprise = QtGui.QPushButton(":)")
    self.vbButtons = QtGui.QVBoxLayout()
    self.hbBox = QtGui.QHBoxLayout()
    
    # Setup layout
    self.vbProgress.addWidget(self.prProgress1)
    self.vbProgress.addWidget(self.prProgress2)
    self.vbButtons.addWidget(self.btnStart)
    self.vbButtons.addWidget(self.btnSurprise)
    self.hbBox.addLayout(self.vbButtons)
    self.hbBox.addLayout(self.vbProgress)
    self.hbBox.addWidget(self.btnQuit)
    self.mainWindow.setLayout(self.hbBox)
    
    # Configure widgets
    self.mainWindow.resize(500, 0)
    self.prProgress1.setMinimum(0)
    self.prProgress2.setMinimum(0)
    self.prProgress1.setMaximum(100)
    self.prProgress2.setMaximum(100)
    
    # Instanciate workers. This may be done elsewhere as needed, not
    # specifically in init, and maybe not handled as separate attributes, but
    # maybe a queue or a list
    self.worker1 = WorkerThread(self, 10, False)
    self.worker2 = WorkerThread(self, 15, True)
    
    # Connect signals: progression, error and finished events
    self.worker1.advanced.connect(self.prProgress1.setValue)
    self.worker2.advanced.connect(self.prProgress2.setValue)
    self.worker1.error_raised.connect(self.onError)
    self.worker2.error_raised.connect(self.onError)
    self.worker1.finished.connect(self.workerFinished)
    self.worker2.finished.connect(self.workerFinished)
    
    # Connect signals of widgets
    self.btnQuit.clicked.connect(self.quit)
    self.btnStart.clicked.connect(self.startWorkers)
    self.btnSurprise.clicked.connect(self.surprise)
  
  def run(self):
    # Main loop
    self.mainWindow.show()
    return self.exec_()
  
  def startWorkers(self):
    # Start the workers
    self.btnQuit.setEnabled(False)
    self.btnStart.setEnabled(False)
    self.worker1.start()
    self.worker2.start()
  
  def onError(self, error_msg):
    # If a thread had an error, display a dialog box to the user
    msgBox = QtGui.QMessageBox()
    msgBox.setText(error_msg)
    msgBox.exec_()
  
  def workerFinished(self):
    # When a thread notifies it has finished, check if we can enable the buttons
    if self.worker1.isFinished() and self.worker2.isFinished():
      self.btnQuit.setEnabled(True)
      self.btnStart.setEnabled(True)
  
  def surprise(self):
    # Demonstrates that the UI is still responsive meanwhile
    self.btnSurprise.setText(random.choice([
      ":S", ":(", ":D", ":P", ":O", ";)"
    ]))

def main():
  app = SampleApp()
  return app.run()

if __name__ == '__main__':
  main()