If you use GUIs in Python much, then you know that sometimes you need to execute some long running process every now and then. Of course, if you do that as you would with a command line program, then you'll be in for a surprise. In most cases, you'll end up blocking your GUI's event loop and the user will see your program freeze. What can you do to get around just mishaps? Start the task in another thread or process of course! In this article, we'll look at how to do this with wxPython and Python's threading module.
In the wxPython world, there are three related "threadsafe" methods. If you do not use one of these three when you go to update your user interface, then you may experience weird issues. Sometimes your GUI will work just fine. Other times, it will crash Python for no apparent reason. Thus the need for the threadsafe methods. Here are the three thread-safe methods that wxPython provides:
According to Robin Dunn (creator of wxPython), wx.CallAfter uses wx.PostEvent to send an event to the application object. The application will have an event handler bound to that event and will react according to whatever the programmer has coded upon receipt of the event. It is my understanding that wx.CallLater calls wx.CallAfter with a specified time limit so that you can tell it how long to wait before sending the event.
Robin Dunn also pointed out that the Python Global Interpreter Lock (GIL) will prevent more than one thread to be executing Python bytecodes at the same time, which may limit how many CPU cores are utilized by your program. On the flip-side, he also said that "wxPython releases the GIL while making calls to wx APIs so other threads can run at that time". In other words, your mileage may vary when using threads on multicore machines. I found this discussion to be interesting and confusing...
Anyway, what this means in regard to the three wx-methods is that wx.CallLater is the most abstract threadsafe method with wx.CallAfter next and wx.PostEvent being the most low-level. In the following examples, you will see how to use wx.CallAfter and wx.PostEvent to update your wxPython program.
On the wxPython mailing list, you'll see the experts telling others to use wx.CallAfter along with PubSub to communicate with their wxPython applications from another thread. I've probably even told people to do that. So in the following example, that's exactly what we're going to do:
import time
import wx
from threading import Thread
from wx.lib.pubsub import Publisher
########################################################################
class TestThread(Thread):
"""Test Worker Thread Class."""
#----------------------------------------------------------------------
def __init__(self):
"""Init Worker Thread Class."""
Thread.__init__(self)
self.start() # start the thread
#----------------------------------------------------------------------
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
for i in range(6):
time.sleep(10)
wx.CallAfter(self.postTime, i)
time.sleep(5)
wx.CallAfter(Publisher().sendMessage, "update", "Thread finished!")
#----------------------------------------------------------------------
def postTime(self, amt):
"""
Send time to GUI
"""
amtOfTime = (amt + 1) * 10
Publisher().sendMessage("update", amtOfTime)
########################################################################
class MyForm(wx.Frame):
#----------------------------------------------------------------------
def __init__(self):
wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial")
# Add a panel so it looks the correct on all platforms
panel = wx.Panel(self, wx.ID_ANY)
self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here")
self.btn = btn = wx.Button(panel, label="Start Thread")
btn.Bind(wx.EVT_BUTTON, self.onButton)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5)
sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5)
panel.SetSizer(sizer)
# create a pubsub receiver
Publisher().subscribe(self.updateDisplay, "update")
#----------------------------------------------------------------------
def onButton(self, event):
"""
Runs the thread
"""
TestThread()
self.displayLbl.SetLabel("Thread started!")
btn = event.GetEventObject()
btn.Disable()
#----------------------------------------------------------------------
def updateDisplay(self, msg):
"""
Receives data from thread and updates the display
"""
t = msg.data
if isinstance(t, int):
self.displayLbl.SetLabel("Time since thread started: %s seconds" % t)
else:
self.displayLbl.SetLabel("%s" % t)
self.btn.Enable()
#----------------------------------------------------------------------
# Run the program
if __name__ == "__main__":
app = wx.PySimpleApp()
frame = MyForm().Show()
app.MainLoop()
We'll be using Python's time module to fake our long running process. However, feel free to put something better in its place. In a real life example, I use a thread to open Adobe Reader and send a PDF to a printer. That might not seem like anything special, but when I didn't use a thread, the print button in my application would stay stuck down while the document was sent to the printer and my GUI just hung until that was done. Even a a second or two is noticeable to the user!
Anyway, let's see how this works. In our thread class (reproduced below), we override the "run" method so it does what we want. This thread is started when we instantiate it because we have "self.start()" in its __init__ method. In the "run" method, we loop over a range of 6, sleeping for 10 seconds in between iterations and then update our user interface using wx.CallAfter and PubSub. When the loop finishes, we send a final message to our application to let the user know what happened.
########################################################################
class TestThread(Thread):
"""Test Worker Thread Class."""
#----------------------------------------------------------------------
def __init__(self):
"""Init Worker Thread Class."""
Thread.__init__(self)
self.start() # start the thread
#----------------------------------------------------------------------
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
for i in range(6):
time.sleep(10)
wx.CallAfter(self.postTime, i)
time.sleep(5)
wx.CallAfter(Publisher().sendMessage, "update", "Thread finished!")
#----------------------------------------------------------------------
def postTime(self, amt):
"""
Send time to GUI
"""
amtOfTime = (amt + 1) * 10
Publisher().sendMessage("update", amtOfTime)
You'll notice that in our wxPython code, we start the thread using a button event handler. We also disable the button so we don't accidentally start additional threads. That would be pretty confusing if we had a bunch of them going and the UI would randomly say that it was done when it wasn't. That is a good exercise for the reader though. You could display the PID of the thread so you'd know which was which...and you might want to output this information to a scrolling text control so you can see the activity of the various threads.
The last piece of interest here is probably the PubSub receiver and its event handler:
def updateDisplay(self, msg):
"""
Receives data from thread and updates the display
"""
t = msg.data
if isinstance(t, int):
self.displayLbl.SetLabel("Time since thread started: %s seconds" % t)
else:
self.displayLbl.SetLabel("%s" % t)
self.btn.Enable()
See how we extract the message from the thread and use it to update our display? We also use the type of data we receive to tell us what to show the user. Pretty cool, huh? Now let's go down a level and check out how to do it with wx.PostEvent instead.
The following code is based on an example from the wxPython wiki. It's a little bit more complicated than the wx.CallAfter code we just looked at, but I'm confident that we can figure it out.
import time
import wx
from threading import Thread
# Define notification event for thread completion
EVT_RESULT_ID = wx.NewId()
def EVT_RESULT(win, func):
"""Define Result Event."""
win.Connect(-1, -1, EVT_RESULT_ID, func)
class ResultEvent(wx.PyEvent):
"""Simple event to carry arbitrary result data."""
def __init__(self, data):
"""Init Result Event."""
wx.PyEvent.__init__(self)
self.SetEventType(EVT_RESULT_ID)
self.data = data
########################################################################
class TestThread(Thread):
"""Test Worker Thread Class."""
#----------------------------------------------------------------------
def __init__(self, wxObject):
"""Init Worker Thread Class."""
Thread.__init__(self)
self.wxObject = wxObject
self.start() # start the thread
#----------------------------------------------------------------------
def run(self):
"""Run Worker Thread."""
# This is the code executing in the new thread.
for i in range(6):
time.sleep(10)
amtOfTime = (i + 1) * 10
wx.PostEvent(self.wxObject, ResultEvent(amtOfTime))
time.sleep(5)
wx.PostEvent(self.wxObject, ResultEvent("Thread finished!"))
########################################################################
class MyForm(wx.Frame):
#----------------------------------------------------------------------
def __init__(self):
wx.Frame.__init__(self, None, wx.ID_ANY, "Tutorial")
# Add a panel so it looks the correct on all platforms
panel = wx.Panel(self, wx.ID_ANY)
self.displayLbl = wx.StaticText(panel, label="Amount of time since thread started goes here")
self.btn = btn = wx.Button(panel, label="Start Thread")
btn.Bind(wx.EVT_BUTTON, self.onButton)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.displayLbl, 0, wx.ALL|wx.CENTER, 5)
sizer.Add(btn, 0, wx.ALL|wx.CENTER, 5)
panel.SetSizer(sizer)
# Set up event handler for any worker thread results
EVT_RESULT(self, self.updateDisplay)
#----------------------------------------------------------------------
def onButton(self, event):
"""
Runs the thread
"""
TestThread(self)
self.displayLbl.SetLabel("Thread started!")
btn = event.GetEventObject()
btn.Disable()
#----------------------------------------------------------------------
def updateDisplay(self, msg):
"""
Receives data from thread and updates the display
"""
t = msg.data
if isinstance(t, int):
self.displayLbl.SetLabel("Time since thread started: %s seconds" % t)
else:
self.displayLbl.SetLabel("%s" % t)
self.btn.Enable()
#----------------------------------------------------------------------
# Run the program
if __name__ == "__main__":
app = wx.PySimpleApp()
frame = MyForm().Show()
app.MainLoop()
Let's break this down a bit. For me, the most confusing stuff is the first three pieces:
# Define notification event for thread completion
EVT_RESULT_ID = wx.NewId()
def EVT_RESULT(win, func):
"""Define Result Event."""
win.Connect(-1, -1, EVT_RESULT_ID, func)
class ResultEvent(wx.PyEvent):
"""Simple event to carry arbitrary result data."""
def __init__(self, data):
"""Init Result Event."""
wx.PyEvent.__init__(self)
self.SetEventType(EVT_RESULT_ID)
self.data = data
The EVT_RESULT_ID is the key here. It links the thread to the wx.PyEvent and that weird "EVT_RESULT" function. In the wxPython code, we bind an event handler to the EVT_RESULT function. This allows us to us wx.PostEvent in the thread to send an event to our custom event class, ResultEvent. What does this do? It sends the data on to the wxPython program by emitting that custom EVT_RESULT that we bound to. I hope that all makes sense.
Once you've got that figured out in your head, read on. Are you ready? Good! You'll notice that our TestThread class is pretty much the same as before except that we're using wx.PostEvent to send our messages to the GUI instead of PubSub. The API in our GUI's display updater is unchanged. We still just use the message's data property to extract the data we want. That's all there is to it!
Hopefully you now know how to use basic threading techniques in your wxPython programs. There are several other threading methods too which we didn't have a chance to cover here, such as using wx.Yield or Queues. Fortunately, the wxPython wiki covers these topics pretty well, so be sure to check out the links below if you're interested in those methods.
Copyright © 2024 Mouse Vs Python | Powered by Pythonlibrary