Clock

clock

Default Clock implementation built on the running asyncio loop.

Entities and event dispatchers schedule background coroutines through a Clock. Production code uses AsyncioClock; tests can substitute a no-op or recording implementation.

AsyncioClock

Schedule coroutines onto the currently-running asyncio loop.

Source code in src/haclient/core/clock.py

class AsyncioClock:
    """Schedule coroutines onto the currently-running asyncio loop."""

    def loop(self) -> asyncio.AbstractEventLoop | None:
        """Return the running event loop, if any.

        Returns
        -------
        asyncio.AbstractEventLoop or None
            The active loop, or ``None`` if called outside one.
        """
        try:
            return asyncio.get_running_loop()
        except RuntimeError:
            return None

    def schedule(self, coro: Awaitable[Any]) -> None:
        """Schedule *coro* on the running loop without blocking.

        Parameters
        ----------
        coro : Awaitable
            The coroutine or awaitable to run.
        """
        loop = self.loop()
        if loop is not None and loop.is_running():
            loop.create_task(_await_and_log(coro))
        else:  # pragma: no cover - only reached without a running loop
            asyncio.ensure_future(coro)

loop

loop() -> asyncio.AbstractEventLoop | None

Return the running event loop, if any.

Returns:

Type	Description
AbstractEventLoop or None	The active loop, or None if called outside one.

Source code in src/haclient/core/clock.py

def loop(self) -> asyncio.AbstractEventLoop | None:
    """Return the running event loop, if any.

    Returns
    -------
    asyncio.AbstractEventLoop or None
        The active loop, or ``None`` if called outside one.
    """
    try:
        return asyncio.get_running_loop()
    except RuntimeError:
        return None

schedule

schedule(coro: Awaitable[Any]) -> None

Schedule coro on the running loop without blocking.

Parameters:

Name	Type	Description	Default
coro	Awaitable	The coroutine or awaitable to run.	required

Source code in src/haclient/core/clock.py

def schedule(self, coro: Awaitable[Any]) -> None:
    """Schedule *coro* on the running loop without blocking.

    Parameters
    ----------
    coro : Awaitable
        The coroutine or awaitable to run.
    """
    loop = self.loop()
    if loop is not None and loop.is_running():
        loop.create_task(_await_and_log(coro))
    else:  # pragma: no cover - only reached without a running loop
        asyncio.ensure_future(coro)