-
-
Save walkermatt/2871026 to your computer and use it in GitHub Desktop.
from threading import Timer | |
def debounce(wait): | |
""" Decorator that will postpone a functions | |
execution until after wait seconds | |
have elapsed since the last time it was invoked. """ | |
def decorator(fn): | |
def debounced(*args, **kwargs): | |
def call_it(): | |
fn(*args, **kwargs) | |
try: | |
debounced.t.cancel() | |
except(AttributeError): | |
pass | |
debounced.t = Timer(wait, call_it) | |
debounced.t.start() | |
return debounced | |
return decorator |
import unittest | |
import time | |
from debounce import debounce | |
class TestDebounce(unittest.TestCase): | |
@debounce(10) | |
def increment(self): | |
""" Simple function that | |
increments a counter when | |
called, used to test the | |
debounce function decorator """ | |
self.count += 1 | |
def setUp(self): | |
self.count = 0 | |
def test_debounce(self): | |
""" Test that the increment | |
function is being debounced. | |
The counter should only be incremented | |
once 10 seconds after the last call | |
to the function """ | |
self.assertTrue(self.count == 0) | |
self.increment() | |
self.increment() | |
time.sleep(9) | |
self.assertTrue(self.count == 0) | |
self.increment() | |
self.increment() | |
self.increment() | |
self.increment() | |
self.assertTrue(self.count == 0) | |
time.sleep(10) | |
self.assertTrue(self.count == 1) | |
if __name__ == '__main__': | |
unittest.main() |
def debounce(wait):
""" Decorator that will postpone a functions
execution until after wait seconds
have elapsed since the last time it was invoked. """
def decorator(fn):
def debounced(*args, **kwargs):
def call_it():
debounced._timer = None
debounced._last_call = time.time()
return fn(*args, **kwargs)
time_since_last_call = time.time() - debounced._last_call
if time_since_last_call >= wait:
return call_it()
if debounced._timer is None:
debounced._timer = threading.Timer(wait - time_since_last_call, call_it)
debounced._timer.start()
debounced._timer = None
debounced._last_call = 0
return debounced
return decorator
This debounce decorator will directly call the function if the last call is more than wait
seconds ago. Otherwise it starts a timer to call the function after wait
seconds. This means if the decorated function is called once every second, the first call is passed through and the second call (which happens 1s after the first call) is scheduled to be executed after 14s. Calls that happen after a call has been scheduled will be ignored. One could also change the decorator so that always the last given arguments are used for the next scheduled call.
@kylebebak your version is not "bounce" but as your description
ensures function that can only be called once every
s
seconds.you version will always invoke the first call, and all the calls afterwards before the "s" seconds will be omitted
@neekey
Thanks for mentioning this, you're totally right! I think what I've written is typically called throttle, but I'm not sure this is the right name either.
@kylebebak here is a great visualization for understanding throttle vs debounce:
http://demo.nimius.net/debounce_throttle/
Didn't know about threading.Timer
, I like this solution!
An optimization would be to protect the Timer
creation using a threading.Lock
. This would prevent problems with debounce
being called between Timer
's initialization and calling .start()
.
Another would be to use functools.wraps
to preserve docstrings.
Hi everyone ! A few months ago I had the same need to have a working debounce annotation, after stumbling upon this discussion I created this open source project: https://github.com/salesforce/decorator-operations
The idea of the project is to regroup useful annotations such as debounce, throttle, filter... There are only 4 annotations available for now, but you're more than welcome to suggest new features or suggestions on how to improve the existing ones :D
👍 thanks for sharing
@KarlPatach that sounds good 👍
// feature 1
In addition to debounce
and throttle
also has a situation, which can has both feature.
if calls too frequently, the executor may delayed indefinitely.
some states cannot be updated in time.
then wish debounce has a max delay time allow execute one time.
// I don't known how to call it, has a debounce time, and a max delay time
// feature 2
if asyncio version has supported, that will be nice.
// finally
I wonder is them thread safe
@KarlPatach That's awesome 🎉
My implementation (fully-typed and thread-safe):
import threading
from typing import Any, Callable, Optional, TypeVar, cast
class Debouncer:
def __init__(self, f: Callable[..., Any], interval: float):
self.f = f
self.interval = interval
self._timer: Optional[threading.Timer] = None
self._lock = threading.Lock()
def __call__(self, *args, **kwargs) -> None:
with self._lock:
if self._timer is not None:
self._timer.cancel()
self._timer = threading.Timer(self.interval, self.f, args, kwargs)
self._timer.start()
VoidFunction = TypeVar("VoidFunction", bound=Callable[..., None])
def debounce(interval: float):
"""
Wait `interval` seconds before calling `f`, and cancel if called again.
The decorated function will return None immediately,
ignoring the delayed return value of `f`.
"""
def decorator(f: VoidFunction) -> VoidFunction:
if interval <= 0:
return f
return cast(VoidFunction, Debouncer(f, interval))
return decorator
@kylebebak your version is not "bounce" but as your description
you version will always invoke the first call, and all the calls afterwards before the "s" seconds will be omitted