Skip to main content
se eliminaron 4 caracteres en el cuerpo
Origen Enlace
FJSevilla
  • 62.1k
  • 7
  • 43
  • 68
import sys
import time
from PyQt5 import QtCore, QtWidgets



class Worker(QtCore.QObject):
    # Señal que nos permitirá transferir información entre los hilos
    resultado = QtCore.pyqtSignal(int)
    
    def __init__(self):
        super().__init__()
    
    @QtCore.pyqtSlot()
    def iniciar_conexion(self):
        self.resultado.emit(1)
        '''Función que hará el trabajo bloqueante en el hilo'''
        time.sleep(105) # Simulamos proceso bloqueante
        self.resultado.emit(-1)


class App(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Ejemplo QThread')
        self.setGeometry(100, 100, 400, 300)

        self.boton_iniciar = QtWidgets.QPushButton('Conectar', self)
        self.boton_iniciar.move(100150, 120150) 
        self.boton_iniciar.clicked.connect(self.conectar)
        
        self.label_estado = QtWidgets.QLabel(self)
        self.label_estado.setGeometry(QtCore.QRect(140160, 18080,  400, 50))
        self.label_estado.setText('Sin conectar')
        
        self._thread = None
        self._worker = None


    
    @QtCore.pyqtSlot()
    def conectar(self):
        """
        Slot encargado de lanzar el hilo
        """
        self.boton_iniciar.setEnabled(False)
        # Instanciamos un QThread
        self._thread = QtCore.QThread()
        # Instanciamos nuestra clase worker
        self._worker = Worker()
        self._thread.started.connect(self._worker.iniciar_conexion)
        # Movemos el worquer al hilo
        self._worker.moveToThread(self._thread)
        # Conectamos la señal del worker con un slot encargado de actualizar ella labelgui
        self._worker.resultado.connect(self._estado)
        self._thread.start() 
        
        
    @QtCore.pyqtSlot(int)
    def _estado(self, res):
        if res == 1:  # Conexión en proceso
            self.label_estado.setText("Conectando...")
            return
            
        elif res == 0:    # Conexión exitosa
            self.label_estado.setText("Conexión completada")
            self._thread.quit()
            self._thread.wait()
            
        else:           # Conexión fallida
            self.label_estado.setText("Conexión fallida")
            self.boton_iniciar.setText("Reintentar")
            self.boton_iniciar.setEnabled(True)
            
        self._thread.quit()
        self._thread.wait()
 
        

if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    ventana = App()
    ventana.show()
    sys.exit(app.exec_())
import sys
import time
from PyQt5 import QtCore, QtWidgets



class Worker(QtCore.QObject):
    # Señal que nos permitirá transferir información entre los hilos
    resultado = QtCore.pyqtSignal(int)
    
    def __init__(self):
        super().__init__()
    
    @QtCore.pyqtSlot()
    def iniciar_conexion(self):
        self.resultado.emit(1)
        '''Función que hará el trabajo bloqueante en el hilo'''
        time.sleep(10) # Simulamos proceso bloqueante
        self.resultado.emit(-1)


class App(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Ejemplo QThread')
        self.setGeometry(100, 100, 400, 300)

        self.boton_iniciar = QtWidgets.QPushButton('Conectar', self)
        self.boton_iniciar.move(100, 120) 
        self.boton_iniciar.clicked.connect(self.conectar)
        
        self.label_estado = QtWidgets.QLabel(self)
        self.label_estado.setGeometry(QtCore.QRect(140, 180,  400, 50))
        self.label_estado.setText('Sin conectar')
        
        self._thread = None
        self._worker = None


    
    @QtCore.pyqtSlot()
    def conectar(self):
        """
        Slot encargado de lanzar el hilo
        """
        self.boton_iniciar.setEnabled(False)
        # Instanciamos un QThread
        self._thread = QtCore.QThread()
        # Instanciamos nuestra clase worker
        self._worker = Worker()
        self._thread.started.connect(self._worker.iniciar_conexion)
        # Movemos el worquer al hilo
        self._worker.moveToThread(self._thread)
        # Conectamos la señal del worker con un slot encargado de actualizar el label
        self._worker.resultado.connect(self._estado)
        self._thread.start() 
        
        
    @QtCore.pyqtSlot(int)
    def _estado(self, res):
        if res == 1:  # Conexión en proceso
            self.label_estado.setText("Conectando...")
            return
            
        elif res == 0:    # Conexión exitosa
            self.label_estado.setText("Conexión completada")
            self._thread.quit()
            self._thread.wait()
            
        else:           # Conexión fallida
            self.label_estado.setText("Conexión fallida")
            self.boton_iniciar.setText("Reintentar")
            self.boton_iniciar.setEnabled(True)
            
        self._thread.quit()
        self._thread.wait()
 


if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    ventana = App()
    ventana.show()
    sys.exit(app.exec_())
import sys
import time
from PyQt5 import QtCore, QtWidgets



class Worker(QtCore.QObject):
    # Señal que nos permitirá transferir información entre los hilos
    resultado = QtCore.pyqtSignal(int)
    
    def __init__(self):
        super().__init__()
    
    @QtCore.pyqtSlot()
    def iniciar_conexion(self):
        self.resultado.emit(1)
        '''Función que hará el trabajo bloqueante en el hilo'''
        time.sleep(5) # Simulamos proceso bloqueante
        self.resultado.emit(-1)


class App(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Ejemplo QThread')
        self.setGeometry(100, 100, 400, 300)

        self.boton_iniciar = QtWidgets.QPushButton('Conectar', self)
        self.boton_iniciar.move(150, 150) 
        self.boton_iniciar.clicked.connect(self.conectar)
        
        self.label_estado = QtWidgets.QLabel(self)
        self.label_estado.setGeometry(160, 80,  400, 50)
        self.label_estado.setText('Sin conectar')
        
        self._thread = None
        self._worker = None


    
    @QtCore.pyqtSlot()
    def conectar(self):
        """
        Slot encargado de lanzar el hilo
        """
        self.boton_iniciar.setEnabled(False)
        # Instanciamos un QThread
        self._thread = QtCore.QThread()
        # Instanciamos nuestra clase worker
        self._worker = Worker()
        self._thread.started.connect(self._worker.iniciar_conexion)
        # Movemos el worquer al hilo
        self._worker.moveToThread(self._thread)
        # Conectamos la señal del worker con un slot encargado de actualizar la gui
        self._worker.resultado.connect(self._estado)
        self._thread.start() 
        
        
    @QtCore.pyqtSlot(int)
    def _estado(self, res):
        if res == 1:  # Conexión en proceso
            self.label_estado.setText("Conectando...")
            return
            
        elif res == 0:    # Conexión exitosa
            self.label_estado.setText("Conexión completada")
            self._thread.quit()
            self._thread.wait()
            
        else:           # Conexión fallida
            self.label_estado.setText("Conexión fallida")
            self.boton_iniciar.setText("Reintentar")
            self.boton_iniciar.setEnabled(True)
            
        self._thread.quit()
        self._thread.wait()
        

if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    ventana = App()
    ventana.show()
    sys.exit(app.exec_())
Origen Enlace
FJSevilla
  • 62.1k
  • 7
  • 43
  • 68

Primero dos observaciones:

  • No interactúes jamás directamente con un widget desde un hilo/proceso que no sea el principal. Solo se debe interactuar con la GUI desde su propio hilo. Lo que si puedes es informar mediante métodos seguros al hilo principal desde el hilo hijo para que el primero haga los cambios necesarios en la interfaz. Una forma muy simple es usar señales.

  • Qt (y PyQt por supuesto) tiene su propio mecanismo para crear hilos, QThread. Deberías usarlo en vez de los hilos nativos de Python, lo cual permite, por ejemplo, establecer una comunicación mediante señales entre los hilos de forma simple y segura.

Dicho esto, un ejemplo reproducible y simplificado de lo que deseas podría ser el siguiente:

import sys
import time
from PyQt5 import QtCore, QtWidgets



class Worker(QtCore.QObject):
    # Señal que nos permitirá transferir información entre los hilos
    resultado = QtCore.pyqtSignal(int)
    
    def __init__(self):
        super().__init__()
    
    @QtCore.pyqtSlot()
    def iniciar_conexion(self):
        self.resultado.emit(1)
        '''Función que hará el trabajo bloqueante en el hilo'''
        time.sleep(10) # Simulamos proceso bloqueante
        self.resultado.emit(-1)


class App(QtWidgets.QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('Ejemplo QThread')
        self.setGeometry(100, 100, 400, 300)

        self.boton_iniciar = QtWidgets.QPushButton('Conectar', self)
        self.boton_iniciar.move(100, 120) 
        self.boton_iniciar.clicked.connect(self.conectar)
        
        self.label_estado = QtWidgets.QLabel(self)
        self.label_estado.setGeometry(QtCore.QRect(140, 180,  400, 50))
        self.label_estado.setText('Sin conectar')
        
        self._thread = None
        self._worker = None


    
    @QtCore.pyqtSlot()
    def conectar(self):
        """
        Slot encargado de lanzar el hilo
        """
        self.boton_iniciar.setEnabled(False)
        # Instanciamos un QThread
        self._thread = QtCore.QThread()
        # Instanciamos nuestra clase worker
        self._worker = Worker()
        self._thread.started.connect(self._worker.iniciar_conexion)
        # Movemos el worquer al hilo
        self._worker.moveToThread(self._thread)
        # Conectamos la señal del worker con un slot encargado de actualizar el label
        self._worker.resultado.connect(self._estado)
        self._thread.start() 
        
        
    @QtCore.pyqtSlot(int)
    def _estado(self, res):
        if res == 1:  # Conexión en proceso
            self.label_estado.setText("Conectando...")
            return
            
        elif res == 0:    # Conexión exitosa
            self.label_estado.setText("Conexión completada")
            self._thread.quit()
            self._thread.wait()
            
        else:           # Conexión fallida
            self.label_estado.setText("Conexión fallida")
            self.boton_iniciar.setText("Reintentar")
            self.boton_iniciar.setEnabled(True)
            
        self._thread.quit()
        self._thread.wait()



if __name__ == '__main__':
    app = QtWidgets.QApplication(sys.argv)
    ventana = App()
    ventana.show()
    sys.exit(app.exec_())