Taskable Mixin

busylight_core.mixins.taskable

Asynchronous task support for animating lights.

Attributes

busylight_core.mixins.taskable.TaskFunction module-attribute

TaskFunction = (
    Callable[["TaskableMixin"], None]
    | Callable[["TaskableMixin"], Awaitable[None]]
)

busylight_core.mixins.taskable.TaskInstantiator module-attribute

TaskInstantiator = Callable[
    [str, TaskFunction, float | None], Task | Timer
]

Classes

busylight_core.mixins.taskable.TaskPriority

Bases: IntEnum

Task priority levels for scheduling.

Used to control task execution priority and enable priority-based cancellation of tasks.

Attributes

busylight_core.mixins.taskable.TaskPriority.LOW class-attribute instance-attribute

LOW = 1

busylight_core.mixins.taskable.TaskPriority.NORMAL class-attribute instance-attribute

NORMAL = 2

busylight_core.mixins.taskable.TaskPriority.HIGH class-attribute instance-attribute

HIGH = 3

busylight_core.mixins.taskable.TaskPriority.CRITICAL class-attribute instance-attribute

CRITICAL = 4

busylight_core.mixins.taskable.TaskInfo dataclass

TaskInfo(task, priority, name, created_at)

Information about a managed task.

Contains task metadata including priority, creation time, and status information for enhanced task monitoring and debugging.

Attributes

busylight_core.mixins.taskable.TaskInfo.task instance-attribute

task

busylight_core.mixins.taskable.TaskInfo.priority instance-attribute

priority

busylight_core.mixins.taskable.TaskInfo.name instance-attribute

name

busylight_core.mixins.taskable.TaskInfo.created_at instance-attribute

created_at

busylight_core.mixins.taskable.TaskInfo.is_running property

is_running

Check if task is currently running.

Returns True if the task has not completed, been cancelled, or failed.

busylight_core.mixins.taskable.TaskInfo.is_cancelled property

is_cancelled

Check if task was cancelled.

Returns True if the task was explicitly cancelled before completion.

busylight_core.mixins.taskable.TaskInfo.has_exception property

has_exception

Check if task completed with an exception.

Returns True if the task completed but raised an unhandled exception.

busylight_core.mixins.taskable.TaskInfo.exception property

exception

Get task exception if any.

Returns the exception that caused the task to fail, or None if the task completed successfully or is still running.

busylight_core.mixins.taskable.TaskableMixin

TaskableMixin(*args, **kwargs)

Associate and manage asynchronous and synchronous tasks.

Provides enhanced task management with automatic strategy selection (asyncio vs threading), prioritization, error handling, and task monitoring capabilities for Light instances.

The environment automatically determines whether to use asyncio tasks or threading timers based on the presence of a running event loop.

Initialize TaskableMixin with task storage for both strategies.

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

def __init__(self, *args: object, **kwargs: object) -> None:
    """Initialize TaskableMixin with task storage for both strategies."""
    super().__init__(*args, **kwargs)
    # Initialize task storage for both strategies
    self._thread_tasks: dict[str, threading.Timer] = {}
    self._task_lock = threading.Lock()

Attributes

busylight_core.mixins.taskable.TaskableMixin.event_loop cached property

event_loop

The default event loop.

Returns the currently running event loop, or creates a new one if no event loop is currently running.

busylight_core.mixins.taskable.TaskableMixin.tasks cached property

tasks

Active asyncio tasks that are associated with this instance.

Dictionary mapping task names to their corresponding asyncio.Task objects.

busylight_core.mixins.taskable.TaskableMixin.task_info cached property

task_info

Enhanced task information with priority and status tracking.

Dictionary mapping task names to TaskInfo objects containing metadata about priority, creation time, and current status.

busylight_core.mixins.taskable.TaskableMixin.task_strategy cached property

task_strategy

Return the appropriate task instantiation function based on environment.

Automatically detects if we're in an asyncio context and returns the corresponding task creation function. Both functions have identical call signatures for transparent usage.

:return: Function to create tasks (asyncio or threading based)

Functions

busylight_core.mixins.taskable.TaskableMixin.add_task

add_task(
    name,
    func,
    priority=NORMAL,
    replace=False,
    interval=None,
)

Create a task using environment-appropriate strategy.

The environment (asyncio vs non-asyncio) automatically determines whether to use asyncio tasks or threading timers. Both strategies support the same function signatures and periodic execution.

:param name: Unique identifier for the task :param func: Function to execute (sync or async) :param priority: Task priority (used for asyncio tasks only) :param replace: Whether to replace existing task with same name :param interval: For periodic tasks, repeat interval in seconds :return: Created asyncio.Task or threading.Timer

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

def add_task(
    self,
    name: str,
    func: TaskFunction,
    priority: TaskPriority = TaskPriority.NORMAL,
    replace: bool = False,
    interval: float | None = None,
) -> asyncio.Task | threading.Timer:
    """Create a task using environment-appropriate strategy.

    The environment (asyncio vs non-asyncio) automatically determines whether
    to use asyncio tasks or threading timers. Both strategies support the same
    function signatures and periodic execution.

    :param name: Unique identifier for the task
    :param func: Function to execute (sync or async)
    :param priority: Task priority (used for asyncio tasks only)
    :param replace: Whether to replace existing task with same name
    :param interval: For periodic tasks, repeat interval in seconds
    :return: Created asyncio.Task or threading.Timer
    """
    # Clean up existing task if replacing
    if replace:
        self.cancel_task(name)
    elif self._task_exists(name):
        return self._get_existing_task(name)

    # Backward compatibility: handle coroutine functions that expect to be called
    # with arguments in the old style (coroutine(self) instead of func(self))
    if interval is None and asyncio.iscoroutinefunction(func):
        # Old-style coroutine that should be called directly
        return self._create_asyncio_task_legacy(name, func, priority)

    # Use environment-determined strategy
    return self.task_strategy(name, func, interval)

busylight_core.mixins.taskable.TaskableMixin.cancel_task

cancel_task(name)

Cancel task regardless of which strategy created it.

:param name: Name of task to cancel :return: The cancelled task/timer or None if not found

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

def cancel_task(self, name: str) -> asyncio.Task | threading.Timer | None:
    """Cancel task regardless of which strategy created it.

    :param name: Name of task to cancel
    :return: The cancelled task/timer or None if not found
    """
    # Try asyncio first
    if name in self.tasks:
        return self._cancel_asyncio_task(name)

    # Ensure _thread_tasks is initialized
    if not hasattr(self, "_thread_tasks"):
        self._thread_tasks = {}
        self._task_lock = threading.Lock()

    # Try threading
    if name in self._thread_tasks:
        return self._cancel_thread_task(name)

    return None

busylight_core.mixins.taskable.TaskableMixin.cancel_tasks

cancel_tasks(priority=None)

Cancel all tasks or tasks of specific priority.

Cancels either all tasks or only tasks matching the specified priority level.

:param priority: If specified, only cancel tasks of this priority level

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

def cancel_tasks(self, priority: TaskPriority | None = None) -> None:
    """Cancel all tasks or tasks of specific priority.

    Cancels either all tasks or only tasks matching the specified priority level.

    :param priority: If specified, only cancel tasks of this priority level
    """
    if priority is None:
        # Cancel all asyncio tasks (check if initialized)
        if hasattr(self, "_tasks") or "tasks" in self.__dict__:
            for task in self.tasks.values():
                task.cancel()
            self.tasks.clear()

        if hasattr(self, "_task_info") or "task_info" in self.__dict__:
            self.task_info.clear()

        # Cancel all threading tasks
        if hasattr(self, "_thread_tasks"):
            with self._task_lock:
                for timer in self._thread_tasks.values():
                    timer.cancel()
                self._thread_tasks.clear()
    # Priority-based cancellation for asyncio (existing)
    elif hasattr(self, "_task_info") or "task_info" in self.__dict__:
        to_cancel = [
            name
            for name, task_info in self.task_info.items()
            if task_info.priority == priority and task_info.is_running
        ]
        for name in to_cancel:
            self.task_info[name].task.cancel()
        self._cleanup_completed_tasks()

busylight_core.mixins.taskable.TaskableMixin.get_task_status

get_task_status(name)

Get detailed status information for a task.

Returns comprehensive status information including running state, exceptions, priority, and creation time.

:param name: Name of task to check :return: Dictionary with task status details or None if not found

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

def get_task_status(self, name: str) -> dict[str, Any] | None:
    """Get detailed status information for a task.

    Returns comprehensive status information including running state,
    exceptions, priority, and creation time.

    :param name: Name of task to check
    :return: Dictionary with task status details or None if not found
    """
    task_info = self.task_info.get(name)
    if not task_info:
        task = self.tasks.get(name)
        if task:
            return {
                "name": name,
                "running": not task.done(),
                "cancelled": task.cancelled(),
                "has_exception": task.done()
                and not task.cancelled()
                and task.exception() is not None,
                "exception": task.exception() if task.done() else None,
                "priority": "unknown",
                "created_at": "unknown",
            }
        return None

    return {
        "name": task_info.name,
        "running": task_info.is_running,
        "cancelled": task_info.is_cancelled,
        "has_exception": task_info.has_exception,
        "exception": task_info.exception,
        "priority": task_info.priority.name,
        "created_at": task_info.created_at,
    }

busylight_core.mixins.taskable.TaskableMixin.list_active_tasks

list_active_tasks()

Get list of currently active task names.

Returns sorted list of task names that are currently running.

:return: List of task names that are currently running

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

def list_active_tasks(self) -> list[str]:
    """Get list of currently active task names.

    Returns sorted list of task names that are currently running.

    :return: List of task names that are currently running
    """
    active = []

    # Check asyncio tasks
    for name, task_info in self.task_info.items():
        if task_info.is_running:
            active.append(name)

    for name, task in self.tasks.items():
        if name not in self.task_info and not task.done():
            active.append(name)

    # Check threading tasks
    with self._task_lock:
        for name, timer in self._thread_tasks.items():
            if timer.is_alive():
                active.append(name)

    return sorted(active)