Module `livekit.agents.utils.aio.debounce`

Functions

def debounced(delay: float) ‑> Callable[[Callable[[], Coroutine[Any, Any, ~T]]], Debounced[~T]]

Expand source code

def debounced(delay: float) -> Callable[[Callable[[], Coroutine[Any, Any, T]]], Debounced[T]]:
    def decorator(func: Callable[[], Coroutine[Any, Any, T]]) -> Debounced[T]:
        return Debounced(func, delay)

    return decorator

Classes

class Debounced (func: Callable[[], Coroutine[Any, Any, ~T]], delay: float)

Expand source code

class Debounced(Generic[T]):
    def __init__(self, func: Callable[[], Coroutine[Any, Any, T]], delay: float) -> None:
        self._func = func
        self._delay = delay
        self._task: Optional[asyncio.Task[T]] = None

    def schedule(self) -> asyncio.Task[T]:
        self.cancel()

        async def _func_with_timer() -> T:
            await asyncio.sleep(self._delay)
            return await self._func()

        self._task = asyncio.create_task(_func_with_timer())
        return self._task

    def run(self) -> asyncio.Task[T]:
        self.cancel()

        self._task = asyncio.create_task(self._func())
        return self._task

    def cancel(self) -> None:
        if self._task is not None and not self._task.done():
            self._task.cancel()
            self._task = None

    def is_running(self) -> bool:
        return self._task is not None and not self._task.done() and not self._task.cancelled()

    def __call__(self) -> asyncio.Task[T]:
        return self.run()

Abstract base class for generic types.

On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::

class Mapping[KT, VT]:
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

On older versions of Python, however, generic classes have to explicitly inherit from Generic.

After a class has been declared to be generic, it can then be used as follows::

def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Ancestors

typing.Generic

Methods

def cancel(self) ‑> None

Expand source code

def cancel(self) -> None:
    if self._task is not None and not self._task.done():
        self._task.cancel()
        self._task = None

def is_running(self) ‑> bool

Expand source code

def is_running(self) -> bool:
    return self._task is not None and not self._task.done() and not self._task.cancelled()

def run(self) ‑> _asyncio.Task[~T]

Expand source code

def run(self) -> asyncio.Task[T]:
    self.cancel()

    self._task = asyncio.create_task(self._func())
    return self._task

def schedule(self) ‑> _asyncio.Task[~T]

Expand source code

def schedule(self) -> asyncio.Task[T]:
    self.cancel()

    async def _func_with_timer() -> T:
        await asyncio.sleep(self._delay)
        return await self._func()

    self._task = asyncio.create_task(_func_with_timer())
    return self._task