remotivelabs.topology.time.threaded_ticker

class ThreadedTicker:

ThreadedTicker provides similar functionality to asyncio.Task for timer-based operations. It calls the on_tick callback at regular intervals. The biggest difference is that ThreadedTicker schedules the callback to run on the event loop, but have no further control over the callback execution. Hence, it cannot compensate for long callback execution times. The upside is that it is guaranteed to run at the specified interval due to the timer being a thread-based solution. However, as mentioned before, if the callback takes longer than the interval, the number of scheduled callbacks will accumulate over time.

Note that the timer is a best effort timer. The actual interval may be longer than the specified interval due to thread scheduling and callback scheduling time.

ThreadedTicker( interval_in_sec: float, on_tick: remotivelabs.topology.time.callback.OnTickCallback, loop: asyncio.events.AbstractEventLoop)
running: bool
def start(self) -> None:
def cancel(self) -> None:
def create_ticker( interval_in_sec: float, on_tick: remotivelabs.topology.time.callback.OnTickCallback, loop: asyncio.events.AbstractEventLoop | None = None) -> ThreadedTicker:

Creates a ticker that calls the on_tick function at the given interval.

Note that the timer is a best effort timer. The actual interval may be longer than the specified interval due to thread scheduling and callback execution time.

Arguments:
  • interval_in_sec: The time interval in seconds between each tick.
  • on_tick: Callback coroutine that will be called on each tick. Make sure to handle exceptions in the callback, as they will not be caught by the ticker.
  • loop: The event loop to schedule callbacks on. If None, uses the current loop.