Comment utiliser la bibliothèque PyQt de Python pour gérer les threads

PyQt fournit un cadre robuste pour créer des interfaces utilisateur graphiques dans les applications Python. Pour garantir des mises à jour fluides et réactives de l'interface utilisateur lors de l'exécution de tâches en arrière-plan, Qt utilise des threads qui s'exécutent simultanément avec le thread principal de l'interface graphique. Cependant, l'utilisation efficace des threads nécessite un examen attentif.

Comme indiqué dans l'article de Maya Posch, la réimplémentation de la méthode run ne devrait pas être l'approche privilégiée lorsque l'on travaille avec QThreads. Pensez plutôt à utiliser des signaux et des emplacements pour faciliter la communication entre les threads. Pour illustrer la mise en œuvre correcte, un exemple pratique sera présenté.

Exemple de gestion des threads PyQt

Dans l'exemple, nous créons un thread de travail distinct qui gère les calculs longs tandis que le thread principal de l'interface graphique gère l'interface graphique. Le thread de travail communique les mises à jour d'état à l'interface graphique via des signaux.

Pour démarrer le calcul, l'utilisateur clique sur le bouton "Démarrer". Le bouton "Annuler" peut être utilisé pour terminer l'opération et réinitialiser le thread de travail. Veuillez noter que l'arrêt forcé des threads n'est généralement pas recommandé mais est utilisé à des fins de démonstration.

Code Python

from PyQt4 import QtGui, QtCore
import sys
import random

class Example(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

        # Create a gui object.
        self.gui = Window()

        # Create a new worker thread.
        self.createWorkerThread()

        # Make any cross object connections.
        self._connectSignals()

        self.gui.show()


    def _connectSignals(self):
        self.gui.button_cancel.clicked.connect(self.forceWorkerReset)
        self.signalStatus.connect(self.gui.updateStatus)
        self.parent().aboutToQuit.connect(self.forceWorkerQuit)


    def createWorkerThread(self):

        # Setup the worker object and the worker_thread.
        self.worker = WorkerObject()
        self.worker_thread = QtCore.QThread()
        self.worker.moveToThread(self.worker_thread)
        self.worker_thread.start()

        # Connect any worker signals
        self.worker.signalStatus.connect(self.gui.updateStatus)
        self.gui.button_start.clicked.connect(self.worker.startWork)


    def forceWorkerReset(self):      
        if self.worker_thread.isRunning():
            print('Terminating thread.')
            self.worker_thread.terminate()

            print('Waiting for thread termination.')
            self.worker_thread.wait()

            self.signalStatus.emit('Idle.')

            print('building new working object.')
            self.createWorkerThread()


    def forceWorkerQuit(self):
        if self.worker_thread.isRunning():
            self.worker_thread.terminate()
            self.worker_thread.wait()


class WorkerObject(QtCore.QObject):

    signalStatus = QtCore.pyqtSignal(str)

    def __init__(self, parent=None):
        super(self.__class__, self).__init__(parent)

    @QtCore.pyqtSlot()        
    def startWork(self):
        for ii in range(7):
            number = random.randint(0,5000**ii)
            self.signalStatus.emit('Iteration: {}, Factoring: {}'.format(ii, number))
            factors = self.primeFactors(number)
            print('Number: ', number, 'Factors: ', factors)
        self.signalStatus.emit('Idle.')

    def primeFactors(self, n):
        i = 2
        factors = []
        while i * i  1:
            factors.append(n)
        return factors


class Window(QtGui.QWidget):

    def __init__(self):
        QtGui.QWidget.__init__(self)
        self.button_start = QtGui.QPushButton('Start', self)
        self.button_cancel = QtGui.QPushButton('Cancel', self)
        self.label_status = QtGui.QLabel('', self)

        layout = QtGui.QVBoxLayout(self)
        layout.addWidget(self.button_start)
        layout.addWidget(self.button_cancel)
        layout.addWidget(self.label_status)

        self.setFixedSize(400, 200)

    @QtCore.pyqtSlot(str)
    def updateStatus(self, status):
        self.label_status.setText(status)


if __name__=='__main__':
    app = QtGui.QApplication(sys.argv)
    example = Example(app)
    sys.exit(app.exec_())

Cet exemple démontre l'utilisation appropriée de QThreads dans une application PyQt, permettant des opérations en arrière-plan efficaces sans geler l'interface utilisateur.