Tkinter: cómo utilizar los hilos para evitar que el bucle del evento principal se “congele”

Tengo una pequeña prueba de GUI con un botón de “Inicio” y una barra de progreso. El comportamiento deseado es:

  • Haga clic en Inicio
  • La barra de progreso oscila durante 5 segundos
  • Paradas de la barra de progreso

El comportamiento observado es que el botón “Inicio” se congela durante 5 segundos, luego se muestra una barra de progreso (sin swing).

Aquí está mi código hasta ahora:

class GUI: def __init__(self, master): self.master = master self.test_button = Button(self.master, command=self.tb_click) self.test_button.configure( text="Start", background="Grey", padx=50 ) self.test_button.pack(side=TOP) def progress(self): self.prog_bar = ttk.Progressbar( self.master, orient="horizontal", length=200, mode="indeterminate" ) self.prog_bar.pack(side=TOP) def tb_click(self): self.progress() self.prog_bar.start() # Simulate long running process t = threading.Thread(target=time.sleep, args=(5,)) t.start() t.join() self.prog_bar.stop() root = Tk() root.title("Test Button") main_ui = GUI(root) root.mainloop() 

Basándome en la información de Bryan Oakley aquí , entiendo que necesito usar hilos. Intenté crear un hilo, pero supongo que dado que el hilo se inicia desde el hilo principal, no ayuda.

Tuve la idea de colocar la parte lógica en una clase diferente y crear una instancia de la GUI desde esa clase, similar al código de ejemplo de A. Rodas aquí .

Mi pregunta:

No puedo averiguar cómo codificarlo para que este comando:

 self.test_button = Button(self.master, command=self.tb_click) 

llama a una función que se encuentra en la otra clase. ¿Es esto algo malo que hacer o incluso es posible? ¿Cómo crearía una segunda clase que pueda manejar el self.tb_click? Intenté seguir el código de ejemplo de A. Rodas que funciona a la perfección. Pero no puedo descubrir cómo implementar su solución en el caso de un widget de botón que activa una acción.

Si, por el contrario, debo manejar el subproceso desde la única clase de GUI, ¿cómo crearía uno que no interfiera con el subproceso principal?

Cuando se une al nuevo hilo en el hilo principal, esperará hasta que el hilo termine, por lo que la GUI se bloqueará aunque esté usando multihilo.

Si desea colocar la parte lógica en una clase diferente, puede crear subclases de subprocesos directamente y luego comenzar un nuevo objeto de esta clase cuando presione el botón. El constructor de esta subclase de Thread puede recibir un objeto Queue y luego podrá comunicarlo con la parte GUI. Así que mi sugerencia es:

  1. Crear un objeto de cola en el hilo principal
  2. Crear un nuevo hilo con acceso a esa cola
  3. Compruebe periódicamente la cola en el hilo principal

Luego, tiene que resolver el problema de lo que sucede si el usuario hace clic dos veces en el mismo botón (generará un nuevo hilo con cada clic), pero puede solucionarlo desactivando el botón de inicio y habilitándolo nuevamente después de llamarse a self.prog_bar.stop() .

 import Queue class GUI: # ... def tb_click(self): self.progress() self.prog_bar.start() self.queue = Queue.Queue() ThreadedTask(self.queue).start() self.master.after(100, self.process_queue) def process_queue(self): try: msg = self.queue.get(0) # Show result of the task if needed self.prog_bar.stop() except Queue.Empty: self.master.after(100, self.process_queue) class ThreadedTask(threading.Thread): def __init__(self, queue): threading.Thread.__init__(self) self.queue = queue def run(self): time.sleep(5) # Simulate long running process self.queue.put("Task finished") 

El problema es que t.join () bloquea el evento de clic, el hilo principal no regresa al ciclo de eventos para procesar las repeticiones. Vea Por qué aparece la barra de progreso ttk después del proceso en Tkinter o la barra de progreso TTK bloqueada al enviar correo electrónico

Voy a presentar la base para una solución alternativa. No es específico de una barra de progreso de Tk per se, pero ciertamente se puede implementar muy fácilmente para eso.

Aquí hay algunas clases que le permiten ejecutar otras tareas en el fondo de Tk, actualizar los controles de Tk cuando lo desee y ¡no bloquear la interfaz gráfica de usuario!

Aquí está la clase TkRepeatingTask y BackgroundTask:

 import threading class TkRepeatingTask(): def __init__( self, tkRoot, taskFuncPointer, freqencyMillis ): self.__tk_ = tkRoot self.__func_ = taskFuncPointer self.__freq_ = freqencyMillis self.__isRunning_ = False def isRunning( self ) : return self.__isRunning_ def start( self ) : self.__isRunning_ = True self.__onTimer() def stop( self ) : self.__isRunning_ = False def __onTimer( self ): if self.__isRunning_ : self.__func_() self.__tk_.after( self.__freq_, self.__onTimer ) class BackgroundTask(): def __init__( self, taskFuncPointer ): self.__taskFuncPointer_ = taskFuncPointer self.__workerThread_ = None self.__isRunning_ = False def taskFuncPointer( self ) : return self.__taskFuncPointer_ def isRunning( self ) : return self.__isRunning_ and self.__workerThread_.isAlive() def start( self ): if not self.__isRunning_ : self.__isRunning_ = True self.__workerThread_ = self.WorkerThread( self ) self.__workerThread_.start() def stop( self ) : self.__isRunning_ = False class WorkerThread( threading.Thread ): def __init__( self, bgTask ): threading.Thread.__init__( self ) self.__bgTask_ = bgTask def run( self ): try : self.__bgTask_.taskFuncPointer()( self.__bgTask_.isRunning ) except Exception as e: print repr(e) self.__bgTask_.stop() 

Aquí hay una prueba de Tk que demuestra el uso de estos. Simplemente agregue esto al final del módulo con esas clases en él si desea ver la demostración en acción:

 def tkThreadingTest(): from tkinter import Tk, Label, Button, StringVar from time import sleep class UnitTestGUI: def __init__( self, master ): self.master = master master.title( "Threading Test" ) self.testButton = Button( self.master, text="Blocking", command=self.myLongProcess ) self.testButton.pack() self.threadedButton = Button( self.master, text="Threaded", command=self.onThreadedClicked ) self.threadedButton.pack() self.cancelButton = Button( self.master, text="Stop", command=self.onStopClicked ) self.cancelButton.pack() self.statusLabelVar = StringVar() self.statusLabel = Label( master, textvariable=self.statusLabelVar ) self.statusLabel.pack() self.clickMeButton = Button( self.master, text="Click Me", command=self.onClickMeClicked ) self.clickMeButton.pack() self.clickCountLabelVar = StringVar() self.clickCountLabel = Label( master, textvariable=self.clickCountLabelVar ) self.clickCountLabel.pack() self.threadedButton = Button( self.master, text="Timer", command=self.onTimerClicked ) self.threadedButton.pack() self.timerCountLabelVar = StringVar() self.timerCountLabel = Label( master, textvariable=self.timerCountLabelVar ) self.timerCountLabel.pack() self.timerCounter_=0 self.clickCounter_=0 self.bgTask = BackgroundTask( self.myLongProcess ) self.timer = TkRepeatingTask( self.master, self.onTimer, 1 ) def close( self ) : print "close" try: self.bgTask.stop() except: pass try: self.timer.stop() except: pass self.master.quit() def onThreadedClicked( self ): print "onThreadedClicked" try: self.bgTask.start() except: pass def onTimerClicked( self ) : print "onTimerClicked" self.timer.start() def onStopClicked( self ) : print "onStopClicked" try: self.bgTask.stop() except: pass try: self.timer.stop() except: pass def onClickMeClicked( self ): print "onClickMeClicked" self.clickCounter_+=1 self.clickCountLabelVar.set( str(self.clickCounter_) ) def onTimer( self ) : print "onTimer" self.timerCounter_+=1 self.timerCountLabelVar.set( str(self.timerCounter_) ) def myLongProcess( self, isRunningFunc=None ) : print "starting myLongProcess" for i in range( 1, 10 ): try: if not isRunningFunc() : self.onMyLongProcessUpdate( "Stopped!" ) return except : pass self.onMyLongProcessUpdate( i ) sleep( 1.5 ) # simulate doing work self.onMyLongProcessUpdate( "Done!" ) def onMyLongProcessUpdate( self, status ) : print "Process Update: %s" % (status,) self.statusLabelVar.set( str(status) ) root = Tk() gui = UnitTestGUI( root ) root.protocol( "WM_DELETE_WINDOW", gui.close ) root.mainloop() if __name__ == "__main__": tkThreadingTest() 

Dos puntos de importación que enfatizaré sobre BackgroundTask:

1) La función que ejecuta en la tarea en segundo plano debe tomar un puntero de función que invocará y respetará, lo que permite que la tarea se cancele a mitad de camino, si es posible.

2) Debe asegurarse de que la tarea en segundo plano se detiene al salir de la aplicación. ¡Ese hilo aún se ejecutará incluso si tu gui está cerrada si no lo abordas!