NotQuiteParadise2

Scheduler

class ScheduledItem(func: Callable, last_ts: float, next_ts: float, interval: float, repeat: bool)[source]

Bases: object

Describes a scheduled callback.

__init__(func: Callable, last_ts: float, next_ts: float, interval: float, repeat: bool) None
func: Callable
interval: float
last_ts: float
next_ts: float
repeat: bool
class Scheduler(time_function: typing.Callable = <built-in function perf_counter>)[source]

Bases: object

Class for scheduling functions.

Probably not thread safe.

The function should have a prototype that includes dt as the first argument, which gives the elapsed time, in time units, since the last clock tick.

def callback(dt):

pass

Items will be scheduled and rescheduled in such a way to prevent several items from running during the same tick. Enabling a “soft” reschedule means that item is tolerant to having the interval slightly modified.

Initialise a Scheduler, with optional custom time function.

Parameters

time_function – Return the elapsed time

__init__(time_function: typing.Callable = <built-in function perf_counter>)[source]

Initialise a Scheduler, with optional custom time function.

Parameters

time_function – Return the elapsed time

call_scheduled_functions(dt: float)[source]

Call scheduled functions that elapsed on the last update_time.

Parameters
  • dt – The elapsed time since the last update to pass to each

  • function. (scheduled) –

get_counter() float[source]

Get internal counter value

get_idle_time() Optional[float][source]

Get the time until the next item is scheduled.

Returns

Time until the next scheduled event in time units, or None if there is no events scheduled.

get_interval() float[source]

Get the average amount of time passed between each tick.

Useful for calculating FPS if this clock is used with the display. Returned value is averaged from last 10 ticks.

Value will be 0.0 if before 1st tick.

Returns

Average amount of time passed between each tick

get_running_time() float[source]

Get time clock has been running

get_schedule() List[nqp.core.scheduler.ScheduledItem][source]

Return copy of the schedule.

schedule(func: Callable, delay: float, interval: float, repeat: bool, soft: bool)[source]
schedule_interval(func: Callable, interval: float = 0.0, delay: float = 0.0, soft: bool = False)[source]

Schedule a function to run on an interval.

NOTE! If delay==0.0, then the interval will start the next frame, meaning, the callback will happen the next tick, and then follow the interval. If you want to avoid this, then set delay to the same value as the interval.

Items are rescheduled after they are executed. That means that by default, the interval of items may not be consistent with the initial time with was scheduled.

If the interval is not specified, the function will be executed every time the scheduler is ticked.

Parameters
  • func – Function to be called

  • interval – Repeat on this interval

  • delay – Delay in time unit until it is called for first time

  • soft – See notes about Soft Scheduling

Returns

Reference to scheduled item

schedule_once(func: Callable, delay: float = 0.0, soft: bool = False) nqp.core.scheduler.ScheduledItem[source]

Schedule a function to be run once sometime in the future.

If the delay is not specified, the function will be executed during the next tick.

Parameters
  • func – Function to be called

  • delay – Delay in time unit until it is called

  • soft – See notes about Soft Scheduling

Returns

Reference to scheduled item.

set_time(time_stamp: float) float[source]

Set the clock manually and do not call scheduled functions.

Parameters

time_stamp – This will become the new value of the clock.

Returns

The number of time units since the last update, or 0.0 if this was the first update.

tick() float[source]

Cause clock to update and call scheduled functions.

This updates the clock’s internal measure of time and returns the difference since the last update (or since the clock was created).

Will call any scheduled functions that have elapsed.

Returns

The number of time units since the last “tick”, or 0 if this was the first tick.

unschedule(func) None[source]

Remove a function from the schedule.

NOTE: do not unschedule own function during function call

If the function appears in the schedule more than once, all occurrences are removed. If the function was not scheduled, no error is raised.

Parameters

func – The function to remove from the schedule.

remove(items, func)[source]

Remove scheduled items with func func from list remove_from and return True if it was removed