Timer thread.
Attributes
| Attribute | Type | Description |
|---|
| Entry | type = Entry | Reference to the Entry class used for scheduling individual timer events. |
| Schedule | type = Schedule | Reference to the Schedule class used to manage the queue of timer entries. |
| running | bool = False | Boolean flag indicating whether the timer thread is currently active and processing events. |
| on_tick | Optional[Callable[[float], None]] = None | Optional callback function executed on every timer tick with the current delay value. |
Constructor
Signature
def Timer(
schedule: Optional[Schedule] = None,
on_error: Optional[Callable[[Exception], None]] = None,
on_tick: Optional[Callable[[float], None]] = None,
on_start: Optional[Callable[['Timer'], None]] = None,
max_interval: Optional[float] = None,
kwargs: Any
) - > None
Parameters
| Name | Type | Description |
|---|
| schedule | Optional[Schedule] = None | An optional Schedule instance to manage timed entries. |
| on_error | Optional[Callable[[Exception], None]] = None | Callback function to execute when an error occurs. |
| on_tick | Optional[Callable[[float], None]] = None | Callback function to execute on every timer tick. |
| on_start | Optional[Callable[['Timer'], None]] = None | Callback function to execute when the timer starts. |
| max_interval | Optional[float] = None | The maximum sleep interval allowed between ticks. |
| kwargs | Any | Additional keyword arguments passed to the threading.Thread superclass. |
Signature
def Timer(
schedule: Optional[Schedule],
on_error: Optional[Callable[[Exception], None]],
on_tick: Optional[Callable[[float], None]],
on_start: Optional[Callable[['Timer'], None]],
max_interval: Optional[float]
) - > null
Parameters
| Name | Type | Description |
|---|
| schedule | Optional[Schedule] | An existing Schedule instance to manage timed events; defaults to a new Schedule if not provided |
| on_error | Optional[Callable[[Exception], None]] | Callback function triggered when an exception occurs during task execution |
| on_tick | Optional[Callable[[float], None]] | Callback function triggered on every timer tick with the current delay value |
| on_start | Optional[Callable[['Timer'], None]] | Callback function triggered immediately before the thread starts |
| max_interval | Optional[float] | The maximum allowed sleep interval between scheduled events |
Methods
start()
@classmethod
def start() - > None
Starts the timer thread and prints a stack trace if TIMER_DEBUG is enabled.
Returns
| Type | Description |
|---|
None | Nothing |
run()
@classmethod
def run() - > None
Executes the main timer loop, processing scheduled entries and sleeping for the calculated intervals until shutdown.
Returns
| Type | Description |
|---|
None | Nothing |
stop()
@classmethod
def stop() - > None
Signals the timer thread to shut down and waits for it to terminate gracefully.
Returns
| Type | Description |
|---|
None | Nothing |
ensure_started()
@classmethod
def ensure_started() - > None
Verifies if the timer thread is active and starts it if it is currently stopped or not yet alive.
Returns
| Type | Description |
|---|
None | Nothing |
enter()
@classmethod
def enter(
entry: Entry,
eta: float,
priority: Optional[int]
) - > Entry
Schedules an existing entry to be executed at a specific absolute time.
Parameters
| Name | Type | Description |
|---|
| entry | Entry | The entry object containing the task to be executed |
| eta | float | The absolute timestamp (Estimated Time of Arrival) when the entry should run |
| priority | Optional[int] | The execution priority used to break ties for events scheduled at the same time |
Returns
| Type | Description |
|---|
Entry | The scheduled entry object |
call_at()
@classmethod
def call_at() - > Entry
Schedules a function call to occur at a specific absolute timestamp.
Returns
| Type | Description |
|---|
Entry | The entry object representing the scheduled call |
enter_after()
@classmethod
def enter_after() - > Entry
Schedules an existing entry to be executed after a relative delay in seconds.
Returns
| Type | Description |
|---|
Entry | The scheduled entry object |
call_after()
@classmethod
def call_after() - > Entry
Schedules a function call to occur after a specified number of seconds.
Returns
| Type | Description |
|---|
Entry | The entry object representing the scheduled call |
call_repeatedly()
@classmethod
def call_repeatedly() - > Entry
Schedules a function to be called at fixed intervals repeatedly.
Returns
| Type | Description |
|---|
Entry | The entry object representing the recurring task |
exit_after()
@classmethod
def exit_after(
secs: float,
priority: int
) - > None
Schedules the entire process to exit after a specified delay.
Parameters
| Name | Type | Description |
|---|
| secs | float | The number of seconds to wait before calling sys.exit |
| priority | int | The priority of the exit task relative to other scheduled events |
Returns
| Type | Description |
|---|
None | Nothing |
cancel()
@classmethod
def cancel(
tref: Entry
) - > None
Cancels a previously scheduled entry so that it will not be executed.
Parameters
| Name | Type | Description |
|---|
| tref | Entry | The reference to the entry that should be cancelled |
Returns
| Type | Description |
|---|
None | Nothing |
clear()
@classmethod
def clear() - > None
Removes all scheduled entries from the timer's queue.
Returns
| Type | Description |
|---|
None | Nothing |
empty()
@classmethod
def empty() - > bool
Checks whether the timer's schedule queue is currently empty.
Returns
| Type | Description |
|---|
bool | True if there are no scheduled entries, False otherwise |
queue()
@classmethod
def queue() - > list
Provides access to the list of currently scheduled entries.
Returns
| Type | Description |
|---|
list | A list of Entry objects representing the current schedule |