Hi,
I recently needed to run a function every half a second and the first option
was threading.Timer, but it just tick once.
Based on https://stackoverflow.com/a/48741004 I derived a class from Timer to
run endlessly, but there are two things I do not like:
- The function, args and kwargs attributes of Timer are just the same as the
_target, _args and _kwargs from Thread.
- It feels better that Timer can "tick" a given number of times.
So I changed Timer to be more inline with the Thread class and make it possible
to run the function more than once:
class Timer(Thread):
FOREVER = Ellipsis
def __init__(self, interval, function, args=None, kwargs=None,
name=None, *, ticks=1, daemon=None):
# Keep compatibility with the old Timer, where the default
# args and kwargs where mutable.
if args is None:
args = []
if kwargs is None:
kwargs = {}
Thread.__init__(self, target=function,
args=args, kwargs=kwargs,
name=name, daemon=daemon)
self.interval = interval
self.ticks = ticks
self.finished = Event()
def cancel(self):
"""Stop the timer if it hasn't finished yet."""
self.finished.set()
def run(self):
try:
while not self.finished.wait(self.interval):
if self._target and not self.finished.is_set():
self._target(*self._args, **self._kwargs)
if self.ticks is self.FOREVER:
continue
self.ticks -= 1
if self.ticks <= 0:
break
finally:
# Remove the reference to the function for the same
# reason it is done in Thread.run
del self._target
# However, for backwards compatibility do not remove
# _args and _kwargs as is done in Thread.run
# del self._args, self._kwargs
self.finished.set()
def join(self, timeout=None):
# If the timer was set to repeat endlessly
# it will never join if it is not cancelled before.
if self.ticks is self.FOREVER:
self.cancel()
super().join(timeout=timeout)
@property
def function(self):
return getattr(self, '_target', None)
@function.setter
def function(self, value):
self._target = value
@property
def args(self):
return self._args
@args.setter
def args(self, value):
self._args = value
@property
def kwargs(self):
return self._kwargs
@kwargs.setter
def kwargs(self, value):
self._kwargs = value
The default option for ticks is to run once, just like the current Timer.
With this version it is possible to do things like:
>>> t = Timer(10.0, f, ticks=3)
>>> t.start()
# Will run f three times, with 10 seconds between each call.
>>> t = Timer(10.0, f, ticks=Timer.FOREVER, daemon=True)
>>> t.start()
# Will run f every 10 seconds until the end of the program
# It also stops the thread if the program is interrupted.
>>> t = Timer(10.0, f, ticks=..., daemon=True)
>>> t.start()
# Same as above, but I like the usage of Ellipsis here :)
Other option is to leave Timer as it is and create a new class, so it is not
necessary to keep the compatibility with the function, args and kwargs
attributes:
class Ticker(Thread):
FOREVER = Ellipsis
def __init__(self, interval, target=None, args=(), kwargs=None,
group=None, name=None, *, ticks=1, daemon=None):
super().__init__(group=group, target=target,
args=args, kwargs=kwargs,
name=name, daemon=daemon)
self.interval = interval
self.ticks = ticks
self.finished = Event()
def cancel(self):
"""Stop the timer if it hasn't finished yet."""
self.finished.set()
def evaluate(self):
if self._target and not self.finished.is_set():
self._target(*self._args, **self._kwargs)
def run(self):
try:
while not self.finished.wait(self.interval):
self.evaluate()
if self.ticks is self.FOREVER:
continue
self.ticks -= 1
if self.ticks <= 0:
break
finally:
# Remove the reference to the function for the same
# reason it is done in Thread.run
del self._target, self._args, self._kwargs
self.finished.set()
def join(self, timeout=None):
# If the timer was set to repeat endlessly
# it will never join if it is not cancelled before.
if self.ticks is self.FOREVER:
self.cancel()
super().join(timeout=timeout)
Do you think it is useful to try to clean up and add this code to threading or
it is just something too specific to consider changing the standard library?
Regards,
Andrés
_______________________________________________
Python-ideas mailing list -- [email protected]
To unsubscribe send an email to [email protected]
https://mail.python.org/mailman3/lists/python-ideas.python.org/
Message archived at
https://mail.python.org/archives/list/[email protected]/message/33JL6L64UKHLO3ZRIHEQOWRCEFMAL5O7/
Code of Conduct: http://python.org/psf/codeofconduct/
