bobocep.cep.gen.event.BoboGenEventTime

class bobocep.cep.gen.event.BoboGenEventTime(millis: int, datagen: Callable | None = None, gen_timestamp: BoboGenTimestamp | None = None, from_now: bool = True, tz=None)[source]

Bases: BoboGenEvent

An event generator that returns a BoboEventSimple time event if a given amount of time has elapsed. If the time has not elapsed, None is returned instead.

__init__(millis: int, datagen: Callable | None = None, gen_timestamp: BoboGenTimestamp | None = None, from_now: bool = True, tz=None)[source]
Parameters:

  • millis – Milliseconds between event generations. Once the millisecond timer has elapsed, an event can be generated.

  • datagen – Datagen to use in generated event.

  • gen_timestamp – Custom timestamp generator. If None, a BoboGenTimestampEpoch instance is used to generate timestamps.

  • from_now – If True, sets the millisecond timer to the current time; otherwise, the timer is set to 0.

  • tz – Timezone data.

maybe_generate(event_id: str) → BoboEvent | None[source]
Parameters:

event_id – An event ID.

Returns:

A BoboEventSimple instance if one is due to be generated, or None if not.