Sunday, March 18, 2012

Repeating Timer in Python

While working on Scheduler [1], I've encountered an inconvenience caused by lack of auto-repeat timer in standard Python API.
In this post I will present RepeatTimer class that features:
  • RepeatTimer (as its threading.Timer ancestor) is being executed in a separate thread
  • RepeatTimer executes function on its triggering
  • Interface provides ability to trigger RepeatTimer immediately, and start new countdown
  • Interface provides ability to set new interval that will be applied after upcoming triggering
  • cancel method disengage RepeatTimer and kills its thread
Gist below presents the code:
"""
Created on 2011-02-10
@author: Bohdan Mushkevych
@author: Brian Curtin
http://code.activestate.com/lists/python-ideas/8982/
"""
from datetime import datetime
import threading
class RepeatTimer(threading.Thread):
def __init__(self, interval, callable, args=[], kwargs={}):
threading.Thread.__init__(self)
# interval_current shows number of milliseconds in currently triggered <tick>
self.interval_current = interval
# interval_new shows number of milliseconds for next <tick>
self.interval_new = interval
self.callable = callable
self.args = args
self.kwargs = kwargs
self.event = threading.Event()
self.event.set()
self.activation_dt = None
self.__timer = None
def run(self):
while self.event.is_set():
self.activation_dt = datetime.utcnow()
self.__timer = threading.Timer(self.interval_new,
self.callable,
self.args,
self.kwargs)
self.interval_current = self.interval_new
self.__timer.start()
self.__timer.join()
def cancel(self):
self.event.clear()
if self.__timer is not None:
self.__timer.cancel()
def trigger(self):
self.callable(*self.args, **self.kwargs)
if self.__timer is not None:
self.__timer.cancel()
def change_interval(self, value):
self.interval_new = value
view raw repeat_timer.py hosted with ❤ by GitHub
[1] Scheduler at Github
https://github.com/mushkevych/scheduler

No comments: