Search code examples
multithreadingwxpythonwxwidgets

Implementing my own event loop in a wxPython application


I’m writing a wxPython application that will be doing quite a bit of data analysis and display. The way I’ve written it so far has led to problems when two threads try to change something in the GUI at the same time. What I want to do is to set up my own simple queue running on the main thread so that I can ensure that UI updates happen one at a time.

I’m having trouble getting my head around how I’d set up my event loop, though. In general you’d do something like

while True:
    try:
        callback = queue.get(False)
    except Queue.Empty:
        break
    callback()

I assume that if I run that code as-is then WX will not be able to do its thing because it will never receive any events or anything because control never leaves my infinite loop. How can I make this kind of structure coexist with the WX event loop? Or more generally, in a WX application how can I ensure that a certain task is only ever run on the main thread?


Solution

  • You can use wx.callafter, it takes a callable object that is called in the guis mainloop after the current and pending event handlers have been completed. Any extra positional or keyword args are passed on to the callable when it is called.

    Here is an example of gui code that takes advantage of wx.CallAfter when running a separate thread and updating the GUI in the main thread.

    The code is by Andrea Gavana which is found in the wxpython Phoenix docs

    #!/usr/bin/env python
    
    # This sample shows how to take advantage of wx.CallAfter when running a
    # separate thread and updating the GUI in the main thread
    
    import wx
    import threading
    import time
    
    class MainFrame(wx.Frame):
    
        def __init__(self, parent):
            wx.Frame.__init__(self, parent, title='CallAfter example')
    
            panel = wx.Panel(self)
            self.label = wx.StaticText(panel, label="Ready")
            self.btn = wx.Button(panel, label="Start")
            self.gauge = wx.Gauge(panel)
    
            sizer = wx.BoxSizer(wx.VERTICAL)
            sizer.Add(self.label, proportion=1, flag=wx.EXPAND)
            sizer.Add(self.btn, proportion=0, flag=wx.EXPAND)
            sizer.Add(self.gauge, proportion=0, flag=wx.EXPAND)
    
            panel.SetSizerAndFit(sizer)
            self.Bind(wx.EVT_BUTTON, self.OnButton)
    
        def OnButton(self, event):
            """ This event handler starts the separate thread. """
            self.btn.Enable(False)
            self.gauge.SetValue(0)
            self.label.SetLabel("Running")
    
            thread = threading.Thread(target=self.LongRunning)
            thread.start()
    
        def OnLongRunDone(self):
            self.gauge.SetValue(100)
            self.label.SetLabel("Done")
            self.btn.Enable(True)
    
        def LongRunning(self):
            """This runs in a different thread.  Sleep is used to
             simulate a long running task."""
            time.sleep(3)
            wx.CallAfter(self.gauge.SetValue, 20)
            time.sleep(5)
            wx.CallAfter(self.gauge.SetValue, 70)
            time.sleep(4)
            wx.CallAfter(self.OnLongRunDone)
    
    if __name__ == "__main__":
        app = wx.App(0)
        frame = MainFrame(None)
        frame.Show()
        app.MainLoop()