Taskable

busylight_core.mixins.taskable

Asynchronous task support for animating lights.

Classes

busylight_core.mixins.taskable.TaskableMixin

Associate and manage asynchronous tasks.

Tasks can be added and cancelled.

Attributes

busylight_core.mixins.taskable.TaskableMixin.event_loop cached property

event_loop

The default event loop.

busylight_core.mixins.taskable.TaskableMixin.tasks cached property

tasks

Active tasks that are associated with this class.

Functions

busylight_core.mixins.taskable.TaskableMixin.add_task

add_task(name, coroutine)

Create a new task using coroutine as the body and stash it in the tasks dict.

Using name as a key for the tasks dictionary.

:name: str :coroutine: Awaitable :return: asyncio.Task

Source code in src/busylight_core/mixins/taskable.py

def add_task(self, name: str, coroutine: Awaitable) -> asyncio.Task:
    """Create a new task using coroutine as the body and stash it in the tasks dict.

    Using name as a key for the tasks dictionary.

    :name: str
    :coroutine: Awaitable
    :return: asyncio.Task
    """
    try:
        return self.tasks[name]
    except KeyError:
        pass

    # >py3.7, create_task takes a `name` parameter
    self.tasks[name] = self.event_loop.create_task(coroutine(self))

    return self.tasks[name]

busylight_core.mixins.taskable.TaskableMixin.cancel_task

cancel_task(name)

Cancel a task associated with name if it exists.

If the task exists the cancelled task is returned, otherwise None.

:name: str :return: None | asyncio.Task

Source code in src/busylight_core/mixins/taskable.py

def cancel_task(self, name: str) -> asyncio.Task | None:
    """Cancel a task associated with name if it exists.

    If the task exists the cancelled task is returned, otherwise None.

    :name: str
    :return: None | asyncio.Task
    """
    try:
        task = self.tasks[name]
        del self.tasks[name]
        task.cancel()
    except (KeyError, AttributeError):
        pass
    else:
        return task

    return None

busylight_core.mixins.taskable.TaskableMixin.cancel_tasks

cancel_tasks()

Cancel all tasks and return nothing.

Source code in src/busylight_core/mixins/taskable.py

def cancel_tasks(self) -> None:
    """Cancel all tasks and return nothing."""
    for task in self.tasks.values():
        task.cancel()
    self.tasks.clear()