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() |
This comment has been minimized.
This comment has been minimized.
Hey @vladignatyev The first link you mentioned all calls sleep() A proper debounce should never call sleep. I just wrote a better one here: |
This comment has been minimized.
This comment has been minimized.
Actually, @esromneb, your implementation behaves as the throttle function of underscore.js. The difference being your implementation triggers on the first call, and blocks any calls made in the specified period afterwards. The debounce function of underscore.js however, waits the specified amount of time after the first call, before actually executing it. If in that waiting period another call is made, the timer is reset. I can't think of an easy way to achieve this without sleep. Whether this is how a debounce function works in an electrical system I do not know, but which functionality is preferred depends on the use case. In my scenario, I wish to process a file after it has stopped being changed for a while, thus the underscore.js version is prefered. |
This comment has been minimized.
This comment has been minimized.
FWIW, here's a version of the debounce decorator that works in Twisted Python. from twisted.internet import reactor
def debounce(wait):
def decorator(fn):
def debounced(*args, **kw):
try:
debounced.delayed_call
except AttributeError:
def call_it():
del debounced.delayed_call
fn(*args, **kw)
debounced.delayed_call = reactor.callLater(wait, call_it)
else:
debounced.delayed_call.reset(wait)
return debounced
return decorator |
This comment has been minimized.
This comment has been minimized.
Here's another way. This uses Python 3's nonlocal keyword. The implementation is direct and very easy to understand. import time
def debounce(s):
"""Decorator ensures function that can only be called once every `s` seconds.
"""
def decorate(f):
t = None
def wrapped(*args, **kwargs):
nonlocal t
t_ = time.time()
if t is None or t_ - t >= s:
result = f(*args, **kwargs)
t = time.time()
return result
return wrapped
return decorate Try it out. @debounce(3)
def hi(name):
print('hi {}'.format(name))
hi('dude')
time.sleep(1)
hi('mike')
time.sleep(1)
hi('mary')
time.sleep(1)
hi('jane') |
This comment has been minimized.
This comment has been minimized.
@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 |
This comment has been minimized.
This comment has been minimized.
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 |
This comment has been minimized.
This comment has been minimized.
@neekey |
This comment has been minimized.
This comment has been minimized.
@kylebebak here is a great visualization for understanding throttle vs debounce: |
This comment has been minimized.
This comment has been minimized.
Didn't know about An optimization would be to protect the Another would be to use |
This comment has been minimized.
Look at https://github.com/vladignatyev/limiter
It is like debounce but it doesn't use timer and locks current thread.
Your implementation is similar to this one from JS LoDash framework: https://github.com/lodash/lodash/blob/master/dist/lodash.js#L4975
Have a nice day!