Skip to content

Instantly share code, notes, and snippets.

@alexbw
Created September 1, 2011 20:11
Show Gist options
  • Save alexbw/1187132 to your computer and use it in GitHub Desktop.
Save alexbw/1187132 to your computer and use it in GitHub Desktop.
A repeating timer in Python
class RepeatingTimer(object):
"""
USAGE:
from time import sleep
def myFunction(inputArgument):
print(inputArgument)
r = RepeatingTimer(0.5, myFunction, "hello")
r.start(); sleep(2); r.interval = 0.05; sleep(2); r.stop()
"""
def __init__(self,interval, function, *args, **kwargs):
super(RepeatingTimer, self).__init__()
self.args = args
self.kwargs = kwargs
self.function = function
self.interval = interval
def start(self):
self.callback()
def stop(self):
self.interval = False
def callback(self):
if self.interval:
self.function(*self.args, **self.kwargs)
Timer(self.interval, self.callback, ).start()
@RXProgramming
Copy link

RXProgramming commented Oct 23, 2020

def run(self):
    while not self.finished.is_set():
        self.finished.wait(self.interval)
        self.function(*self.args, **self.kwargs)
    self.finished.set()

One question... What for self.finished.set() needed if program come there only if event already set.
And one more thing guys. I am trying to create timer wich can be paused and started again:

`class RXTimer( Thread ):
def init(self, interval, function, args=None, kwargs=None):
Thread.init(self)
self.interval = interval
self.function = function
self.args = args if args is not None else []
self.kwargs = kwargs if kwargs is not None else {}
self.paused = Event()
self.active = Event()
self.active.clear()
self.paused.set()

def run( self ):
    while True:
        self.active.wait()
        self.paused.clear()
        while not self.paused.set():
            self.paused.wait( self.interval )
            self.function( *self.args, **self.kwargs )
        self.active.clear()

def switch( self ):
    if self.active.is_set():
        self.paused.set()
    else:
        self.active.set()
    
def isActive( self ): return self.active.is_set()            

'`
But problem is self.paused.wait( self.interval ) do not lock thread and thread call self.function( *self.args, **self.kwargs ) without pauses.

Why?

@xuancu
Copy link

xuancu commented Jul 13, 2022

from threading import Timer
import time

class RepeatingTimer(object):
"""
USAGE:
from time import sleep
r = RepeatingTimer(_print, 0.5, "hello")
r.start(); sleep(2); r.interval = 0.05; sleep(2); r.stop()
"""

def __init__(self, function, interval, *args, **kwargs):
    super(RepeatingTimer, self).__init__()
    self.args = args
    self.kwargs = kwargs
    self.function = function
    self.interval = interval

def start(self):
    self.callback()
    
def stop(self):
    self.interval = False
    
def callback(self):
    if self.interval:
        self.function(*self.args, **self.kwargs)
        Timer(self.interval, self.callback, ).start()

def dowork():
global count
count = count + 1

count = 0
timer = RepeatingTimer(dowork, 0.0001)
timer.start()
time.sleep(1.0)
timer.stop()

print(count)
output:
3039

The theoretical output should be 10000. But here it is 3039. I'm not understanding why!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment