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):[CODE=python]#-----------------------------------------------------------------------------
# Name: sched_thread_te st.py
# Purpose: Test sched.Scheduler in a thread
#
# Author: Barton
#
# Created: 2007/11/18
# RCS-ID: $Id: sched_thread_te st.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.ente r(interval, 1, self.SimEvent, args)
self.running = False
self.runTread = threading.Threa d(target=self.R un)

def SimEvent(self, *args):
"""Reschedu le this handler and call the action fuction"""
startTime = time.time()
self.actionFuct (args)
interval = self.interval - (time.time() - startTime)
if self.running:
self.timer.ente r(interval, 1, self.SimEvent, args)

def Run(self):
self.running = True
self.timer.run( )

def Start(self):
self.runTread.s tart()

def Stop(self):
self.running = False

def BG_Task(thrdEve nt):
def periodicTask(*a rgs):
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.Threa d(target=BG_Tas k, args=(threading Event,))
print 'starting thread'
thrd.start()
time.sleep(2)
print 'stopping thread'
threadingEvent. set()
thrd.join()
[/CODE]

[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:[CODE=python]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.ente r(interval, 1, self.SimEvent, args)
self.thrdEvent = threading.Event ()
self.runTread = threading.Threa d(target=self.R un, args=(self.thrd Event, self.timer))

def SimEvent(self, *args):
"""Reschedu le 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.ente r(interval, 1, self.SimEvent, args)

def Run(self, thrdEvent, timer):
thrdEvent.set()
timer.run()

def Start(self):
self.runTread.s tart()

def Stop(self):
self.thrdEvent. clear()
[/CODE]

[bartonc]

Here's the cleaned up version:[CODE=python]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.ente r(interval, 1, self.SimEvent, args)
self.thrdEvent = threading.Event ()
self.runTread = threading.Threa d(target=self.R un, args=(self.thrd Event, self.timer))

def SimEvent(self, *args):
"""Reschedu le 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.ente r(interval, 1, self.SimEvent, args)

def Run(self, thrdEvent, timer):
thrdEvent.set()
timer.run()

def Start(self):
self.runTread.s tart()

def Stop(self):
self.thrdEvent. clear()
[/CODE]

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:[CODE=python]#-----------------------------------------------------------------------------
# Name: SchedThreadTest Frame.py
# Purpose: Test sched.Scheduler in a thread in wx
#
# Author: <your name>
#
# Created: 2007/11/18
# RCS-ID: $Id: SchedThreadTest Frame.py $
# Copyright: (c) 2007
# Licence: <your licence>
#-----------------------------------------------------------------------------
#Boa:Frame:Fram e1

import wx

import threading
import time

from sched_thread_te st import BG_Timer

def create(parent):
return Frame1(parent)

[wxID_FRAME1, wxID_FRAME1BUTT ON1, wxID_FRAME1PANE L1,
] = [wx.NewId() for _init_ctrls in range(3)]

class Frame1(wx.Frame ):
def _init_ctrls(sel f, prnt):
# generated method, don't edit
wx.Frame.__init __(self, id=wxID_FRAME1, name='', parent=prnt, pos=wx.Point(13 2, 132),
size=wx.Size(40 0, 250), style=wx.DEFAUL T_FRAME_STYLE,
title='Sched Thread Test Frame')
self.SetClientS ize(wx.Size(392 , 223))
self.Bind(wx.EV T_CLOSE, self.OnFrame1Cl ose)

self.panel1 = wx.Panel(id=wxI D_FRAME1PANEL1, name='panel1', parent=self, pos=wx.Point(0, 0),
size=wx.Size(39 2, 223), style=wx.TAB_TR AVERSAL)

self.button1 = wx.Button(id=wx ID_FRAME1BUTTON 1, label='End Task', name='button1',
parent=self.pan el1, pos=wx.Point(10 4, 72), size=wx.Size(14 4, 24), style=0)
self.button1.Bi nd(wx.EVT_BUTTO N, self.OnButton1, id=wxID_FRAME1B UTTON1)

def __init__(self, parent):
self._init_ctrl s(parent)

self.threadingE vent = threading.Event ()
self.thrd = threading.Threa d(target=self.B G_Task, args=(self.thre adingEvent,))
print 'starting thread'
self.thrd.start ()


def BG_Task(self, thrdEvent):
def periodicTask(*a rgs):
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.threadingE vent.set()
self.thrd.join( )

def OnFrame1Close(s elf, event):
self.OnButton1( None)
event.Skip()
[/CODE]

[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:[CODE=python]#-----------------------------------------------------------------------------
# Name: wxSchedThreadTe st.py
# Purpose: The wxApp
#
# Author: <your name>
#
# Created: 2007/11/19
# RCS-ID: $Id: wxSchedThreadTe st.py $
# Copyright: (c) 2007
# Licence: <your licence>
#-----------------------------------------------------------------------------
#!/usr/bin/env python
#Boa:App:BoaApp

import wx

import SchedThreadTest Frame

modules ={u'SchedThread TestFrame': [1,
'Main frame of Application',
u'SchedThreadTe stFrame.py']}

class BoaApp(wx.App):
def OnInit(self):
self.main = SchedThreadTest Frame.create(No ne)
self.main.Show( )
self.SetTopWind ow(self.main)
return True

def main():
application = BoaApp(redirect =1)
application.Mai nLoop()

if __name__ == '__main__':
main()
[/CODE]

[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:[CODE=python]

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.s et()

def run(self):
self.finished.w ait(self.interv al)
if not self.finished.i sSet():
self.function(* self.args, **self.kwargs)
self.finished.s et()[/CODE]

[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():[CODE=python]#-----------------------------------------------------------------------------
# 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.exitfunc tion mechanism,
## # so it may be posible to do too much here.
## self.Stop()

def __init__(self, actionFunct, *args):
# The function to perform
self.actionFunc t = 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.runEnableF lag = 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=s elf.Run, args=(self.runE nableFlag, 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.runEnableF lag.clear()
# record the ammount of time taken by
startTime = time()
# the task that gets called
self.actionFunc t(*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.runEnableF lag.isSet():
self.event = self.timer.ente r(interval, 1, self.SimEvent, ())

def Run(self, runEnableFlag, timer):
"""The target of the internal thread for starting the scheduler."""
runEnableFlag.s et()
timer.run()
# if sched.scheduler .run() returns, clear the flag
runEnableFlag.c lear()

def Start(self, milliseconds, oneShot=False):
"""Manage resetting of the interval by cancel()ing the sched.event"""
if milliseconds <= 0:
raise ValueError, "millisecon ds must be greater that zero."
# interval is scaled to milliseconds
self.interval = float(milliseco nds)/1000
self.oneShot = oneShot
# Are we being re-Start()ed? #
if (not self.event is None) and (not self.timer.empt y()):
# Yes, so remove this event, if it has not completed
self.timer.canc el(self.event)
# enter a 1st priority task into the scheduler's queue
self.event = self.timer.ente r(self.interval , 1, self.SimEvent, ())
if not self.runEnableF lag.isSet():
self.runThread. start()

def Stop(self):
if (self.event is not None) and (not self.timer.empt y()):
# remove this event, if it has not completed
self.timer.canc el(self.event)
# and let the start threat manage the flag
else:
self.runEnableF lag.clear()

def IsRunning(self) :
return self.runThread. isAlive()

def GetInterval(sel f):
"""Rescale and cast the interval to int before returning."""
return int(self.interv al * 1000)

def ThreadSafeTest( ):
# test thread safety #
def BG_Task(thrdEve nt):
def periodicTask(ar g1, arg2):
print arg1, arg2,
print time()

t = Timer(periodicT ask, '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=B G_Task, args=(threading Event,))
print 'starting thread'
thrd.start()
sleep(2)
print 'stopping thread'
threadingEvent. set()
thrd.join()

if __name__ == "__main__":
def periodicTask(ar g1, arg2):
print arg1, arg2,
print time()

t = Timer(periodicT ask, '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[/CODE]

[highvelcty]

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