event_scheduler¶

event_scheduler.event_scheduler¶

class event_scheduler.event_scheduler.Event(time, priority, action, argument, kwargs, id)¶: Bases: event_scheduler.event_scheduler.Event

class event_scheduler.event_scheduler.EventScheduler(thread_name=None, timefunc=<built-in function monotonic>, timer_class=<class 'threading.Timer'>)¶

Bases: object

The Event Scheduler is an always-on scheduler which is able to accept and run events based on its scheduled time and priority. Inspiration for the API was taken from python’s sched module (https://docs.python.org/3/library/sched.html).

cancel(event: event_scheduler.event_scheduler.Event) → int¶

Remove an event from the queue using the id returned by enter()/enterabs(). If the event is not in the queue, this is a no-op.

Parameters: event – The event to be cancelled.
Returns: 0 if the event was successfully removed/not in the queue, -1 otherwise.
Return type: int

cancel_all() → int¶

Clear all events from the queue. If the queue is already empty, this is a no-op.

Returns: 0 if all the events were successfully cleared, -1 otherwise.
Return type: int

cancel_recurring(event_id) → int¶

Remove recurring event from the queue using the id returned by enter_recurring(). If the recurring event is not in the queue, this is a no-op.

Parameters: event_id (int) – The id of the recurring event to be cancelled.
Returns: 0 if the event was successfully removed/not in the queue, -1 otherwise.
Return type: int

enter(delay, priority, action, arguments=(), kwargs=<object object>) → event_scheduler.event_scheduler.Event ¶

Enter a new event in the queue to occur at a time relative to the current time.

Parameters

delay – The relative time the event will be scheduled to execute. Eg. If you pass in 1 as the delay, the event will be scheduled to execute in 1 + now() seconds from now.
priority (int) – The priority the event will execute with. If two events are scheduled for the same time, the event with the lower priority will execute first.
action (callable) – The function which will invoked when the event executes.
arguments (optional) – Variable length argument list for the action.
kwargs (dict, optional) – Keyword arguments for the action.

Returns

The scheduled event if the scheduler is running, None otherwise. This can be used to cancel the event later, if necessary.

Return type

Event

Raises

ValueError – If the 0 > priority >= sys.maxsize

Warning

Long running actions will stall the internal thread and may impact the scheduling of other events.

enter_recurring(interval, priority, action, arguments=(), kwargs=<object object>) → int¶

Enter a new recurring event in the queue to occur at a specified interval.

Parameters

interval – The interval time the event will be scheduled to execute. Eg. If you pass in 5 as the delay, the event will be scheduled to execute every 5 seconds starting five seconds from when it’s entered.
priority (int) – The priority the event will execute with. If two events are scheduled for the same time, the event with the lower priority will execute first.
action (callable) – The function which will invoked when the event executes.
arguments (optional) – Variable length argument list for the action.
kwargs (dict, optional) – Keyword arguments for the action.

Returns

An event id of the recurring event if the scheduler is running, None otherwise. This id can be used to cancel the event later, if necessary.

Return type

int

Raises

ValueError – If the 0 > priority >= sys.maxsize

Warning

Long running actions will stall the internal thread and may impact the scheduling of other events.

enterabs(time, priority, action, arguments=(), kwargs=<object object>) → event_scheduler.event_scheduler.Event ¶

Enter a new event in the queue to occur at an absolute time.

Parameters

time – The absolute time the event will be scheduled to execute.
priority (int) – The priority the event will execute with. If two events are scheduled for the same time, the event with the lower priority will execute first.
action (callable) – The function which will invoked when the event executes.
arguments (optional) – Variable length argument list for the action.
kwargs (dict, optional) – Keyword arguments for the action.

Returns

The scheduled event if the scheduler is running, None otherwise. This can be used to cancel the event later, if necessary.

Return type

Event

Raises

ValueError – If the 0 > priority >= sys.maxsize

Warning

Long running actions will stall the internal thread and may impact the scheduling of other events.

property queue¶

Return an ordered list of upcoming events. Events are named tuples with fields for: time, priority, action, arguments, kwargs, id

Returns: All the events currently in the queue ordered from the soonest to occur and by priority,
Return type: list

start() → int¶

Start the event scheduler and enable it to start taking events.

Returns: 0 if the event scheduler was successfully started, -1 if the scheduler has already been started or is in the process of stopping.
Return type: int

stop(hard_stop: bool = False) → int¶

Stop the event scheduler and stop its internal thread. Will not be able to take in new events when invoked.

Parameters: hard_stop (bool, optional) – If set to False, wait until all events execute at their scheduled time before stopping. If set to True, will stop the scheduler right away and discard all pending events.
Returns: 0 if the event scheduler was successfully stopped, -1 if the scheduler is already in the process of stopping/already stopped.
Return type: int

class event_scheduler.event_scheduler.SchedulerStatus(value)¶

Bases: enum.Enum

An enumeration.

RUNNING = 0¶

STOPPED = 2¶

STOPPING = 1¶

event_scheduler.test_util¶

class event_scheduler.test_util.TestTimer(interval, function, args=None, kwargs=None)¶

Bases: threading.Timer

The TestTimer is designed to test your code that incorporates the event scheduler. Passing the TestTimer.monotonic and TestTimer into the event scheduler gives you full control of the event scheduler’s monotonic time. Passing the event scheduler’s condition variable enables full synchronization with the event scheduler’s internal thread.

classmethod advance_time(increment: float) → None¶

Advance the class-owned monotonic time. :param increment: The value to advance the monotonic time.

Raises: ValueError – Will be raised if given a negative increment value.

cancel()¶: Stop the timer if it hasn’t finished yet.

classmethod monotonic() → float¶

Get the class-owned monotonic time.

Returns: The current monotonic time represented by this class.
Return type: float

classmethod reset()¶: Reset the internal state of the TestTimer class. This will reset the monotonic time to 0 and remove the set event scheduler.

classmethod set_event_scheduler(event_scheduler: event_scheduler.event_scheduler.EventScheduler)¶

Set the event scheduler of the application being tested.

Parameters: event_scheduler – The application’s event scheduler.

start()¶: Start the timer.