ipywidgets 事件节流

# 为了保证 `JupyterLite` 可用，需要 notebook 开头添加：
%pip install -q ipywidgets

Note: you may need to restart the kernel to use updated packages.

节流(Throttling)是另一种可用于限制回调的技术。与去抖动不同，去抖动在自上一次（尝试）调用函数以来未经过一定时间时会忽略对函数的调用，而节流则会限制调用的速率。这确保了函数被定期调用。

下面展示了一个同步解决方案。同样地，如果你想使用线程而不是异步编程，你可以用 from threading import Timer 替换 Timer 类。

import asyncio
from time import time

class Timer:
    def __init__(self, timeout, callback):
        self._timeout = timeout
        self._callback = callback

    async def _job(self):
        await asyncio.sleep(self._timeout)
        self._callback()

    def start(self):
        self._task = asyncio.ensure_future(self._job())

    def cancel(self):
        self._task.cancel()

def throttle(wait):
    """ Decorator that prevents a function from being called
        more than once every wait period. """
    def decorator(fn):
        time_of_last_call = 0
        scheduled, timer = False, None
        new_args, new_kwargs = None, None
        def throttled(*args, **kwargs):
            nonlocal new_args, new_kwargs, time_of_last_call, scheduled, timer
            def call_it():
                nonlocal new_args, new_kwargs, time_of_last_call, scheduled, timer
                time_of_last_call = time()
                fn(*new_args, **new_kwargs)
                scheduled = False
            time_since_last_call = time() - time_of_last_call
            new_args, new_kwargs = args, kwargs
            if not scheduled:
                scheduled = True
                new_wait = max(0, wait - time_since_last_call)
                timer = Timer(new_wait, call_it)
                timer.start()
        return throttled
    return decorator

要了解它与去抖动器的行为有何不同，这里是一个相同的滑块示例，其节流值显示在文本框中。注意它在限制回调速率的同时，交互性有多强。

import ipywidgets as widgets
slider = widgets.IntSlider()
text = widgets.IntText()

@throttle(0.2)
def value_changed(change):
    text.value = change.new
slider.observe(value_changed, 'value')

widgets.VBox([slider, text])