Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Tkinter: How to use threads to preventing main event loop from "freezing"

I have a small GUI test with a "Start" button and a Progress bar. The desired behavior is:

  • Click Start
  • Progressbar oscillates for 5 seconds
  • Progressbar stops

The observed behavior is the "Start" button freezes for 5 seconds, then a Progressbar is displayed (no oscillation).

Here is my code so far:

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

Based on the information from Bryan Oakley here, I understand that I need to use threads. I tried creating a thread, but I'm guessing that since the thread is started from within the main thread, it doesn't help.

I had the idea to place the logic portion in a different class, and instantiate the GUI from within that class, similar to the example code by A. Rodas here.

My question:

I can't figure out how to code it so that this command:

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

calls a function that is located in the other class. Is this a Bad Thing to do or is it even possible? How would I create a 2nd class that can handle the self.tb_click? I tried following along to A. Rodas' example code which works beautifully. But I cannot figure out how to implement his solution in the case of a Button widget that triggers an action.

If I should instead handle the thread from within the single GUI class, how would one create a thread that doesn't interfere with the main thread?

like image 430
Dirty Penguin Avatar asked May 25 '13 01:05

Dirty Penguin


People also ask

How do you stop a thread from looping in Python?

Threaded stoppable function. Instead of subclassing threading. Thread , one can modify the function to allow stopping by a flag. We need an object, accessible to running function, to which we set the flag to stop running.

What is tkinter event loop?

A Tkinter application runs most of its time inside an event loop, which is entered via the mainloop method. It waiting for events to happen. Events can be key presses or mouse operations by the user. Tkinter provides a mechanism to let the programmer deal with events.


2 Answers

When you join the new thread in the main thread, it will wait until the thread finishes, so the GUI will block even though you are using multithreading.

If you want to place the logic portion in a different class, you can subclass Thread directly, and then start a new object of this class when you press the button. The constructor of this subclass of Thread can receive a Queue object and then you will be able to communicate it with the GUI part. So my suggestion is:

  1. Create a Queue object in the main thread
  2. Create a new thread with access to that queue
  3. Check periodically the queue in the main thread

Then you have to solve the problem of what happens if the user clicks two times the same button (it will spawn a new thread with each click), but you can fix it by disabling the start button and enabling it again after you call 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_nowait()             # 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):         super().__init__()         self.queue = queue     def run(self):         time.sleep(5)  # Simulate long running process         self.queue.put("Task finished") 
like image 113
A. Rodas Avatar answered Oct 09 '22 20:10

A. Rodas


I will submit the basis for an alternate solution. It is not specific to a Tk progress bar per se, but it can certainly be implemented very easily for that.

Here are some classes that allow you to run other tasks in the background of Tk, update the Tk controls when desired, and not lock up the gui!

Here's class TkRepeatingTask and 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() 

Here's a Tk test which demos the use of these. Just append this to the bottom of the module with those classes in it if you want to see the demo in 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() 

Two import points I'll stress about BackgroundTask:

1) The function you run in the background task needs to take a function pointer it will both invoke and respect, which allows the task to be cancelled mid way through - if possible.

2) You need to make sure the background task is stopped when you exit your application. That thread will still run even if your gui is closed if you don't address that!

like image 37
BuvinJ Avatar answered Oct 09 '22 21:10

BuvinJ