Sublime Forum

Debounce `on_modified` and `on_selection_modified` events with a decorator

#1

Since this is a recurring problem with writing plugins that listen to changes to the buffer (selection) but do not want to execute on every change, I’ve written a small self-sufficient decorator that you can use on such events to debounce its execution. It may or may not make it into the sublime_lib dependency library available to Package Control at some point.

See the post linked below for the latest version.

from functools import partial, wraps

import sublime
import sublime_plugin


def _debounced_callback(view, old_change_count, callback):
    if view.is_valid() and view.change_count() == old_change_count:
        callback()


def debounced(delay_in_ms, sync=False):
    """Delay calls to on_(selection_)modified(_async) event hooks until they weren't triggered for n ms.

    Works on both `EventListener` and `ViewEventListener` classes.

    Calls are only made when the `view` is still "valid" according to ST's API,
    so it's not necessary to check it in the wrapped function.
    """

    def decorator(func):
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            view = self.view if hasattr(self, 'view') else args[0]
            if not view.is_valid():
                return
            change_count = view.change_count()
            callback = partial(func, self, *args, **kwargs)
            set_timeout = sublime.set_timeout if sync else sublime.set_timeout_async
            set_timeout(partial(_debounced_callback, view, change_count, callback), delay_in_ms)

        return wrapper

    return decorator


class DebouncedListener(sublime_plugin.EventListener):
    @debounced(500)
    def on_modified_async(self, view):
        print("debounced on EventListener", view.id())


class DebouncedViewListener(sublime_plugin.ViewEventListener):
    @debounced(1000)
    def on_modified_async(self):
        print("debounced on ViewEventListener", self.view.id())
5 Likes
#2

The above version didn’t actually work properly for on_selection_modified hooks because it relied on the change count only. Furthermore, it would spawn quite a few potentially unnecessary set_timeout callbacks. The following snippet uses a slightly different approach and addresses these issues (and requires Python 3.8):

from functools import partial, wraps
import time

import sublime
import sublime_plugin


def debounced(delay_in_ms, sync=False):
    """Delay calls to event hooks until they weren't triggered for n ms.

    Performs view-specific tracking and is best suited for the
    `on_modified` and `on_selection_modified` methods
    and their `_async` variants.
    The `view` is taken from the first argument for `EventListener`s
    and from the instance for `ViewEventListener`s.

    Calls are only made when the `view` is still "valid" according to ST's API,
    so it's not necessary to check it in the wrapped function.
    """

    # We assume that locking is not necessary because each function will be called
    # from either the ui or the async thread only.
    set_timeout = sublime.set_timeout if sync else sublime.set_timeout_async

    def decorator(func):
        to_call_times = {}

        def _debounced_callback(view, callback):
            vid = view.id()
            if not (threshold := to_call_times.get(vid)) or not view.is_valid():
                return
            diff = threshold - time.time() * 1000
            if diff > 0:
                set_timeout(partial(_debounced_callback, view, callback), diff)
            else:
                del to_call_times[vid]
                callback()

        @wraps(func)
        def wrapper(self, *args, **kwargs):
            view = self.view if hasattr(self, 'view') else args[0]
            if not view.is_valid():
                return
            vid = view.id()
            call_time = time.time() * 1000 + delay_in_ms
            old_call_time = to_call_times.get(vid)
            to_call_times[vid] = call_time
            if old_call_time:
                return
            callback = partial(func, self, *args, **kwargs)
            set_timeout(partial(_debounced_callback, view, callback), delay_in_ms)

        return wrapper

    return decorator


class DebouncedListener(sublime_plugin.EventListener):
    @debounced(500)
    def on_modified(self, view):
        print("debounced EventListener.on_modified", view.id())

    @debounced(500)
    def on_modified_async(self, view):
        print("debounced EventListener.on_modified_async", view.id())


class DebouncedViewListener(sublime_plugin.ViewEventListener):
    @debounced(1000)
    def on_modified_async(self):
        print("debounced ViewEventListener.on_modified_async", self.view.id())

    @debounced(2000)
    def on_selection_modified_async(self):
        print("debounced ViewEventListener.on_selection_modified_async", self.view.id())

    @debounced(1500)
    def on_selection_modified(self):
        print("debounced ViewEventListener.on_selection_modified", self.view.id())
0 Likes
#3

I’d suggest following changes:

  1. remove vid from to_call_time if related view object is invalid.
  2. resolve python 3.3 incompatibility (I don’t se a valuable reason to use assignment expressions), especially not after (1.)
  3. determining a busy flag by checking, whether vid is in to_call_time, should be enough. No need to retrieve and check its value.

But nevertheless, this is the best approach to implement debouncing, I’ve seen so far. Nice work!!!

from functools import partial, wraps
import time

import sublime
import sublime_plugin


def debounced(delay_in_ms, sync=False):
    """Delay calls to event hooks until they weren't triggered for n ms.

    Performs view-specific tracking and is best suited for the
    `on_modified` and `on_selection_modified` methods
    and their `_async` variants.
    The `view` is taken from the first argument for `EventListener`s
    and from the instance for `ViewEventListener`s.

    Calls are only made when the `view` is still "valid" according to ST's API,
    so it's not necessary to check it in the wrapped function.
    """

    # We assume that locking is not necessary because each function will be called
    # from either the ui or the async thread only.
    set_timeout = sublime.set_timeout if sync else sublime.set_timeout_async

    def decorator(func):
        to_call_times = {}

        def _debounced_callback(view, callback):
            vid = view.id()
            threshold = to_call_times.get(vid)
            if not threshold:
                return
            if not view.is_valid():
                del to_call_times[vid]
                return
            diff = threshold - time.time() * 1000
            if diff > 0:
                set_timeout(partial(_debounced_callback, view, callback), diff)
            else:
                del to_call_times[vid]
                callback()

        @wraps(func)
        def wrapper(self, *args, **kwargs):
            view = self.view if hasattr(self, 'view') else args[0]
            if not view.is_valid():
                return
            vid = view.id()
            busy = vid in to_call_times
            to_call_times[vid] = time.time() * 1000 + delay_in_ms
            if busy:
                return
            callback = partial(func, self, *args, **kwargs)
            set_timeout(partial(_debounced_callback, view, callback), delay_in_ms)

        return wrapper

    return decorator


class DebouncedListener(sublime_plugin.EventListener):
    @debounced(500)
    def on_modified(self, view):
        print("debounced EventListener.on_modified", view.id())

    @debounced(500)
    def on_modified_async(self, view):
        print("debounced EventListener.on_modified_async", view.id())


class DebouncedViewListener(sublime_plugin.ViewEventListener):
    @debounced(1000)
    def on_modified_async(self):
        print("debounced ViewEventListener.on_modified_async", self.view.id())

    @debounced(2000)
    def on_selection_modified_async(self):
        print("debounced ViewEventListener.on_selection_modified_async", self.view.id())

    @debounced(1500)
    def on_selection_modified(self):
        print("debounced ViewEventListener.on_selection_modified", self.view.id())
2 Likes
#4

I came up with something kind of similar in order to execute code when a view is no longer loading. Usage: call_on_load(view, callback). This post made me realize that perhaps I should also check for view.is_valid().

import functools

import sublime


INTERVAL = 0.0001
TIMEOUT = 8.0


def set_timeout(value):  # just for testing
    global TIMEOUT
    TIMEOUT = value


def _wrap(cond, fun, *args, **kwargs):
    if not hasattr(fun, "__timeout"):
        fun.__timeout = TIMEOUT
        fun.__interval = INTERVAL
        fun.__timed_out = False
    if fun.__timeout <= 0:
        fun.__timed_out = True  # for testing
        raise TimeoutError(f"{fun!r} timed out")
    return functools.partial(call_when, cond, fun, *args, **kwargs)


def call_when(cond, fun, *args, **kwargs):
    """Calls function at a later time, if a condition is satisfied
    within a certain amount of time.
    """
    if cond():
        sublime.set_timeout(lambda: fun(*args, **kwargs), 0)
        return

    wrapper = _wrap(cond, fun, *args, **kwargs)
    sublime.set_timeout(wrapper, fun.__interval * 1000)
    # 0.0001, 0.0002, 0.0004, 0.0008, 0.0016, 0.0032, 0.0064,
    # 0.0128, 0.02, 0.02, 0.02, 0.02, ...
    fun.__interval = min(fun.__interval * 2, 0.02)
    fun.__timeout -= fun.__interval


def call_when_async(cond, fun, *args, **kwargs):
    if cond():
        sublime.set_timeout_async(lambda: fun(*args, **kwargs), 0)
        return

    wrapper = _wrap(cond, fun, *args, **kwargs)
    sublime.set_timeout_async(wrapper, fun.__interval * 1000)
    fun.__interval = min(fun.__interval * 2, 0.02)
    fun.__timeout -= fun.__interval


def call_on_load(view, fun, *args, **kwargs):
    call_when(lambda: not view.is_loading(), fun, *args, **kwargs)


def call_on_load_async(view, fun, *args, **kwargs):
    call_when_async(lambda: not view.is_loading(), fun, *args, **kwargs)
0 Likes