Periodic Timers

Xronos offers several ways to implement timed behavior, and the easiest place to start is with periodic timers. To follow along, copy the example below into a new file called timer.py.

plainwith type hintsdiagram

import datetime

import xronos

MAX_COUNT = 5


class Timed(xronos.Reactor):
    _timer = xronos.PeriodicTimerDeclaration()

    def __init__(self, period):
        super().__init__()
        self._timer.period = period
        self._count = 0

    @xronos.reaction
    def hello(self, interface):
        interface.add_trigger(self.startup)
        return lambda: print(f"{self.get_time()}: {self.name} says 'Hello!'")

    @xronos.reaction
    def on_timer(self, interface):
        interface.add_trigger(self._timer)
        shutdown_effect = interface.add_effect(self.shutdown)

        def handler():
            self._count += 1
            print(
                f"{self.get_time()}: {self.name}'s timer triggered "
                f"(count={self._count})"
            )
            if self._count == MAX_COUNT:
                shutdown_effect.trigger_shutdown()

        return handler

    @xronos.reaction
    def goodbye(self, interface):
        interface.add_trigger(self.shutdown)
        return lambda: print(f"{self.get_time()}: {self.name} says 'Goodbye!'")


def main():
    env = xronos.Environment()
    env.create_reactor("timed", Timed, period=datetime.timedelta(seconds=1))
    env.execute()


if __name__ == "__main__":
    main()

To create a periodic timer, instantiate xronos.PeriodicTimerDeclaration as a class attribute. This declares a PeriodicTimer object that the Xronos runtime initializes for you automatically. You can then interact with the timer object using self._timer.

The __init__ method of Timed configures the timer’s period from an argument it receives, and also initializes the state variable self._count to 0.

The Timed reactor defines three reactions. The hello reaction will feel familiar from the “Hello, World!” example — it prints the current time, the reactor’s name, and ‘Hello!’. The goodbye reaction produces similar output in response to the builtin shutdown event.

The on_timer reaction declares the timer as its trigger, and also defines an effect — specifically a ShutdownEffect that can be used to stop the program. Its handler is a little more involved than the simple lambda we’ve seen so far: it defines a nested handler function and returns it.

def handler():
    self._count += 1
    print(
        f"{self.get_time()}: {self.name}'s timer triggered "
        f"(count={self._count})"
    )
    if self._count == MAX_COUNT:
        shutdown_effect.trigger_shutdown()

return handler

The handler increments self._count, prints the current time, the timer’s name, and the current count, and then checks whether the count has reached MAX_COUNT. When it has, trigger_shutdown() is called to stop the program.

The program runs for about 4 seconds and produces output similar to the following:

$ python timer.py
2024-11-29 10:17:13.893071: timed says 'Hello!'
2024-11-29 10:17:13.893071: timed's timer triggered (count=1)
2024-11-29 10:17:14.893071: timed's timer triggered (count=2)
2024-11-29 10:17:15.893071: timed's timer triggered (count=3)
2024-11-29 10:17:16.893071: timed's timer triggered (count=4)
2024-11-29 10:17:17.893071: timed's timer triggered (count=5)
2024-11-29 10:17:17.893071: timed says 'Goodbye!'

Take a close look at the printed timestamps. The timestamp from hello matches the first line printed by on_timer. That’s because we didn’t configure an offset, so the first timer event fires immediately at startup — simultaneous with the startup event. You’ll also notice that the individual timer events are spaced exactly one second apart.

How does Xronos achieve that kind of precision? Time in the Xronos SDK works differently from the wall-clock time most programs use. The runtime manages its own internal clock, and that clock is central to how program execution is controlled. One key property: the time a reaction observes is always equal to the timestamp of the event that triggered it. As a result, any two calls to get_time() within the same reaction handler will return the same value.

This synchrony of timestamps carries across multiple reactors as well. Try modifying the instantiation code as shown below, so the program runs three Timed reactors with periods of 1 second, 2 seconds, and 3 seconds.

plainwith type hintsdiagram

def main():
    env = xronos.Environment()
    env.create_reactor("timed1", Timed, period=datetime.timedelta(seconds=1))
    env.create_reactor("timed2", Timed, period=datetime.timedelta(seconds=2))
    env.create_reactor("timed3", Timed, period=datetime.timedelta(seconds=3))
    env.execute()

Running the modified program produces output similar to the following:

$ python timer.py
2024-11-29 10:48:57.423695: timed2 says 'Hello!'
2024-11-29 10:48:57.423695: timed3 says 'Hello!'
2024-11-29 10:48:57.423695: timed1 says 'Hello!'
2024-11-29 10:48:57.423695: timed2's timer triggered (count=1)
2024-11-29 10:48:57.423695: timed3's timer triggered (count=1)
2024-11-29 10:48:57.423695: timed1's timer triggered (count=1)
2024-11-29 10:48:58.423695: timed1's timer triggered (count=2)
2024-11-29 10:48:59.423695: timed2's timer triggered (count=2)
2024-11-29 10:48:59.423695: timed1's timer triggered (count=3)
2024-11-29 10:49:00.423695: timed3's timer triggered (count=2)
2024-11-29 10:49:00.423695: timed1's timer triggered (count=4)
2024-11-29 10:49:01.423695: timed2's timer triggered (count=3)
2024-11-29 10:49:01.423695: timed1's timer triggered (count=5)
2024-11-29 10:49:01.423695: timed2 says 'Goodbye!'
2024-11-29 10:49:01.423695: timed3 says 'Goodbye!'
2024-11-29 10:49:01.423695: timed1 says 'Goodbye!'

Notice how all three reactors say ‘Hello!’ or ‘Goodbye!’ at the same timestamp, and how the timestamps of their individual timer events align with each other.