Executing Code Every Certain Time

In this post we will see how to schedule a function to run in the background every certain period of time. In other languages this is usually implemented using a Timer class. The Python standard library ships with the threading.Timer class, but it only allows you to execute code once, after a certain time has elapsed.

import threading
def f():
    print("Hello world!")
# Run the function after 3 seconds.
t = threading.Timer(3, f)
t.start()
print("This runs before the f() function.")

The threading.Timer class moves the function execution into a child or secondary thread, thus t.start() does not block, and the final print() is executed before the f() function runs.

Following this same principle, we can use the threading.Thread class to move a function into a child thread, where a code execution is repeated every certain time, with the help of the timer.sleep() (docs) standard function.

import threading
import time
# Task to be executed at regular intervals.
def timer():
    while True:
        print("Hello world!")
        time.sleep(3)   # 3 seconds.
# Start execution in the background.
t = threading.Thread(target=timer)
t.start()

The logic is simple: inside the function we use an infinite loop, but on each iteration we pause (sleep) the execution for 3 seconds (or however long you want). The pause does not affect the main program execution because the function runs in a secondary thread.

But what happens if we want to stop the timer? Python threads do not include a method to stop their execution. Then we need a way to tell timer() that it should stop running the loop so the function can finish. A basic solution for communicating threads is the threading.Event class, which is roughly a safe way to share a boolean between two threads. Thus, it is possible to create a boolean that indicates whether the timer should be executed or not, and that can be read and modified both from the main thread and from within the timer() function, which runs in a child thread.

import threading
import time
def timer(timer_runs):
    # (4) The code runs while the boolean is true.
    while timer_runs.is_set():
        print("Hello world!")
        time.sleep(3)   # 3 seconds.
# (1) Create the boolean which indicates whether the secondary
# thread runs or not.
timer_runs = threading.Event()
# (2) Initialize it as true.
timer_runs.set()
# (3) Pass it as an argument so it can be read by the timer.
t = threading.Thread(target=timer, args=(timer_runs,))
t.start()

At some other point in the code, when you want to stop the timer execution, just call the clear() method to change the value of the boolean to False.

# Stop the timer execution.
timer_runs.clear()

For example, the following code launches the timer and waits 10 seconds before stopping it.

import threading
import time
def timer(timer_runs):
    while timer_runs.is_set():
        print("Hello world!")
        time.sleep(3)   # 3 seconds.
timer_runs = threading.Event()
timer_runs.set()
t = threading.Thread(target=timer, args=(timer_runs,))
t.start()
# Wait 10 seconds and then stop the timer.
time.sleep(10)
timer_runs.clear()
print("The timer has been stopped!")

And here is the output:

Hello world!
Hello world!
Hello world!
Hello world!
The timer has been stopped!

This logic can be abstracted into a parent Timer class that takes care of all the looping and pausing details, so you can then create multiple timers with less code and a more elegant API:

import threading
import time
class Timer(threading.Thread):
    def __init__(self):
        self._timer_runs = threading.Event()
        self._timer_runs.set()
        super().__init__()
    def run(self):
        while self._timer_runs.is_set():
            self.timer()
            time.sleep(self.__class__.interval)
    def stop(self):
        self._timer_runs.clear()

Now defining a timer is nothing more than this:

class HelloWorldTimer(Timer):
    interval = 3   # Interval in seconds.
    # Function to be executed.
    def timer(self):
        print("Hello world!")

And to start and stop it:

hello_world_timer = HelloWorldTimer()
hello_world_timer.start()
time.sleep(10)
hello_world_timer.stop()