55 lines
1.6 KiB
Python
55 lines
1.6 KiB
Python
|
import threading
|
||
|
|
import time
|
||
|
|
|
||
|
|
# Threaded repeating timer thing
|
||
|
|
# From https://stackoverflow.com/a/40965385/16432246
|
||
|
|
# It shouldn't drift, but that's untested
|
||
|
|
# Keep in mind you can't catch Exceptions for this, it just crashes the thread.
|
||
|
|
# Some more scheduling stuff: https://www.redwood.com/article/python-job-scheduling/
|
||
|
|
class RepeatedTimer(object):
|
||
|
|
'''
|
||
|
|
Run stuff repeatedly every x seconds
|
||
|
|
Example usage (from SO and ported to Python 3):
|
||
|
|
from time import sleep
|
||
|
|
|
||
|
|
def hello(name):
|
||
|
|
print "Hello %s!" % name
|
||
|
|
|
||
|
|
print()"starting...")
|
||
|
|
rt = RepeatedTimer(1, hello, "World") # it auto-starts, no need of rt.start()
|
||
|
|
try:
|
||
|
|
sleep(5) # your long-running job goes here...
|
||
|
|
catch:
|
||
|
|
rt.stop() # better in a try/catch block to make sure the program ends!
|
||
|
|
'''
|
||
|
|
|
||
|
|
def __init__(self, interval, function, *args, **kwargs):
|
||
|
|
'''
|
||
|
|
Run a functions with arguments every
|
||
|
|
'''
|
||
|
|
self._timer = None
|
||
|
|
self.interval = interval
|
||
|
|
self.function = function
|
||
|
|
self.args = args
|
||
|
|
self.kwargs = kwargs
|
||
|
|
self.is_running = False
|
||
|
|
self.next_call = time.time()
|
||
|
|
self.start()
|
||
|
|
|
||
|
|
def _run(self):
|
||
|
|
self.is_running = False
|
||
|
|
self.start()
|
||
|
|
self.function(*self.args, **self.kwargs)
|
||
|
|
|
||
|
|
def start(self):
|
||
|
|
if not self.is_running:
|
||
|
|
self.next_call += self.interval
|
||
|
|
self._timer = threading.Timer(
|
||
|
|
self.next_call - time.time(), self._run)
|
||
|
|
self._timer.start()
|
||
|
|
self.is_running = True
|
||
|
|
|
||
|
|
def stop(self):
|
||
|
|
self._timer.cancel()
|
||
|
|
self.is_running = False
|