import sched
import time
from typing import Any, Callable
from moderngl_window.timers.base import BaseTimer
[docs]
class Scheduler:
def __init__(self, timer: BaseTimer):
"""Create a Scheduler object to handle events.
Args:
timer (BaseTimer): timer to use, subclass of BaseTimer.
Raises:
ValueError: timer is not a valid argument.
"""
if not isinstance(timer, BaseTimer):
raise ValueError(
"timer, {}, has to be a instance of BaseTimer or a callable!".format(timer)
)
self._events: dict[int, sched.Event] = dict()
self._event_id = 0
self._scheduler = sched.scheduler(lambda: timer.time, time.sleep)
[docs]
def run_once(
self,
action: Callable[[Any], Any],
delay: float,
*,
priority: int = 1,
arguments: tuple[Any, ...] = (),
kwargs: dict[Any, Any] = dict(),
) -> int:
"""Schedule a function for execution after a delay.
Args:
action (callable):
function to be called.
delay (float):
delay in seconds.
priority (int, optional):
priority for this event, lower is more important. Defaults to 1.
arguments (tuple, optional):
arguments for the action. Defaults to ().
kwargs (dict, optional):
keyword arguments for the action. Defaults to dict().
Returns:
int: event id that can be canceled.
"""
event = self._scheduler.enter(delay, priority, action, arguments, kwargs)
self._events[self._event_id] = event
self._event_id += 1
return self._event_id - 1
[docs]
def run_at(
self,
action: Callable[[Any], Any],
time: float,
*,
priority: int = 1,
arguments: tuple[Any, ...] = (),
kwargs: dict[Any, Any] = dict(),
) -> int:
"""Schedule a function to be executed at a certain time.
Args:
action (callable):
function to be called.
time (float):
epoch time at which the function should be called.
priority (int, optional):
priority for this event, lower is more important. Defaults to 1.
arguments (tuple, optional):
arguments for the action. Defaults to ().
kwargs (dict, optional):
keyword arguments for the action. Defaults to dict().
Returns:
int: event id that can be canceled.
"""
event = self._scheduler.enterabs(time, priority, action, arguments, kwargs)
self._events[self._event_id] = event
self._event_id += 1
return self._event_id - 1
[docs]
def run_every(
self,
action: Callable[[Any], Any],
delay: float,
*,
priority: int = 1,
initial_delay: float = 0.0,
arguments: tuple[Any, ...] = (),
kwargs: dict[Any, Any] = dict(),
) -> int:
"""Schedule a recurring function to be called every `delay` seconds after a initial delay.
Args:
action (callable):
function to be called.
delay (float):
delay in seconds.
priority (int, optional):
priority for this event, lower is more important. Defaults to 1.
initial_delay (float, optional):
initial delay in seconds before executing for the first time.
Defaults to 0. arguments (tuple, optional):
arguments for the action. Defaults to ().
kwargs (dict, optional):
keyword arguments for the action. Defaults to dict().
Returns:
int: event id that can be canceled.
"""
recurring_event = self._recurring_event_factory(
action, arguments, kwargs, (delay, priority), self._event_id
)
event = self._scheduler.enter(initial_delay, priority, recurring_event)
self._events[self._event_id] = event
self._event_id += 1
return self._event_id - 1
def _recurring_event_factory(
self,
function: Callable[[Any], Any],
arguments: tuple[Any, ...],
kwargs: dict[Any, Any],
scheduling_info: tuple[Any, Any],
id: int,
) -> Callable[[], None]:
"""Factory for creating recurring events that will reschedule themselves.
Args:
function (callable): function to be called.
arguments (tuple): arguments for the function.
kwargs (dict): keyword arguments for the function.
scheduling_info (tuple): tuple of information for scheduling the task.
id (int): event id this event should be assigned to.
"""
def _f() -> None:
function(*arguments, **kwargs)
event = self._scheduler.enter(*scheduling_info, _f)
self._events[id] = event
return _f
[docs]
def execute(self) -> None:
"""Run the scheduler without blocking and execute any expired events."""
self._scheduler.run(blocking=False)
[docs]
def cancel(self, event_id: int, delay: float = 0) -> None:
"""Cancel a previously scheduled event.
Args:
event_id (int): event to be canceled
delay (float, optional): delay before canceling the event. Defaults to 0.
"""
if delay == 0:
self._cancel(event_id)
else:
self.run_once(self._cancel, delay, priority=0, arguments=(event_id,))
def _cancel(self, event_id: int) -> None:
if event_id not in self._events:
raise ValueError("Recurring event with id {} does not exist".format(event_id))
event = self._events.pop(event_id)
self._scheduler.cancel(event)