Add Debounce, debounced, and throttled to fastcore.aio

fastcore/_modidx.py

@@ -8,9 +8,20 @@
   'syms': { 'fastcore.aio': { 'fastcore.aio.CachedAwaitable': ('aio.html#cachedawaitable', 'fastcore/aio.py'),
                               'fastcore.aio.CachedAwaitable.__await__': ('aio.html#cachedawaitable.__await__', 'fastcore/aio.py'),
                               'fastcore.aio.CachedAwaitable.__init__': ('aio.html#cachedawaitable.__init__', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce': ('aio.html#debounce', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.__call__': ('aio.html#debounce.__call__', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.__get__': ('aio.html#debounce.__get__', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.__init__': ('aio.html#debounce.__init__', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.__set_name__': ('aio.html#debounce.__set_name__', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce._fire': ('aio.html#debounce._fire', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce._put_latest': ('aio.html#debounce._put_latest', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce._run': ('aio.html#debounce._run', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.cancel': ('aio.html#debounce.cancel', 'fastcore/aio.py'),
+                              'fastcore.aio.Debounce.flush': ('aio.html#debounce.flush', 'fastcore/aio.py'),
                               'fastcore.aio._get_loop': ('aio.html#_get_loop', 'fastcore/aio.py'),
                               'fastcore.aio.acache': ('aio.html#acache', 'fastcore/aio.py'),
                               'fastcore.aio.ctx_sync': ('aio.html#ctx_sync', 'fastcore/aio.py'),
+                              'fastcore.aio.debounced': ('aio.html#debounced', 'fastcore/aio.py'),
                               'fastcore.aio.is_async_callable': ('aio.html#is_async_callable', 'fastcore/aio.py'),
                               'fastcore.aio.iter_sync': ('aio.html#iter_sync', 'fastcore/aio.py'),
                               'fastcore.aio.mapa': ('aio.html#mapa', 'fastcore/aio.py'),
@@ -20,6 +31,7 @@
                               'fastcore.aio.reawaitable': ('aio.html#reawaitable', 'fastcore/aio.py'),
                               'fastcore.aio.run_sync': ('aio.html#run_sync', 'fastcore/aio.py'),
                               'fastcore.aio.then': ('aio.html#then', 'fastcore/aio.py'),
+                              'fastcore.aio.throttled': ('aio.html#throttled', 'fastcore/aio.py'),
                               'fastcore.aio.to_aiter': ('aio.html#to_aiter', 'fastcore/aio.py')},
             'fastcore.all': {},
             'fastcore.ansi': {},

fastcore/aio.py

@@ -1,17 +1,17 @@
-"""Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then`
+"""Bridging async and sync code: `run_sync`, `iter_sync`, `ctx_sync`, `maybe_await`, and `then`, plus `Debounce` for coalescing bursts of calls
 
 Docs: https://fastcore.fast.ai/aio.html.md"""
 
 # AUTOGENERATED! DO NOT EDIT! File to edit: ../nbs/03c_aio.ipynb.
 
 # %% auto #0
 __all__ = ['run_sync', 'iter_sync', 'ctx_sync', 'maybe_await', 'then', 'acache', 'CachedAwaitable', 'reawaitable',
-           'is_async_callable', 'to_aiter', 'maybe_aiter', 'mapa', 'noopa']
+           'is_async_callable', 'to_aiter', 'maybe_aiter', 'mapa', 'noopa', 'Debounce', 'debounced', 'throttled']
 
 # %% ../nbs/03c_aio.ipynb #7e2193be
 import asyncio,threading
 from contextlib import contextmanager
-from functools import wraps
+from functools import wraps,update_wrapper
 from .imports import *
 from .xtras import UNSET
 from .basics import *
@@ -136,3 +136,77 @@ async def mapa(f, items):
 async def noopa(x=None, *args, **kwargs):
     "Do nothing (async)"
     return x
+
+# %% ../nbs/03c_aio.ipynb #314d1e40
+class Debounce:
+    "Coalesce calls to `f` into one, made `wait` secs after calls stop (or `max_wait` secs after they start)"
+    def __init__(self, f, wait, max_wait=None, leading=False, trailing=True):
+        assert leading or trailing,"one of `leading`/`trailing` must be set"
+        store_attr()
+        update_wrapper(self, f, updated=())
+        self._task,self._q,self._name = None,None,getattr(f, '__name__', None)
+        try: self._loop = asyncio.get_running_loop()
+        except RuntimeError: self._loop = None
+
+    def _put_latest(self, q, call):
+        if q.full(): q.get_nowait()
+        q.put_nowait(call)
+
+    def __call__(self, *args, **kw):
+        try: loop = asyncio.get_running_loop()
+        except RuntimeError:
+            if self._loop is None: raise RuntimeError('Debounce must first be called from an async event loop') from None
+            return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw))
+
+        if self._loop is None: self._loop = loop
+        elif loop is not self._loop: return self._loop.call_soon_threadsafe(partial(self.__call__, *args, **kw))
+
+        self._deadline = self._loop.time()+self.wait
+        call = (args, kw)
+        if self._task and not self._task.done(): self._put_latest(self._q, call)
+        else:
+            self._q = asyncio.Queue(maxsize=1)
+            self._cap = self._loop.time()+self.max_wait if self.max_wait else float('inf')
+            if not self.leading: self._put_latest(self._q, call)
+            self._task = self._loop.create_task(self._run(self._q, call if self.leading else None))
+
+    async def _fire(self, p): await maybe_await(self.f(*p[0], **p[1]))
+    async def _run(self, q, first):
+        if first is not None: await self._fire(first)
+
+        while True:
+            delay = min(self._deadline,self._cap) - self._loop.time()
+            if delay>0: await asyncio.sleep(delay); continue
+            if not self.trailing:
+                if not q.empty(): q.get_nowait()  # drop the rest of the burst
+                return
+            if q.empty(): return
+            await self._fire(q.get_nowait())
+            if q.empty(): return
+            if self.max_wait: self._cap = self._loop.time()+self.max_wait
+
+    def cancel(self):
+        "Discard any pending call"
+        if self._task and not self._task.done(): self._task.cancel()
+        self._task,self._q = None,None
+
+    async def flush(self):
+        "Run any pending call now instead of waiting"
+        p = self._q.get_nowait() if self._q and not self._q.empty() else None
+        self.cancel()
+        if p is not None: await self._fire(p)
+
+    def __set_name__(self, owner, name): self._name = name
+    def __get__(self, obj, objtype=None):
+        if obj is None: return self
+        res = obj.__dict__[self._name] = Debounce(partial(self.f, obj), self.wait, self.max_wait, self.leading, self.trailing)
+        return res
+
+# %% ../nbs/03c_aio.ipynb #7d9f6bcf
+def debounced(wait, max_wait=None, leading=False, trailing=True):
+    "Decorator: `Debounce` a function or method"
+    return partial(Debounce, wait=wait, max_wait=max_wait, leading=leading, trailing=trailing)
+
+def throttled(wait, leading=False):
+    "Decorator: fire at most once per `wait` secs (`Debounce` with `max_wait=wait`)"
+    return debounced(wait, max_wait=wait, leading=leading)

fastcore/basics.py

@@ -1135,7 +1135,8 @@ def _inner(f):
             for o in functools.WRAPPER_ASSIGNMENTS: setattr(nf, o, getattr(f,o))
             nf.__name__ = _nm
             nf.__qualname__ = f"{c_.__name__}.{_nm}"
-            if hasattr(nf.__code__, 'co_qualname'): nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)
+            if hasattr(nf, '__code__') and hasattr(nf.__code__, 'co_qualname'):
+                nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)
             if hasattr(c_, _nm) and not hasattr(c_, onm): setattr(c_, onm, getattr(c_, _nm))
             if cls_method:    attr = _clsmethod(nf)
             elif static_method:  attr = staticmethod(nf)

nbs/01_basics.ipynb

@@ -6529,7 +6529,8 @@
     "            for o in functools.WRAPPER_ASSIGNMENTS: setattr(nf, o, getattr(f,o))\n",
     "            nf.__name__ = _nm\n",
     "            nf.__qualname__ = f\"{c_.__name__}.{_nm}\"\n",
-    "            if hasattr(nf.__code__, 'co_qualname'): nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)\n",
+    "            if hasattr(nf, '__code__') and hasattr(nf.__code__, 'co_qualname'):\n",
+    "                nf.__code__ = nf.__code__.replace(co_qualname=nf.__qualname__)\n",
     "            if hasattr(c_, _nm) and not hasattr(c_, onm): setattr(c_, onm, getattr(c_, _nm))\n",
     "            if cls_method:    attr = _clsmethod(nf)\n",
     "            elif static_method:  attr = staticmethod(nf)\n",
@@ -8582,6 +8583,7 @@
  ],
  "metadata": {
   "solveit": {
+   "autosave": false,
    "default_code": false,
    "mode": "learning",
    "use_fence": false,