Skip to content

Schedulers

pamiq_core.utils.schedulers.Scheduler 

Scheduler(callbacks: Callback | CallbackIterable | None = None)

Bases: ABC

Abstract base class for schedulers.

Schedulers periodically execute registered callbacks based on specific conditions defined by subclasses. The base class provides functionality for managing callbacks.

Initialize the scheduler.

PARAMETER DESCRIPTION
callbacks

Optional iterable of callback functions to register or callback.

TYPE: Callback | CallbackIterable | None DEFAULT: None

Source code in src/pamiq_core/utils/schedulers.py 
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(self, callbacks: Callback | CallbackIterable | None = None) -> None:
    """Initialize the scheduler.

    Args:
        callbacks: Optional iterable of callback functions to register or callback.
    """
    if callbacks is None:
        callbacks = []
    if isinstance(callbacks, Callable):
        callbacks = [callbacks]

    self._callbacks = list(callbacks)

register_callback 

register_callback(callback: Callback) -> None

Register a new callback function.

PARAMETER DESCRIPTION
callback

The callback function to register.

TYPE: Callback

Source code in src/pamiq_core/utils/schedulers.py 
32
33
34
35
36
37
38
def register_callback(self, callback: Callback) -> None:
    """Register a new callback function.

    Args:
        callback: The callback function to register.
    """
    self._callbacks.append(callback)

remove_callback 

remove_callback(callback: Callback) -> None

Remove a registered callback function.

PARAMETER DESCRIPTION
callback

The callback function to remove.

TYPE: Callback

RAISES DESCRIPTION
ValueError

If the callback is not registered.
Source code in src/pamiq_core/utils/schedulers.py 
40
41
42
43
44
45
46
47
48
49
def remove_callback(self, callback: Callback) -> None:
    """Remove a registered callback function.

    Args:
        callback: The callback function to remove.

    Raises:
        ValueError: If the callback is not registered.
    """
    self._callbacks.remove(callback)

is_available abstractmethod 

is_available() -> bool

Check if the scheduler should execute callbacks now.

RETURNS DESCRIPTION
bool

True if callbacks should be executed, False otherwise.
Source code in src/pamiq_core/utils/schedulers.py 
51
52
53
54
55
56
57
58
@abstractmethod
def is_available(self) -> bool:
    """Check if the scheduler should execute callbacks now.

    Returns:
        True if callbacks should be executed, False otherwise.
    """
    pass

update abstractmethod 

update() -> None

Check if callbacks should be executed and run them if so.

This method should be called regularly to check if execution conditions are met.

Source code in src/pamiq_core/utils/schedulers.py 
60
61
62
63
64
65
66
67
68
69
@abstractmethod
def update(self) -> None:
    """Check if callbacks should be executed and run them if so.

    This method should be called regularly to check if execution
    conditions are met.
    """
    if self.is_available():
        for callback in self._callbacks:
            callback()

pamiq_core.utils.schedulers.TimeIntervalScheduler 

TimeIntervalScheduler(
    interval: float, callbacks: Callback | CallbackIterable | None = None
)

Bases: Scheduler

Scheduler that executes callbacks at specified time intervals.

This scheduler triggers callbacks when a specified amount of time has elapsed since the last execution.

Initialize the time interval scheduler.

PARAMETER DESCRIPTION
interval

Time interval in seconds between executions.

TYPE: float

callbacks

Optional iterable of callback functions to register.

TYPE: Callback | CallbackIterable | None DEFAULT: None

RAISES DESCRIPTION
ValueError

If interval is negative.
Source code in src/pamiq_core/utils/schedulers.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@override
def __init__(
    self, interval: float, callbacks: Callback | CallbackIterable | None = None
) -> None:
    """Initialize the time interval scheduler.

    Args:
        interval: Time interval in seconds between executions.
        callbacks: Optional iterable of callback functions to register.

    Raises:
        ValueError: If interval is negative.
    """
    super().__init__(callbacks)
    if interval < 0:
        raise ValueError("Interval must be non-negative")
    self._interval = interval
    self._previous_available_time = float("-inf")

is_available 

is_available() -> bool

Check if the specified time interval has elapsed.

RETURNS DESCRIPTION
bool

True if the time interval has elapsed, False otherwise.
Source code in src/pamiq_core/utils/schedulers.py 
 98
 99
100
101
102
103
104
105
@override
def is_available(self) -> bool:
    """Check if the specified time interval has elapsed.

    Returns:
        True if the time interval has elapsed, False otherwise.
    """
    return time.time() - self._previous_available_time > self._interval

update 

update() -> None

Check if the time interval has elapsed and execute callbacks if so.

Updates the previous execution time after callbacks are executed.

Source code in src/pamiq_core/utils/schedulers.py 
107
108
109
110
111
112
113
114
115
116
@override
def update(self) -> None:
    """Check if the time interval has elapsed and execute callbacks if so.

    Updates the previous execution time after callbacks are
    executed.
    """
    super().update()  # Call parent to execute callbacks if available
    if self.is_available():
        self._previous_available_time = time.time()

pamiq_core.utils.schedulers.StepIntervalScheduler 

StepIntervalScheduler(
    interval: int, callbacks: Callback | CallbackIterable | None = None
)

Bases: Scheduler

Scheduler that executes callbacks after a specified number of steps.

This scheduler triggers callbacks when a specified number of steps have been completed since the last execution.

Initialize the step interval scheduler.

PARAMETER DESCRIPTION
interval

Number of steps between executions.

TYPE: int

callbacks

Optional iterable of callback functions to register.

TYPE: Callback | CallbackIterable | None DEFAULT: None

RAISES DESCRIPTION
ValueError

If interval is not positive.
Source code in src/pamiq_core/utils/schedulers.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@override
def __init__(
    self, interval: int, callbacks: Callback | CallbackIterable | None = None
) -> None:
    """Initialize the step interval scheduler.

    Args:
        interval: Number of steps between executions.
        callbacks: Optional iterable of callback functions to register.

    Raises:
        ValueError: If interval is not positive.
    """
    super().__init__(callbacks)
    if interval <= 0:
        raise ValueError("Interval must be positive")
    self._interval = interval
    self._steps_since_last_call = 0

is_available 

is_available() -> bool

Check if the specified number of steps has been reached.

RETURNS DESCRIPTION
bool

True if the step interval has been reached, False otherwise.
Source code in src/pamiq_core/utils/schedulers.py 
145
146
147
148
149
150
151
152
@override
def is_available(self) -> bool:
    """Check if the specified number of steps has been reached.

    Returns:
        True if the step interval has been reached, False otherwise.
    """
    return self._steps_since_last_call >= self._interval

update 

update() -> None

Increment step counter, execute callbacks if interval is reached.

Resets the step counter after callbacks are executed.

Source code in src/pamiq_core/utils/schedulers.py 
154
155
156
157
158
159
160
161
162
163
@override
def update(self) -> None:
    """Increment step counter, execute callbacks if interval is reached.

    Resets the step counter after callbacks are executed.
    """
    self._steps_since_last_call += 1
    super().update()  # Call parent to execute callbacks if available
    if self.is_available():
        self._steps_since_last_call = 0