event_loop.EventLoop

dareplane_utils.general.event_loop.EventLoop(dt_s, stop_event=None, ctx=None)

A class that implements a custom event loop with precise timing.

The EventLoop uses dareplane_utils.general.time.sleep_s for more precise sleep timing at the expense of CPU usage.

Callbacks are the means of interacting with the event loop. There are two types of callbacks:

  • Periodic callbacks: These are executed at regular intervals.

  • One-time callbacks: These are executed once and then removed from the list of callbacks. One-time callback can furthermore be scheduled to run at a specific time in the future.

Callbacks can be any callable function, which usually gets one argument ctx - a context object, that can be of type any. This ensures that any type of input can be implemented. See the example section for more details.

Attributes

Name Type Description
dt_s float The time interval in seconds between callback executions.
stop_event threading.Event An event that, when set, signals the event loop to stop.
last_run_ts float The timestamp of the last callback execution.
callbacks list[Callable] A list of periodic callback functions.
callbacks_once list[Callable] A list of one-time callback functions.
delayed_callbacks_once list[DelayedCallback] A list of delayed one-time callback functions.
ctx Any A context object that is passed to the callback functions.
now float The current timestamp as of time.perf_counter(). Used for time keeping internally.

Examples

>>> def no_arg_callback():
...     print("Running with no args")
>>>
>>> evloop = EventLoop(dt_s=0.1)  # process callbacks every 100ms
>>> evloop.add_callback_once(lambda ctx: no_arg_callback())
>>>
>>> # just to stop the loop after 1s
>>> evloop.add_delayed_callback_once(cb=lambda ctx: evloop.stop_event.set(), dt=1)
>>>
>>> evloop.run()
Running with no args
>>>
>>> # NOTE: This example is left to explain how wrapping works. Creating a wrapper is no longer necessary
>>> #       if no `ctx` is an arg / kwargs, the event_loop.validate_callback() will automatically create
>>> #       a wrapper version for you.
>>> def custom_arg_and_kwarg_callback(a, b=1, c=2):
...     print(f"Running with {a=}, {b=}, {c=}")
>>>
>>> def wrapped_custom_arg_and_kwargs(ctx: dict):
...     custom_arg_and_kwarg_callback(ctx["a"], b=ctx["b"], c=ctx["c"])
>>>
>>> evloop = EventLoop(dt_s=0.1, ctx={"a": 11, "b": 22, "c": 33})
>>> evloop.add_callback_once(wrapped_custom_arg_and_kwargs)
>>>
>>> # just to stop the loop after 1s
>>> evloop.add_delayed_callback_once(cb=lambda ctx: evloop.stop_event.set(), dt=1)
>>>
>>> evloop.run()
Running with a=11, b=22, c=33

Methods

Name Description
add_callback Add a periodic callback to the event loop.
add_callback_once Add a one-time callback to the event loop.
add_callbacks A convenience function to add multiple periodic callbacks to the event loop.
add_callbacks_once Convenience function to add multiple one-time callbacks to the event loop.
add_delayed_callback_once Add a one-time callback to the event loop that is evaluated after a delay.
run Run the event loop, evaluating the callback every self.dt_s seconds
validate_callback Check that every callback accepts at least a kwarg with ‘ctx’.

add_callback

dareplane_utils.general.event_loop.EventLoop.add_callback(cb)

Add a periodic callback to the event loop.

Parameters

Name Type Description Default
cb Callable The callback function to be added. required

add_callback_once

dareplane_utils.general.event_loop.EventLoop.add_callback_once(cb)

Add a one-time callback to the event loop.

Parameters

Name Type Description Default
cb Callable The callback function to be added. required

add_callbacks

dareplane_utils.general.event_loop.EventLoop.add_callbacks(cbs)

A convenience function to add multiple periodic callbacks to the event loop.

Parameters

Name Type Description Default
cbs list[Callable] A list of callback functions to be added. required

add_callbacks_once

dareplane_utils.general.event_loop.EventLoop.add_callbacks_once(cbs)

Convenience function to add multiple one-time callbacks to the event loop.

Parameters

Name Type Description Default
cbs list[Callable] A list of callback functions to be added. required

add_delayed_callback_once

dareplane_utils.general.event_loop.EventLoop.add_delayed_callback_once(
    cb,
    dt=0.0,
)

Add a one-time callback to the event loop that is evaluated after a delay.

Parameters

Name Type Description Default
cb Callable The callback function to be added. required
dt float The delay in seconds before the callback is executed. Defaults to 0.0. 0.0

run

dareplane_utils.general.event_loop.EventLoop.run()

Run the event loop, evaluating the callback every self.dt_s seconds

validate_callback

dareplane_utils.general.event_loop.EventLoop.validate_callback(cb)

Check that every callback accepts at least a kwarg with ‘ctx’. More kwargs are possible.