用于去抖动的 Python 装饰器，包括函数参数

Question

如何在 python 中编写一个 debounce 装饰器，它不仅在调用的函数上而且在函数参数/使用的函数参数组合上去抖动？

去抖动意味着在给定的时间范围内抑制对函数的调用，假设您在 1 秒内调用函数 100 次，但您只想让函数每 10 秒运行一次，去抖动修饰的函数将运行一次函数如果没有进行新的函数调用，则在最后一次函数调用后 10 秒。在这里，我要问的是如何使用特定的函数参数来消除函数调用的抖动。

一个例子可能是去抖动一个人对象的昂贵更新，例如：

@debounce(seconds=10)
def update_person(person_id):
    # time consuming, expensive op
    print('>>Updated person {}'.format(person_id))

然后在函数上去抖动 - 包括函数参数：

update_person(person_id=144)
update_person(person_id=144)
update_person(person_id=144)
>>Updated person 144

update_person(person_id=144)
update_person(person_id=355)
>>Updated person 144
>>Updated person 355

因此，调用具有相同 person_id 的函数 update_person 将被抑制（去抖动），直到 10 秒的去抖动间隔过去，而没有对具有相同 person_id 的函数的新调用。

有一些 debounce 装饰器，但没有一个包含函数参数，例如：https://gist.github.com/walkermatt/2871026

我已经通过函数和参数做了一个类似的油门装饰器：

def throttle(s, keep=60):

    def decorate(f):

        caller = {}

        def wrapped(*args, **kwargs):
            nonlocal caller

            called_args = '{}'.format(*args)
            t_ = time.time()

            if caller.get(called_args, None) is None or t_ - caller.get(called_args, 0) >= s:
                result = f(*args, **kwargs)

                caller = {key: val for key, val in caller.items() if t_ - val > keep}
                caller[called_args] = t_
                return result

            # Keep only calls > keep
            caller = {key: val for key, val in caller.items() if t_ - val > keep}
            caller[called_args] = t_

        return wrapped

    return decorate

主要的要点是它将函数参数保存在调用者[caller_args]中

另见油门和去抖动的区别：http://demo.nimius.net/debounce_throttle/

更新：

在对上述节流装饰器和要点中的 threading.Timer 示例进行一些修改之后，我实际上认为这应该可以工作：

from threading import Timer
from inspect import signature
import time


def debounce(wait):
    def decorator(fn):
        sig = signature(fn)
        caller = {}

        def debounced(*args, **kwargs):
            nonlocal caller

            try:
                bound_args = sig.bind(*args, **kwargs)
                bound_args.apply_defaults()
                called_args = fn.__name__ + str(dict(bound_args.arguments))
            except:
                called_args = ''

            t_ = time.time()

            def call_it(key):
                try:
                    # always remove on call
                    caller.pop(key)
                except:
                    pass

                fn(*args, **kwargs)

            try:
                # Always try to cancel timer
                caller[called_args].cancel()
            except:
                pass

            caller[called_args] = Timer(wait, call_it, [called_args])
            caller[called_args].start()

        return debounced

    return decorator

Answer 1

我同样需要为个人项目构建去抖动注释，在偶然发现the same gist / discussion 之后，我最终得到了以下解决方案：

import threading
def debounce(wait_time):
    """
    Decorator that will debounce a function so that it is called after wait_time seconds
    If it is called multiple times, will wait for the last call to be debounced and run only this one.
    """

    def decorator(function):
        def debounced(*args, **kwargs):
            def call_function():
                debounced._timer = None
                return function(*args, **kwargs)
            # if we already have a call to the function currently waiting to be executed, reset the timer
            if debounced._timer is not None:
                debounced._timer.cancel()

            # after wait_time, call the function provided to the decorator with its arguments
            debounced._timer = threading.Timer(wait_time, call_function)
            debounced._timer.start()

        debounced._timer = None
        return debounced

    return decorator

我创建了一个开源项目来提供 debounce、throttle、filter 等功能......作为装饰器，非常欢迎贡献来改​​进我为这些装饰器提供的解决方案/添加其他有用的装饰器：@ 987654322@