Created
March 10, 2016 19:15
-
-
Save pbaja/cf92d090d026810feefd to your computer and use it in GitHub Desktop.
This file contains 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
import time, threading | |
#This class meant to be really simple, just delaying tasks | |
#But it got some nice features :) | |
''' | |
This example will repeat your function every 1 second, and stop doing that after 10 seconds: | |
from delayer import Delay | |
def my_func(): | |
print("Hello") | |
d = Delay(target=my_func,delay=1,repeat=True) | |
def stop_that(): | |
d.cancel() | |
Delay(target=stop_that,delay=10,daemon=False) | |
''' | |
class Delay(object): | |
''' | |
Parameters: | |
target - Function that will be called after delay has passed | |
args - Arguments for target (if any) | |
delay - Time to wait in seconds before calling target | |
start - If true, counting will start instantly, otherwise you must call .start() | |
daemon - If true, main program will not wait for thread to end before exiting | |
repeat - If true, delay will repeat forever. Only way to stop it is to call .cancel() | |
instant - If true, target will be called once before counting | |
''' | |
def __init__(self, target=None, args=None, delay=0, start=True, daemon=True, repeat=False, instant=False): | |
self.target = target | |
self.delay = delay | |
self.args = args | |
self.canceled = False | |
self.repeat = repeat | |
self.thread = threading.Thread(target=self.__waiting_thread) | |
self.thread.daemon = daemon | |
if start: | |
self.start() | |
if instant: | |
self.__run() | |
def __waiting_thread(self): | |
timer = time.time()+self.delay | |
while not self.canceled: | |
if time.time()>=timer: | |
self.__run() | |
if self.repeat: | |
timer = time.time()+self.delay | |
else: | |
self.canceled = True | |
else: | |
time.sleep(1) | |
def __run(self): | |
if self.args != None: | |
self.target(self.args) | |
else: | |
self.target() | |
def start(self): | |
self.thread.start() | |
def cancel(self): | |
self.canceled = True |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment