A sched.scheduler wrapper to run continuously as a "service", to queue up one-time operations.
A PeriodicOperation based on native threading.Timer, runs as a "service".
Both can be stopped.
Timer could have been used in lieu of the scheduler, but the latter stands to have more precision.
| import sched | |
| import time | |
| import threading | |
| class ThreadedScheduler(sched.sheduler, threading.Thread): | |
| def __init__(self): | |
| threading.Thread.__init__(self, daemon=True) | |
| sched.scheduler.__init__(self) | |
| self._running = False | |
| def cancel(self): | |
| self._running = False | |
| def run(self): | |
| # implement Thread.run() specifically | |
| # this means ThreadedScheduler is NOT a drop-in for sched.scheduler | |
| while self._running: | |
| time.sleep(0.1) | |
| sched.scheduler.run(self) | |
| class PeriodicOperation(threading.Thread): | |
| def __init__(self, period_sec:int, operation:Callable): | |
| threading.Thread.__init__(self, daemon=True) | |
| self._op = operation | |
| self._period = period_sec | |
| self._running = False | |
| def cancel(self): | |
| self._running = False | |
| def run(self): | |
| self._running = True | |
| while self._running: | |
| # Every <self._period> seconds, kick off the op in the background | |
| # and immediately wait another round again without waiting for the | |
| # op to complete (it is on its merry way) | |
| opt = threading.Thread(target=self._op, daemon=True) | |
| t = threading.Timer(self._period, opt.start) | |
| t.start() | |
| t.join() |