Skip to content
Advertisement

Start a Thread Timer directly

I want to trigger the self._runSequence() func directly after start the timer. This is my code:

self.sequenceTimer = RepeatedTimer( sequenceDuration, self.__runSequence, sequence)

If sequenceDuration is equal to 10 the func will be called 10 seconds later (as a normal timer I know..)

This is the RepeatedTimer Class:

from threading import Timer
    
class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
    self.interval   = interval
    self.function   = function
    self.args       = args
    self.kwargs     = kwargs
    self.is_running = False
    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._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

What I already try:

self.sequenceTimer = RepeatedTimer( sequenceDuration, self.__runSequence(sequence), sequence)

But this give me an error when the timer trigger the __runSequence func that tell me I don’t give an argument to __runSequence() func.

Advertisement

Answer

If I understood you correctly, you want the function to be executed when the RepeatedTimer is initialized, and then periodically, right?

Instead of calling start in __init__, call _run. It calls start and execute the function.

import time
from threading import Timer
    
class RepeatedTimer(object):
    def __init__(self, interval, function, *args, **kwargs):
        self._timer     = None
        self.interval   = interval
        self.function   = function
        self.args       = args
        self.kwargs     = kwargs
        self.is_running = False
        self._run()

    def _run(self):
        self.is_running = False
        self.start()
        self.function(*self.args, **self.kwargs)

    def start(self):
        if not self.is_running:
            self._timer = Timer(self.interval, self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False


print("Before initialization")
sequence_timer = RepeatedTimer(10, lambda *args, **kwargs: print(time.time(), *args, **kwargs), "foo", "bar", sep=" ; ")
print("After initialization")
time.sleep(50)
sequence_timer.stop()

Output

Before initialization
1622809236.0391436 ; foo ; bar
After initialization
1622809246.0429533 ; foo ; bar
1622809256.055273 ; foo ; bar
1622809266.0606596 ; foo ; bar
1622809276.0665863 ; foo ; bar

You can see that the first execution occurs before the initialization is complete.


Your first attempt was already correct.

self.sequenceTimer = RepeatedTimer(sequenceDuration, self.__runSequence, sequence)

The second would pass the result of self.__runSequence(sequence) (None unless you explicitly return something else) instead of the method itself.

User contributions licensed under: CC BY-SA
9 People found this is helpful
Advertisement