CODEWITHSUNDEEP
pythonmultithreadingwxpython
ryantmer
ryantmer

Reputation: 673

wxpython - Running threads sequentially without blocking GUI

I've got a GUI script with all my wxPython code in it, and a separate testSequences module that has a bunch of tasks that I run based on input from the GUI. The tasks take a long time to complete (from 20 seconds to 3 minutes), so I want to thread them, otherwise the GUI locks up while they're running. I also need them to run one after another, since they all use the same hardware. (My rationale behind threading is simply to prevent the GUI from locking up.) I'd like to have a "Running" message (with varying number of periods after it, i.e. "Running", "Running.", "Running..", etc.) so the user knows that progress is occurring, even though it isn't visible. I'd like this script to run the test sequences in separate threads, but sequentially, so that the second thread won't be created and run until the first is complete. Since this is kind of the opposite of the purpose of threads, I can't really find any information on how to do this... Any help would be greatly appreciated.

Thanks in advance!

gui.py

import testSequences
from threading import Thread

#wxPython code for setting everything up here...
for j in range(5):
    testThread = Thread(target=testSequences.test1)
    testThread.start()
    while testThread.isAlive():
        #wait until the previous thread is complete
        time.sleep(0.5)
        i = (i+1) % 4
        self.status.SetStatusText("Running"+'.'*i)

testSequences.py

import time
def test1():
    for i in range(10):
        print i
        time.sleep(1)

(Obviously this isn't the actual test code, but the idea is the same.)

Upvotes: 1

Views: 3809

Answers (2)

ryantmer
ryantmer

Reputation: 673

Figured out a way to do this. Instead of creating threads in my gui.py, I created a class that inherits from Thread, and runs all the tests in that class, then posts wxPython events when one test is done (so I can update the status bar) and when all tests are done (so I can inform the user that all tests are complete.

myEVT_TESTDONE = wx.NewEventType()
EVT_TESTDONE = wx.PyEventBinder(myEVT_TESTDONE , 1)
myEVT_ALLDONE = wx.NewEventType()
EVT_ALLDONE = wx.PyEventBinder(myEVT_ALLDONE, 1)

class TestDone(wx.PyCommandEvent):
    def __init__(self, etype, eid, val=None):
        wx.PyCommandEvent.__init__(self, etype, eid)
        self._val = val
    def GetValue(self):
        return self._val

class AllDone(wx.PyCommandEvent):
    def __init__(self, etype, eid):
        wx.PyCommandEvent.__init__(self, etype, eid)

class TestSequence(Thread):
    def __init__(self, parent, queue):
        Thread.__init__(self)
        self._queue = queue
        self._parent = parent
        self.start()
    def run(self):
        testCount = 0
        for test in self._queue:
            #Time-intensive task goes here
            for i in range(10):
                print i
                sleep(1)
            evt = TestDone(myEVT_TESTDONE, -1, i)
            wx.PostEvent(self._parent, evt)
        evt = AllDone(myEVT_ALLDONE, -1)
        wx.PostEvent(self._parent, evt)

class MainSequence(wx.Frame):
    def __init__(self, parent, id, title):
        self.Bind(EVT_TESTDONE, self.testDoneEvt)
        self.Bind(EVT_ALLDONE, self.allDoneEvt)
        #...the rest of the wxPython code
    def testDoneEvt(self, event):
        #Set what to be done after every test, e.g. update progress bar
        step = event.GetValue()
    def allDoneEvt(self, event):
        #Set what to be done after all tests, e.g. display "Tests complete"

program = wx.App()
window = MainSequence(None, -1, 'App title')
program.MainLoop()

Upvotes: 2

ErwinP
ErwinP

Reputation: 412

You cannot wait in the GUI-thread with a while loop because you block the processing of the event-queue. One solution is to poll the state of the thread with a timer:

import wx 
import time
from threading import Thread

def test1():
    for i in range(10):
        print i
        time.sleep(1)

class TestFrame(wx.Frame): 
    def __init__(self): 
        wx.Frame.__init__(self, None, -1, "Test") 
        panel = wx.Panel(self, -1) 
        sizer = wx.BoxSizer(wx.VERTICAL) 
        panel.SetSizer(sizer) 

        self.button = wx.Button(panel, 0, "Start")
        sizer.Add(self.button, 0, wx.ALIGN_LEFT) 
        self.button.Bind(wx.EVT_BUTTON, self.OnButton)

        self.text = wx.StaticText(panel, 0, "No test is running")
        sizer.Add(self.text, 0, wx.ALIGN_LEFT) 

        self.timer = wx.Timer(self)

    def OnButton(self, event):
        self.testThread = Thread(target=test1)
        self.testThread.start()
        self.text.SetLabel("Running")
        self.button.Disable()
        self.Bind(wx.EVT_TIMER, self.PollThread)
        self.timer.Start(20, oneShot=True)
        event.Skip()

    def PollThread(self, event):
        if self.testThread.isAlive():
            self.Bind(wx.EVT_TIMER, self.PollThread)
            self.timer.Start(200, oneShot=True)
            self.text.SetLabel(self.text.GetLabel() + ".")
        else:
            self.button.Enable()
            self.text.SetLabel("Test completed")


app = wx.PySimpleApp() 
TestFrame().Show() 
app.MainLoop()

Upvotes: 2

Related Questions