wxPython - problem w/ timers

[jergosh]

I get the following error in my app:

wx._core.PyAssertionError: C++ assertion "wxThread::IsMain()" failed at ..\..\sr
c\common\timercmn.cpp(66) in wxTimerBase::Start(): timer can only be started fro
m the main thread

The origin of the error is obvious enough but I can't think of any reason I'm not allowed to start timers from threads other than main if I so wish. Is there any going around this limitation?

TIA,
Greg

[bartonc]

The only reason that I can come up with is that the docs say:

Note: A timer can only be used from the main thread.

[jergosh]

The only reason that I can come up with is that the docs say:

Apparently there's no going around that, the timers just won't start. However, I thought about using standard Python threading timers, I wonder if they're safe to use with wxPython?

[bartonc]

Apparently there's no going around that, the timers just won't start. However, I thought about using standard Python threading timers, I wonder if they're safe to use with wxPython?

Ya know; I've been working on a project that needed a timer in a module that lacked a wx event handler. threading.Timer is one option (definitely thread safe), but you'll need to be careful as to how you implement the recurrence and the ability to stop it. I also looked into sched.scheduler which has nicely documented error handling. I'll whip up a threaded test to see how that might work.

[bartonc]

Ya know; I've been working on a project that needed a timer in a module that lacked a wx event handler. threading.Timer is one option (definitely thread safe), but you'll need to be careful as to how you implement the recurrence and the ability to stop it. I also looked into sched.scheduler which has nicely documented error handling. I'll whip up a threaded test to see how that might work.

It looks thread-safe to me (actually, forgive the slopy use threading in the BG_Timer class - I got in a rush there):

#-----------------------------------------------------------------------------
# Name:        sched_thread_test.py
# Purpose:     Test sched.Scheduler in a thread
#
# Author:      Barton
#
# Created:     2007/11/18
# RCS-ID:      $Id: sched_thread_test.py $
# Copyright:   (c) 2007
# Licence:     Free
#-----------------------------------------------------------------------------
import sched, time
import threading
 
class BG_Timer(object):
    def __init__(self, interval, actionFuct, *args):
        self.interval = interval
        self.actionFuct = actionFuct
        self.timer = sched.scheduler(time.time, time.sleep)
        self.timer.enter(interval, 1, self.SimEvent, args)
        self.running = False
        self.runTread = threading.Thread(target=self.Run)
 
    def SimEvent(self, *args):
        """Reschedule this handler and call the action fuction"""
        startTime = time.time()
        self.actionFuct(args)
        interval = self.interval - (time.time() - startTime)
        if self.running:
            self.timer.enter(interval, 1, self.SimEvent, args)
 
    def Run(self):
        self.running = True
        self.timer.run()
 
    def Start(self):
        self.runTread.start()
 
    def Stop(self):
        self.running = False
 
def BG_Task(thrdEvent):
    def periodicTask(*args):
        print time.time()
 
    t = BG_Timer(.25, periodicTask)
    print 'starting bg'
    t.Start()
    print 'running bg'
    thrdEvent.wait()
    print 'stopping bg'
    t.Stop()
 
 
threadingEvent = threading.Event()
thrd = threading.Thread(target=BG_Task, args=(threadingEvent,))
print 'starting thread'
thrd.start()
time.sleep(2)
print 'stopping thread'
threadingEvent.set()
thrd.join()
 

[bartonc]

It looks thread-safe to me (actually, forgive the slopy use threading in the BG_Timer class - I got in a rush there):

Here's the cleaned up version:

class BG_Timer(object):
    def __init__(self, interval, actionFuct, *args):
        self.interval = interval
        self.actionFuct = actionFuct
        self.timer = sched.scheduler(time.time, time.sleep)
        self.timer.enter(interval, 1, self.SimEvent, args)
        self.thrdEvent = threading.Event()
        self.runTread = threading.Thread(target=self.Run, args=(self.thrdEvent, self.timer))
 
    def SimEvent(self, *args):
        """Reschedule this handler and call the action fuction"""
        startTime = time.time()
        self.actionFuct(args)
        interval = self.interval - (time.time() - startTime)
        if self.thrdEvent.isSet():
            self.timer.enter(interval, 1, self.SimEvent, args)
 
    def Run(self, thrdEvent, timer):
        thrdEvent.set()
        timer.run()
 
    def Start(self):
        self.runTread.start()
 
    def Stop(self):
        self.thrdEvent.clear()
 

[bartonc]

Here's the cleaned up version:

class BG_Timer(object):
    def __init__(self, interval, actionFuct, *args):
        self.interval = interval
        self.actionFuct = actionFuct
        self.timer = sched.scheduler(time.time, time.sleep)
        self.timer.enter(interval, 1, self.SimEvent, args)
        self.thrdEvent = threading.Event()
        self.runTread = threading.Thread(target=self.Run, args=(self.thrdEvent, self.timer))
 
    def SimEvent(self, *args):
        """Reschedule this handler and call the action fuction"""
        startTime = time.time()
        self.actionFuct(args)
        interval = self.interval - (time.time() - startTime)
        if self.thrdEvent.isSet():
            self.timer.enter(interval, 1, self.SimEvent, args)
 
    def Run(self, thrdEvent, timer):
        thrdEvent.set()
        timer.run()
 
    def Start(self):
        self.runTread.start()
 
    def Stop(self):
        self.thrdEvent.clear()
 

While I was out cleaning the barn, I realized that I hadn't addressed the core question of "safe with wx". So I put the normal if __name__ == "__main__": guard into the original module. And as expected, everything worked great:

#-----------------------------------------------------------------------------
# Name:        SchedThreadTestFrame.py
# Purpose:     Test sched.Scheduler in a thread in wx
#
# Author:      <your name>
#
# Created:     2007/11/18
# RCS-ID:      $Id: SchedThreadTestFrame.py $
# Copyright:   (c) 2007
# Licence:     <your licence>
#-----------------------------------------------------------------------------
#Boa:Frame:Frame1
 
import wx
 
import threading
import time
 
from sched_thread_test import BG_Timer
 
def create(parent):
    return Frame1(parent)
 
[wxID_FRAME1, wxID_FRAME1BUTTON1, wxID_FRAME1PANEL1, 
] = [wx.NewId() for _init_ctrls in range(3)]
 
class Frame1(wx.Frame):
    def _init_ctrls(self, prnt):
        # generated method, don't edit
        wx.Frame.__init__(self, id=wxID_FRAME1, name='', parent=prnt, pos=wx.Point(132, 132),
                size=wx.Size(400, 250), style=wx.DEFAULT_FRAME_STYLE,
                title='Sched Thread Test Frame')
        self.SetClientSize(wx.Size(392, 223))
        self.Bind(wx.EVT_CLOSE, self.OnFrame1Close)
 
        self.panel1 = wx.Panel(id=wxID_FRAME1PANEL1, name='panel1', parent=self, pos=wx.Point(0, 0),
                size=wx.Size(392, 223), style=wx.TAB_TRAVERSAL)
 
        self.button1 = wx.Button(id=wxID_FRAME1BUTTON1, label='End Task', name='button1',
                parent=self.panel1, pos=wx.Point(104, 72), size=wx.Size(144, 24), style=0)
        self.button1.Bind(wx.EVT_BUTTON, self.OnButton1, id=wxID_FRAME1BUTTON1)
 
    def __init__(self, parent):
        self._init_ctrls(parent)
 
        self.threadingEvent = threading.Event()
        self.thrd = threading.Thread(target=self.BG_Task, args=(self.threadingEvent,))
        print 'starting thread'
        self.thrd.start()
 
 
    def BG_Task(self, thrdEvent):
        def periodicTask(*args):
            print time.time()
 
        t = BG_Timer(.25, periodicTask)
        print 'starting bg'
        t.Start()
        print 'running bg'
        thrdEvent.wait()
        print 'stopping bg'
        t.Stop()
 
    def OnButton1(self, event):
        print 'stopping thread'
        self.threadingEvent.set()
        self.thrd.join()
 
    def OnFrame1Close(self, event):
        self.OnButton1(None)
        event.Skip()
 

[bartonc]

While I was out cleaning the barn, I realized that I hadn't addressed the core question of "safe with wx". So I put the normal if __name__ == "__main__": guard into the original module. And as expected, everything worked great:

And here's a neat trick that redirects stdout to a wxWindow. It comes in very handy for these types of text based tests:

#-----------------------------------------------------------------------------
# Name:        wxSchedThreadTest.py
# Purpose:     The wxApp
#
# Author:      <your name>
#
# Created:     2007/11/19
# RCS-ID:      $Id: wxSchedThreadTest.py $
# Copyright:   (c) 2007
# Licence:     <your licence>
#-----------------------------------------------------------------------------
#!/usr/bin/env python
#Boa:App:BoaApp
 
import wx
 
import SchedThreadTestFrame
 
modules ={u'SchedThreadTestFrame': [1,
                            'Main frame of Application',
                            u'SchedThreadTestFrame.py']}
 
class BoaApp(wx.App):
    def OnInit(self):
        self.main = SchedThreadTestFrame.create(None)
        self.main.Show()
        self.SetTopWindow(self.main)
        return True
 
def main():
    application = BoaApp(redirect=1)
    application.MainLoop()
 
if __name__ == '__main__':
    main()
 

[jergosh]

Wow... that's a lot of work you put into it, thanks very much! I've resolved most of my issues with concurrency now (also thanks to discovering functions like wx.Yield(), wx.FutureCall() etc ;)

Greg

[bartonc]

Wow... that's a lot of work you put into it, thanks very much! I've resolved most of my issues with concurrency now (also thanks to discovering functions like wx.Yield(), wx.FutureCall() etc ;)

Greg

wx.FutureCall() comes in very handy and I use it quite often to do initialization after the window has been created (and shown).

wx.Yield() has (for some reason) been declared deprecated:

::wxYield
bool wxYield()

Calls wxApp::Yield.

This function is kept only for backwards compatibility. Please use the wxApp::Yield method instead in any new code.

But Pythoneers rarely keep a reference to the app object, so you may want to keep an eye out for compatibility issues like this.

[bartonc]

Wow... that's a lot of work you put into it, thanks very much! I've resolved most of my issues with concurrency now (also thanks to discovering functions like wx.Yield(), wx.FutureCall() etc ;)

Greg

You are quite welcome. Actually, it's no trouble at all (given that my IDE - Boa Constructor - did most of the writing).

Since you were able to discover wx.FutureCall(), then you must have the wxPython in Action book. I haven't been able to find documentation for that function anywhere else. If you have a different resource, I'd really like to know about it.

Thanks.

[bartonc]

You are quite welcome. Actually, it's no trouble at all (given that my IDE - Boa Constructor - did most of the writing).

Since you were able to discover wx.FutureCall(), then you must have the wxPython in Action book. I haven't been able to find documentation for that function anywhere else. If you have a different resource, I'd really like to know about it.

Thanks.

This is ripped right out of the threading module in the Python lib directory:

 
def Timer(*args, **kwargs):
    return _Timer(*args, **kwargs)
 
class _Timer(Thread):
    """Call a function after a specified number of seconds:
 
    t = Timer(30.0, f, args=[], kwargs={})
    t.start()
    t.cancel() # stop the timer's action if it's still waiting
    """
 
    def __init__(self, interval, function, args=[], kwargs={}):
        Thread.__init__(self)
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.finished = Event()
 
    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished.set()
 
    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.isSet():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

[bartonc]

Here's the version after I developed are real need (running wxPython in a thread).
I'm considering adding wxEvents to this soon, but for now it's a close enough approximation to drop in as a replacement for wxTimer():

#-----------------------------------------------------------------------------
# Name:        Timer.py
# Purpose:     A replacement for wxTimer that will run in a thread
#
# Author:      Barton Cline
#
# Created:     2007/11/30
# RCS-ID:      $Id: Timer.py $
# Copyright:   (c) 2007
# Licence:     <your licence>
#-----------------------------------------------------------------------------
 
from threading import Thread, Event
import sched
from time import time, sleep
 
class Timer(object):
 
##    def __del__(self):
##        """Useless! GC won't call here while the thread is running."""
##        # So find a way to detect the main thread reaching its termination #
##        # the threading module is hooked into the python.exitfunction mechanism,
##        # so it may be posible to do too much here.
##        self.Stop()
 
    def __init__(self, actionFunct, *args):
        # The function to perform
        self.actionFunct = actionFunct
        # its arguments
        self.args = args
 
        # Handles thread termination correctly #
        self.oneShot = False
 
        # The scheduler:
        self.timer = sched.scheduler(time, sleep)
        self.event = None  #store the result of sched.scheduler.enter()
        # a thread-safe way to signal the scheduler to stop
        self.runEnableFlag = Event() # created in the clear()ed state
 
        # Need a thread to activate the scheduler because sched.scheduler.run() won't return
        self.runThread = Thread(target=self.Run, args=(self.runEnableFlag, self.timer))
 
    def SimEvent(self):
        """Call the function then reschedule this handler. The scheduler will block
        further attempts at calling the actionFunct if an error occurres there."""
        if self.oneShot:
            self.runEnableFlag.clear()
        # record the ammount of time taken by
        startTime = time()
        # the task that gets called
        self.actionFunct(*self.args)
        # scaled to milliseconds
        timeSpent = (time() - startTime)/1000
        if timeSpent > self.interval:
            # just go again after the given interval
            interval = self.interval
        else:
            # adjust for a precise rep-rate
            interval = self.interval - timeSpent
        if self.runEnableFlag.isSet():
            self.event = self.timer.enter(interval, 1, self.SimEvent, ())
 
    def Run(self, runEnableFlag, timer):
        """The target of the internal thread for starting the scheduler."""
        runEnableFlag.set()
        timer.run()
        # if sched.scheduler.run() returns, clear the flag
        runEnableFlag.clear()
 
    def Start(self, milliseconds, oneShot=False):
        """Manage resetting of the interval by cancel()ing the sched.event"""
        if milliseconds <= 0:
            raise ValueError, "milliseconds must be greater that zero."
        # interval is scaled to milliseconds
        self.interval = float(milliseconds)/1000
        self.oneShot = oneShot
        # Are we being re-Start()ed? #
        if (not self.event is None) and (not self.timer.empty()):
            # Yes, so remove this event, if it has not completed
            self.timer.cancel(self.event)
        # enter a 1st priority task into the scheduler's queue
        self.event = self.timer.enter(self.interval, 1, self.SimEvent, ())
        if not self.runEnableFlag.isSet():
            self.runThread.start()
 
    def Stop(self):
        if (self.event is not None) and (not self.timer.empty()):
            # remove this event, if it has not completed
            self.timer.cancel(self.event)
            # and let the start threat manage the flag
        else:
            self.runEnableFlag.clear()
 
    def IsRunning(self):
        return self.runThread.isAlive()
 
    def GetInterval(self):
        """Rescale and cast the interval to int before returning."""
        return int(self.interval * 1000)
 
def ThreadSafeTest():
    # test thread safety #
    def BG_Task(thrdEvent):
        def periodicTask(arg1, arg2):
            print arg1, arg2,
            print time()
 
        t = Timer(periodicTask, 'hello', 'world')
        print 'starting bg at ', time()
        t.Start(250)
        print 'running bg'
        thrdEvent.wait()
        print 'stopping bg'
        t.Stop()
 
    threadingEvent = Event()
    thrd = Thread(target=BG_Task, args=(threadingEvent,))
    print 'starting thread'
    thrd.start()
    sleep(2)
    print 'stopping thread'
    threadingEvent.set()
    thrd.join()
 
if __name__ == "__main__":
    def periodicTask(arg1, arg2):
        print arg1, arg2,
        print time()
 
    t = Timer(periodicTask, 'hello', 'world')
    print 'starting bg at ', time()
    t.Start(250)
    print 'running bg at 250, but interupting and resetting to 125'
    sleep(.76)
    t.Start(125)
    print 'running bg at 125'
    sleep(.5)
    # passed
    print 'running one-shot'
    t.Start(125, True)
##    # passed
##    print 'stopping bg'
##    t.Stop()
##    # XXX FAILED!
##    t = None # will it be garbage collected?
##    # XXX FAILED!
##    del t

[highvelcty]

Thank you! I needed a way to generate timer driven events in a wxPython sub-thread and this works great!