Event Sources API

These are the built-in event sources.

class ml.event_sources.Timer(identifier: str, interval_seconds: float, duration_seconds: float | None = None, on_full: OnFullBehavior = OnFullBehavior.FAIL)

Bases: EventSource

An EventSource that emits a time event at a fixed interval.

This timer is designed to be robust. The behavior when the simulation cannot keep up with the timer’s rate is controlled by the on_full parameter.

  • OnFullBehavior.FAIL (default): A full queue will raise an exception, which is caught by the base EventSource run loop. This stops the timer and logs a clear error, which is the recommended mode for hard real-time simulations where every tick is critical.

  • OnFullBehavior.DROP: A full queue will cause the event to be dropped. The timer will continue running. This is useful for soft real-time simulations where occasional missed ticks are acceptable.

  • OnFullBehavior.OVERWRITE: A full queue will cause an existing event in the queue to be replaced with the new one. This is useful when only the latest value is important.

get_current_time() → float

Gets the most recent time value generated by the timer. This method is thread-safe.

Returns:

The most recent time value emitted by the timer.

run_loop()

A high-precision, drift-free timer loop.

It calculates the sleep time based on the wall clock to prevent drift accumulation over long simulations.