Tkinter: comment utiliser les threads pour empêcher la boucle de l'événement principal de "geler""

j'ai un petit test GUI avec un bouton" Start " et une barre de progression. Le comportement désiré est:

  • Cliquez Sur Démarrer
  • la barre de progression oscille pendant 5 secondes
  • arrêts Progressbar

le comportement observé est le bouton" Start " gèle pendant 5 secondes, puis une barre de progression est affichée (pas d'oscillation).

Voici mon code jusqu'à présent: 

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é sur les informations de Bryan Oakley ici , je comprends que je dois utiliser des fils. J'ai essayé de créer un thread, mais je devine que puisque le thread est démarré à partir du thread principal, il n'aide pas.

j'ai eu l'idée de placer la partie logique dans une classe différente, et instancier L'interface graphique de cette classe, similaire au code d'exemple par A. Rodas ici .

ma question:

Je ne sais pas comment le coder pour que cette commande: 

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

appelle une fonction qui se trouve dans l'autre classe. Est-ce une Mauvaise Chose à faire ou est-ce même possible? Comment pourrais-je créer une 2e classe qui peut gérer soi-même.tb_click? J'ai essayé de suivre le code d'exemple de A. Rodas qui fonctionne très bien. Mais je ne peux pas comprendre comment mettre en œuvre sa solution dans le cas d'un widget Bouton qui déclenche une action.

si je devais à la place manipuler le thread à partir de la classe GUI simple, comment serait-on créer un thread qui n'interfère pas avec le thread principal?

26
demandé sur Community 2013-05-25 05:16:16

3 réponses

lorsque vous rejoignez le nouveau fil dans le fil principal, il attendra jusqu'à ce que le fil se termine, de sorte que l'interface graphique va bloquer même si vous utilisez multithreading.

si vous voulez placer la partie logique dans une classe différente, vous pouvez sous-classe Thread directement, puis démarrer un nouvel objet de cette classe lorsque vous appuyez sur le bouton. Le constructeur de cette sous-classe de Thread peut recevoir un objet File d'attente et ensuite vous pourrez le communiquer avec la partie GUI. Donc, mon suggestion:

  1. Créer une File d'attente de l'objet dans le thread principal
  2. créer un nouveau thread avec l'accès à cette file d'attente
  3. vérifier périodiquement la file d'attente dans le fil principal

alors vous devez résoudre le problème de ce qui se passe si l'utilisateur clique deux fois le même bouton (il va lancer un nouveau fil à chaque clic), mais vous pouvez le corriger en désactivant le bouton Démarrer et en l'activant à nouveau après avoir appelé 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")
41
répondu A. Rodas 2013-05-25 08:17:41

le problème est que T. join () bloque l'événement de clic, le thread principal ne retourne pas à la boucle d'événement pour traiter les repeints. Voir pourquoi TTK Progressbar apparaît après le processus dans Tkinter ou TTK barre de progrès bloqué lors de l'envoi de l'email

2
répondu jmihalicza 2017-05-23 12:18:34

Je soumettrai la base pour une solution alternative. Elle n'est pas spécifique à une barre de progression des savoirs traditionnels en soi, mais elle peut certainement être mise en œuvre très facilement pour cela.

voici quelques classes qui vous permettent d'exécuter d'autres tâches en arrière-plan de Tk, de mettre à jour les contrôles Tk lorsque vous le souhaitez, et de ne pas verrouiller l'interface graphique!

Voici la classe TkRepeatingTask et 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()

voici un test de Tk qui démontre l'utilisation de ces. Il suffit d'ajouter ceci au bas du module avec ces classes si vous voulez voir la démo en action: 

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()

deux points d'importation je vais insister sur BackgroundTask:

1) la fonction que vous exécutez dans la tâche de fond doit prendre un pointeur de fonction qu'elle invoquera et respectera, ce qui permet à la tâche d'être annulée à mi - chemin à travers-si possible.

2) vous devez vous assurer que la tâche de fond est arrêté lorsque vous quittez votre application. Ce fil fonctionnera même si votre interface graphique est fermée si vous ne l'adressez pas!

1
répondu BuvinJ 2017-01-30 23:37:14