Index

busylight_core.mixins

Mixin classes for extending Light functionality.

Classes

busylight_core.mixins.ColorableMixin

Mixin providing individual RGB color component access and manipulation.

Extends Light classes with individual red, green, and blue properties alongside the standard color tuple property. Use this mixin when you need fine-grained control over individual color components or when building device state objects that store colors as separate fields.

Attributes

busylight_core.mixins.ColorableMixin.red class-attribute instance-attribute

red = property(
    lambda self: getattr(self, "_red", 0),
    lambda self, value: setattr(self, "_red", value),
    doc="Red color value.",
)

busylight_core.mixins.ColorableMixin.green class-attribute instance-attribute

green = property(
    lambda self: getattr(self, "_green", 0),
    lambda self, value: setattr(self, "_green", value),
    doc="Green color value.",
)

busylight_core.mixins.ColorableMixin.blue class-attribute instance-attribute

blue = property(
    lambda self: getattr(self, "_blue", 0),
    lambda self, value: setattr(self, "_blue", value),
    doc="Blue color value.",
)

busylight_core.mixins.ColorableMixin.color property writable

color

Get the current RGB color as a tuple.

Returns the combined RGB color values as a 3-tuple. Use this property to read the complete color state when you need all three components together, such as for device updates or color comparisons.

:return: RGB color values as (red, green, blue) tuple with 0-255 range

busylight_core.mixins.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.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.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.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.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.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.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.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.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.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)