Created
May 10, 2014 17:56
-
-
Save liftoff/fae06d7ace9f11325743 to your computer and use it in GitHub Desktop.
A simple periodic callback mechanism that uses the stdlib's threading.Timer
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
""" | |
A periodic callback mechanism that uses `threading.Timer`. | |
""" | |
class PeriodicTimerCallback(object): | |
""" | |
A wrapper that uses either `threading.Timer` to call functions at a | |
specified interval. Example:: | |
>>> from datetime import datetime | |
>>> def print_date(): print(datetime.now()) | |
>>> pc = PeriodicTimerCallback(print_date, 1000) | |
>>> pc.start() | |
2014-05-10 13:54:58.318233 | |
2014-05-10 13:54:59.319531 | |
2014-05-10 13:55:00.320827 | |
2014-05-10 13:55:01.321875 | |
>>> pc.stop() | |
""" | |
def __init__(self, callback, callback_time): | |
self.callback = callback | |
self.callback_time = callback_time | |
from threading import Timer | |
# NOTE: Timer uses seconds | |
def callback_wrapper(): | |
"Runs the callback and restarts the Timer so it will run again" | |
self.callback() | |
self._pc = Timer(callback_time / 1000, callback_wrapper) | |
if self._running: | |
self._pc.start() | |
self._pc = Timer(callback_time / 1000, callback_wrapper) | |
self._running = False | |
def start(self): | |
"""Starts the timer.""" | |
self._running = True | |
self._pc.start() | |
def stop(self): | |
"""Stops the timer.""" | |
self._running = False | |
self._pc.cancel() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment