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_())
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_())
lang-py