Python Debounce

debounce.py

"""A ``debounce`` decorator implementation in Python similar to debounce_ in ``lodash``.

.. _debounce: https://lodash.com/docs#debounce
"""
from functools import wraps
from threading import Timer
import time
from typing import Any, Callable, Dict, Optional, Tuple


class DebouncedCallable:
    """Represents a callable returned from ``debounce``."""

    _pending: bool

    def __call__(self, *args, **kwargs):
        super.__call__(*args, **kwargs)

    def cancel(self):
        """Cancel the delayed function invocation."""

    def flush(self):
        """Immediately invoke the delayed function."""

    @property
    def pending(self):
        """Indicate if there is currently a pending invocation."""
        return self._pending


def debounce(  # noqa: C901
    wait: float,
    max_wait: Optional[float] = None,
    leading: Optional[bool] = False,
    trailing: Optional[bool] = True,
):  # pylint:disable=too-many-locals,too-many-statements
    """A decorator used to create a debounced function.

    Creates a debounced function that delays invoking `func` until after `wait`
    seconds 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. Provide ``leading``
    or ``trailing`` 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 debounced function. Subsequent calls to the debounced function
    return the result of the last ``func`` invocation.

    Notes:
        If both ``leading`` and ``trailing`` options are ``True``, ``func`` is invoked
        on the trailing edge of the timeout only if the debounced function is invoked
        more than once during the ``wait`` timeout.

    Examples:
        ::

            @debounce(2, max_wait=4)
            def debounced_print(text: str):
                print(text)


            debounced_print: DebouncedCallable
            for _ in range(10):
                debounced_print("Hello")
                time.sleep(1)
            debounced_print("Hello")
            debounced_print("World!")

    Args:
        wait: The number of seconds to delay.
        max_wait: The maximum time ``func`` is allowed to be delayed before it's
            invoked.
        leading: Whether to invoke on the leading edge of the timeout.
        trailing: Whether to invoke on the trailing edge of the timeout.

    Returns:
        A decorator used to debounce a function.
    """
    if max_wait:
        max_wait = max(max_wait, wait)

    def _decorator(func: Callable):
        """Inner decorator."""
        _last_args: Tuple = ()
        _last_kwargs: Dict[str, Any] = {}
        _last_call_time: Optional[float] = None
        _last_invoke_time: float = 0
        _result: Optional[Any] = None
        _timer: Optional[Timer] = None

        def _invoke_func(time_: float):
            """Invoke ``func``."""
            nonlocal _last_args, _last_kwargs, _last_invoke_time, _result

            args = _last_args
            kwargs = _last_kwargs

            _last_args = ()
            _last_kwargs = {}
            _last_invoke_time = time_
            _result = func(*args, **kwargs)
            return _result

        def _start_timer(timeout: float, func_: Callable, *args, **kwargs):
            """Start a timer to invoke ``func_`` after ``timeout`` seconds."""
            _timer = Timer(timeout, func_, *args, **kwargs)
            _timer.start()
            return _timer

        def _cancel_timer(timer_: Timer):
            """Stop the ``timer_`` if it hasn't finished yet."""
            timer_.cancel()

        def _leading_edge(time_: float):
            """Invoke ``func`` at the leading edge."""
            # Reset any `max_wait` timer.
            nonlocal _timer, _last_invoke_time

            _last_invoke_time = time_
            # Start the timer for the trailing edge.
            _timer = _start_timer(wait, _timer_expired)
            # Invoke the leading edge.
            return _invoke_func(time_) if leading else _result

        def _remaining_wait(time_: float):
            """Return the remaining wait time until the next invocation."""
            time_since_last_call = time_ - _last_call_time
            remaining_time = wait - time_since_last_call

            if max_wait:
                time_since_last_invoke = time_ - _last_invoke_time
                return min(remaining_time, max_wait - time_since_last_invoke)

            return remaining_time

        def _should_invoke(time_: float):
            """Determine whether to invoke ``func``.

            Returns:
                ``True`` in any of these cases
                    - This is the first call.
                    - Activity has stopped, and we're at the trailing edge.
                    - The system time has gone backwards, and we're treating it as the
                      trailing edge.
                    - We've hit the `max_wait` limit.
                ``False`` otherwise.
            """
            if not _last_call_time:
                return True

            time_since_last_call = time_ - _last_call_time
            if time_since_last_call >= wait or time_since_last_call < 0:
                return True

            if max_wait:
                time_since_last_invoke = time_ - _last_invoke_time
                return time_since_last_invoke >= max_wait

            return False

        def _timer_expired():
            """Check if the timer has expired and invoke ``func`` at the trailing edge.

            Otherwise, schedule the next invocation.
            """
            nonlocal _timer

            now = time.time()
            if _should_invoke(now):
                return _trailing_edge(now)

            # Restart the timer.
            _timer = _start_timer(_remaining_wait(now), _timer_expired)
            return None

        def _trailing_edge(time_: float):
            """Invoke ``func`` at the trailing edge."""
            nonlocal _timer, _last_args, _last_kwargs

            _timer = None

            # Only invoke if we have `last_args` which means `func` has been debounced
            #   at least once.
            if trailing and _last_args:
                return _invoke_func(time_)

            _last_args = ()
            _last_kwargs = {}
            return _result

        def _cancel():
            """Cancel the trailing debounced invocation."""
            # pylint:disable=line-too-long
            # noqa: B950
            nonlocal _timer, _last_args, _last_kwargs, _last_call_time, _last_invoke_time

            if _timer:
                _cancel_timer(_timer)

            _last_invoke_time = 0
            _last_args = ()
            _last_kwargs = {}
            _last_call_time = _timer = None

        def _flush():
            """Immediately invoke ``func``."""
            return _result if not _timer else _trailing_edge(time.time())

        @property
        def _pending():
            """Return whether there is currently a pending invocation."""
            return bool(_timer)

        @wraps(func)
        def _debounced(*args, **kwargs):
            """The debounced version of ``func``."""
            nonlocal _timer, _last_args, _last_kwargs, _last_call_time

            now = time.time()
            should_invoke = _should_invoke(now)

            _last_call_time = now
            _last_args = args
            _last_kwargs = kwargs

            if should_invoke:
                if not _timer:
                    return _leading_edge(_last_call_time)

                if max_wait:
                    # Handle invocations in a tight loop.
                    _timer = _start_timer(wait, _timer_expired)
                    return _invoke_func(_last_call_time)

            if not _timer:
                _timer = _start_timer(wait, _timer_expired)

            return _result

        _debounced.cancel = _cancel
        _debounced.flush = _flush
        _debounced.pending = _pending

        return _debounced

    return _decorator


@debounce(2, max_wait=4)
def debounced_print(text: str):
    print(text)


if __name__ == "__main__":
    debounced_print: DebouncedCallable
    for _ in range(10):
        debounced_print("Hello")
        time.sleep(1)
    debounced_print("Hello")
    debounced_print("World!")