Cómo utilizar la biblioteca PyQt de Python para administrar subprocesos

PyQt proporciona un marco sólido para crear interfaces gráficas de usuario en aplicaciones Python. Para garantizar actualizaciones de la interfaz de usuario fluidas y receptivas mientras se ejecutan tareas en segundo plano, Qt utiliza subprocesos que se ejecutan simultáneamente con el subproceso principal de la GUI. Sin embargo, emplear subprocesos de manera efectiva requiere una consideración cuidadosa.

Como se describe en el artículo de Maya Posch, la reimplementación del método de ejecución no debería ser el enfoque preferido cuando se trabaja con QThreads. En su lugar, considere utilizar señales y ranuras para facilitar la comunicación entre subprocesos. Para ilustrar la implementación correcta, se presentará un ejemplo práctico.

Ejemplo de gestión de subprocesos de PyQt

En el ejemplo, creamos un subproceso de trabajo independiente que maneja cálculos largos mientras que el hilo principal de la GUI gestiona la interfaz gráfica. El hilo de trabajo comunica actualizaciones de estado a la GUI a través de señales.

Para iniciar el cálculo, el usuario hace clic en el botón "Iniciar". El botón "Cancelar" se puede utilizar para finalizar la operación y restablecer el hilo de trabajo. Tenga en cuenta que generalmente no se recomienda la terminación forzada de subprocesos, pero se utiliza con fines de demostración.

Código 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_())

Este ejemplo demuestra el uso adecuado de QThreads en una aplicación PyQt, lo que permite operaciones eficientes en segundo plano sin congelar la interfaz de usuario.