Skip to content

Throttling & Debouncing#

Throttling and debouncing are techniques used to prevent a frequently-emitted signal from triggering a specific callback more than some amount in a specified amount of time.

Throttling means preventing a callback from being called if it has recently been called; it is useful when the callback is expensive.

Debouncing means waiting for a period of time to pass before calling the callback; it is useful when you'd like to wait a moment to see if a user might do additional actions (say, moving a slider or typing in a text field) before "committing" to calling the callback.

psygnal.throttled(func=None, timeout=100, leading=True) #

Create a throttled function that invokes func at most once per timeout.

The throttled function comes with a cancel method to cancel delayed func invocations and a flush method to immediately invoke them. Options to indicate whether func should be invoked on the leading and/or trailing edge of the wait timeout. The func is invoked with the last arguments provided to the throttled function. Subsequent calls to the throttled function return the result of the last func invocation.

This decorator may be used with or without parameters.

Parameters:

  • func (Callable) –

    A function to throttle

  • timeout (int) –

    Timeout in milliseconds to wait before allowing another call, by default 100

  • leading (bool) –

    Whether to invoke the function on the leading edge of the wait timer, by default True

Examples:

from psygnal import Signal, throttled

class MyEmitter:
    changed = Signal(int)

def on_change(val: int)
    # do something possibly expensive
    ...

emitter = MyEmitter()

# connect the `on_change` whenever `emitter.changed` is emitted
# BUT, no more than once every 50 milliseconds
emitter.changed.connect(throttled(on_change, timeout=50))
Source code in psygnal/_throttler.py
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def throttled(
    func: Callable[P, Any] | None = None,
    timeout: int = 100,
    leading: bool = True,
) -> Throttler[P] | Callable[[Callable[P, Any]], Throttler[P]]:
    """Create a throttled function that invokes func at most once per timeout.

    The throttled function comes with a `cancel` method to cancel delayed func
    invocations and a `flush` method to immediately invoke them. Options
    to indicate whether func should be invoked on the leading and/or trailing
    edge of the wait timeout. The func is invoked with the last arguments provided
    to the throttled function. Subsequent calls to the throttled function return
    the result of the last func invocation.

    This decorator may be used with or without parameters.

    Parameters
    ----------
    func : Callable
        A function to throttle
    timeout : int
        Timeout in milliseconds to wait before allowing another call, by default 100
    leading : bool
        Whether to invoke the function on the leading edge of the wait timer,
        by default True

    Examples
    --------
    ```python
    from psygnal import Signal, throttled

    class MyEmitter:
        changed = Signal(int)

    def on_change(val: int)
        # do something possibly expensive
        ...

    emitter = MyEmitter()

    # connect the `on_change` whenever `emitter.changed` is emitted
    # BUT, no more than once every 50 milliseconds
    emitter.changed.connect(throttled(on_change, timeout=50))
    ```
    """

    def deco(func: Callable[P, Any]) -> Throttler[P]:
        policy: EmissionPolicy = "leading" if leading else "trailing"
        return Throttler(func, timeout, policy)

    return deco(func) if func is not None else deco

psygnal.debounced(func=None, timeout=100, leading=False) #

Create a debounced function that delays invoking func.

func will not be invoked until timeout ms have elapsed since the last time the debounced function was invoked.

The debounced function comes with a cancel method to cancel delayed func invocations and a flush method to immediately invoke them. Options indicate whether func should be invoked on the leading and/or trailing edge of the wait timeout. The func is invoked with the last arguments provided to the debounced function. Subsequent calls to the debounced function return the result of the last func invocation.

This decorator may be used with or without parameters.

Parameters:

  • func (Callable) –

    A function to throttle

  • timeout (int) –

    Timeout in milliseconds to wait before allowing another call, by default 100

  • leading (bool) –

    Whether to invoke the function on the leading edge of the wait timer, by default False

Examples:

from psygnal import Signal, debounced

class MyEmitter:
    changed = Signal(int)

def on_change(val: int)
    # do something possibly expensive
    ...

emitter = MyEmitter()

# connect the `on_change` whenever `emitter.changed` is emitted
# ONLY once at least 50 milliseconds have passed since the last signal emission.
emitter.changed.connect(debounced(on_change, timeout=50))
Source code in psygnal/_throttler.py
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def debounced(
    func: Callable[P, Any] | None = None,
    timeout: int = 100,
    leading: bool = False,
) -> Debouncer[P] | Callable[[Callable[P, Any]], Debouncer[P]]:
    """Create a debounced function that delays invoking `func`.

    `func` will not be invoked until `timeout` ms have elapsed since the last time
    the debounced function was invoked.

    The debounced function comes with a `cancel` method to cancel delayed func
    invocations and a `flush` method to immediately invoke them. Options
    indicate whether func should be invoked on the leading and/or trailing edge
    of the wait timeout. The func is invoked with the *last* arguments provided to
    the debounced function. Subsequent calls to the debounced function return the
    result of the last `func` invocation.

    This decorator may be used with or without parameters.

    Parameters
    ----------
    func : Callable
        A function to throttle
    timeout : int
        Timeout in milliseconds to wait before allowing another call, by default 100
    leading : bool
        Whether to invoke the function on the leading edge of the wait timer,
        by default False

    Examples
    --------
    ```python
    from psygnal import Signal, debounced

    class MyEmitter:
        changed = Signal(int)

    def on_change(val: int)
        # do something possibly expensive
        ...

    emitter = MyEmitter()

    # connect the `on_change` whenever `emitter.changed` is emitted
    # ONLY once at least 50 milliseconds have passed since the last signal emission.
    emitter.changed.connect(debounced(on_change, timeout=50))
    ```
    """

    def deco(func: Callable[P, Any]) -> Debouncer[P]:
        policy: EmissionPolicy = "leading" if leading else "trailing"
        return Debouncer(func, timeout, policy)

    return deco(func) if func is not None else deco