Implementing my own event loop in a wxPython application

Question

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?

Answer 1

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

Answer 2

Just post wx.Events to some object (typically the application or the main window). They will be processed in FIFO order, although they will be intermixed with the other GUI events happening on the main thread itself. Of course, you also need to have actual handlers for these events implementing whatever logic you need.

Answer 3

For posterity, here is the decorator I created using Yoriz’s answer.

def run_on_main_thread(fn):
    """Decorator. Forces the function to run on the main thread.

    Any calls to a function that is wrapped in this decorator will
    return immediately; the return value of such a function is not
    available.

    """
    @wraps(fn)
    def deferred_caller(*args, **kwargs):
        # If the application has been quit then trying to use
        # CallAfter will trigger an assertion error via the line
        #     assert app is not None, 'No wx.App created yet'
        # Since assertions are optimized out when -O is used,
        # though, it seems safest to perform the check ourselves,
        # and also catch the exception just in case.
        if wx.GetApp() is not None:
            try:
                wx.CallAfter(fn, *args, **kwargs)
            except AssertionError:
                pass

    return deferred_caller