Event scheduler library already exists and can be "tweaked" to become periodic. - https://docs.python.org/3/library/sched.html
Best Regards, Ronie On Sun, Jun 9, 2019, 2:35 AM <killer...@gmail.com> wrote: > 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 -- python-ideas@python.org > To unsubscribe send an email to python-ideas-le...@python.org > https://mail.python.org/mailman3/lists/python-ideas.python.org/ > Message archived at > https://mail.python.org/archives/list/python-ideas@python.org/message/33JL6L64UKHLO3ZRIHEQOWRCEFMAL5O7/ > Code of Conduct: http://python.org/psf/codeofconduct/ >
_______________________________________________ Python-ideas mailing list -- python-ideas@python.org To unsubscribe send an email to python-ideas-le...@python.org https://mail.python.org/mailman3/lists/python-ideas.python.org/ Message archived at https://mail.python.org/archives/list/python-ideas@python.org/message/I5GEL4PH4YNFTRM2NI4SJHICLCNSN773/ Code of Conduct: http://python.org/psf/codeofconduct/
