Skip to content

Index

busylight_core

Support for USB Connected Lights

Developers, adding support for a new device will entail:

  • Optionally create a new vendor package in the vendors directory.
  • Create a new subclass of busylight_core.light.Light.
  • Implement all the missing abstract methods.
  • Make sure the vendor package imports all the new subclasses.
  • Make sure the vendor package appends the new subclasses to all.
  • Import the new subclasses in busylight_core.init.
  • Add the new subclasses to busylight_core.init.all.

Refer to any of the existing vendor packages as an example.

Please note, if the subclasses are not imported here, the abc.ABC.subclasses machinery will not find them and your new lights will not be recognized.

Classes

busylight_core.HardwareUnsupportedError

Bases: _BaseLightError

The hardware supplied is not supported by this class.

busylight_core.InvalidHardwareError

Bases: _BaseHardwareError

The device dictionary is missing required key/value pairs.

busylight_core.LightUnavailableError

Bases: _BaseLightError

Previously accessible light is now not accessible.

busylight_core.NoLightsFoundError

Bases: _BaseLightError

No lights were discovered by Light or a subclass of Light.

busylight_core.Hardware dataclass

Hardware(
    device_type,
    path,
    vendor_id,
    product_id,
    serial_number,
    manufacturer_string,
    product_string=None,
    release_number=None,
    usage=None,
    usage_page=None,
    interface_number=None,
    bus_type=None,
    is_acquired=False,
)

USB Hardware description.

Attributes
busylight_core.Hardware.device_type instance-attribute
device_type
busylight_core.Hardware.path instance-attribute
path
busylight_core.Hardware.vendor_id instance-attribute
vendor_id
busylight_core.Hardware.product_id instance-attribute
product_id
busylight_core.Hardware.serial_number instance-attribute
serial_number
busylight_core.Hardware.manufacturer_string instance-attribute
manufacturer_string
busylight_core.Hardware.product_string class-attribute instance-attribute
product_string = None
busylight_core.Hardware.release_number class-attribute instance-attribute
release_number = None
busylight_core.Hardware.usage class-attribute instance-attribute
usage = None
busylight_core.Hardware.usage_page class-attribute instance-attribute
usage_page = None
busylight_core.Hardware.interface_number class-attribute instance-attribute
interface_number = None
busylight_core.Hardware.bus_type class-attribute instance-attribute
bus_type = None
busylight_core.Hardware.is_acquired class-attribute instance-attribute
is_acquired = False
busylight_core.Hardware.device_id cached property
device_id

A tuple of the vendor and product identifiers.

busylight_core.Hardware.handle cached property
handle

An I/O handle for this hardware device.

Functions
busylight_core.Hardware.enumerate classmethod
enumerate(by_type=ANY)

List of all connected hardware devices.

Source code in src/busylight_core/hardware.py
@classmethod
def enumerate(cls, by_type: ConnectionType = ConnectionType.ANY) -> list[Hardware]:
    """List of all connected hardware devices."""
    hardware_info = []

    match by_type:
        case ConnectionType.ANY:
            for connection_type in ConnectionType:
                if connection_type <= 0:
                    continue
                with contextlib.suppress(NotImplementedError):
                    hardware_info.extend(cls.enumerate(connection_type))
        case ConnectionType.HID:
            hardware_info.extend(
                cls.from_hid(device_dict) for device_dict in hid.enumerate()
            )
        case ConnectionType.SERIAL:
            hardware_info.extend(
                cls.from_portinfo(port_info) for port_info in list_ports.comports()
            )

        case _:
            msg = f"Device type {by_type} not implemented"
            raise NotImplementedError(msg)

    return hardware_info
busylight_core.Hardware.from_portinfo classmethod
from_portinfo(port_info)

Create a Hardware object from a serial port info object.

Source code in src/busylight_core/hardware.py
@classmethod
def from_portinfo(cls, port_info: ListPortInfo) -> Hardware:
    """Create a Hardware object from a serial port info object."""
    return cls(
        device_type=ConnectionType.SERIAL,
        vendor_id=port_info.vid,
        product_id=port_info.pid,
        path=port_info.device.encode("utf-8"),
        serial_number=port_info.serial_number,
        manufacturer_string=port_info.manufacturer,
        product_string=port_info.product,
        bus_type=1,
    )
busylight_core.Hardware.from_hid classmethod
from_hid(device)

Create a Hardware object from a HID dictionary.

Source code in src/busylight_core/hardware.py
@classmethod
def from_hid(cls, device: dict) -> Hardware:
    """Create a Hardware object from a HID dictionary."""
    return cls(device_type=ConnectionType.HID, **device)
busylight_core.Hardware.acquire
acquire()

Open the hardware device.

Source code in src/busylight_core/hardware.py
def acquire(self) -> None:
    """Open the hardware device."""
    if self.is_acquired:
        logger.debug(f"{self} already acquired")
        return

    match self.device_type:
        case ConnectionType.HID:
            self.handle.open_path(self.path)
            self.is_acquired = True
        case ConnectionType.SERIAL:
            self.handle.open()
            self.is_acquired = True
        case _:
            msg = f"{self.device_type.value.title()} hardware not implemented"
            raise NotImplementedError(msg)
busylight_core.Hardware.release
release()

Close the hardware device.

Source code in src/busylight_core/hardware.py
def release(self) -> None:
    """Close the hardware device."""
    if not self.is_acquired:
        logger.debug(f"{self} already released")
        return

    match self.device_type:
        case ConnectionType.HID | ConnectionType.SERIAL:
            self.handle.close()
            self.is_acquired = False
        case _:
            msg = f"{self.device_type.value.title()} hardware not implemented"
            raise NotImplementedError(msg)

busylight_core.Light

Light(hardware, *, reset=False, exclusive=True)

Bases: ABC, ColorableMixin, TaskableMixin

Base class for USB connected lights.

This base class provides a common interface for all USB connected lights.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Light.event_loop cached property
event_loop

The default event loop.

busylight_core.Light.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Light.red property writable
red

Red color value.

busylight_core.Light.green property writable
green

Green color value.

busylight_core.Light.blue property writable
blue

Blue color value.

busylight_core.Light.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Light.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Light.supported_device_ids class-attribute instance-attribute
supported_device_ids = None
busylight_core.Light.hardware instance-attribute
hardware = hardware
busylight_core.Light.path cached property
path

The path to the hardware device.

busylight_core.Light.platform cached property
platform

The discovered operating system platform name.

busylight_core.Light.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Light.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Light.name cached property
name

Return the marketing name of this light.

busylight_core.Light.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Light.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Light.write_strategy property
write_strategy

Return the write method used by this light.

Functions
busylight_core.Light.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.Light.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.Light.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()
busylight_core.Light.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Light.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Light.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Light.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Light.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Light.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Light.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Light.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Light.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Light.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Light.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Light.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Light.on abstractmethod
on(color, led=0)

Activate the light with the given red, green, blue color tuple.

Source code in src/busylight_core/light.py
@abc.abstractmethod
def on(
    self,
    color: tuple[int, int, int],
    led: int = 0,
) -> None:
    """Activate the light with the given red, green, blue color tuple."""
    raise NotImplementedError
busylight_core.Light.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)

busylight_core.BlinkStick

BlinkStick(hardware, *, reset=False, exclusive=True)

Bases: Light

Agile Innovative BlinkStick status light controller.

The BlinkStick is a USB-connected RGB LED device that can be controlled to display various colors and patterns for status indication.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.BlinkStick.event_loop cached property
event_loop

The default event loop.

busylight_core.BlinkStick.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.BlinkStick.red property writable
red

Red color value.

busylight_core.BlinkStick.green property writable
green

Green color value.

busylight_core.BlinkStick.blue property writable
blue

Blue color value.

busylight_core.BlinkStick.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.BlinkStick.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.BlinkStick.hardware instance-attribute
hardware = hardware
busylight_core.BlinkStick.path cached property
path

The path to the hardware device.

busylight_core.BlinkStick.platform cached property
platform

The discovered operating system platform name.

busylight_core.BlinkStick.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.BlinkStick.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.BlinkStick.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.BlinkStick.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.BlinkStick.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.BlinkStick.supported_device_ids class-attribute
supported_device_ids = {(8352, 16869): 'BlinkStick'}
busylight_core.BlinkStick.channel property writable
channel

Get the current channel number for multi-channel BlinkStick devices.

busylight_core.BlinkStick.index property writable
index

Get the current LED index for addressing individual LEDs.

busylight_core.BlinkStick.variant cached property
variant

Get the BlinkStick variant information based on hardware detection.

busylight_core.BlinkStick.name property
name

Get the device name from the variant information.

Functions
busylight_core.BlinkStick.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.BlinkStick.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.BlinkStick.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()
busylight_core.BlinkStick.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.BlinkStick.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.BlinkStick.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.BlinkStick.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.BlinkStick.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.BlinkStick.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.BlinkStick.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.BlinkStick.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.BlinkStick.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.BlinkStick.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.BlinkStick.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.BlinkStick.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.BlinkStick.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/agile_innovative/blinkstick.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "Agile Innovative"
busylight_core.BlinkStick.on
on(color, led=0)

Turn on the BlinkStick with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for BlinkStick)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/agile_innovative/blinkstick.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the BlinkStick with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for BlinkStick)

    """
    with self.batch_update():
        self.color = color

busylight_core.Fit_StatUSB

Fit_StatUSB(hardware, *, reset=False, exclusive=True)

Bases: Light

CompuLab fit-statUSB status light controller.

The fit-statUSB is a USB-connected RGB LED device that communicates using text-based commands for color control.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Fit_StatUSB.event_loop cached property
event_loop

The default event loop.

busylight_core.Fit_StatUSB.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Fit_StatUSB.red property writable
red

Red color value.

busylight_core.Fit_StatUSB.green property writable
green

Green color value.

busylight_core.Fit_StatUSB.blue property writable
blue

Blue color value.

busylight_core.Fit_StatUSB.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Fit_StatUSB.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Fit_StatUSB.hardware instance-attribute
hardware = hardware
busylight_core.Fit_StatUSB.path cached property
path

The path to the hardware device.

busylight_core.Fit_StatUSB.platform cached property
platform

The discovered operating system platform name.

busylight_core.Fit_StatUSB.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Fit_StatUSB.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Fit_StatUSB.name cached property
name

Return the marketing name of this light.

busylight_core.Fit_StatUSB.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Fit_StatUSB.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Fit_StatUSB.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Fit_StatUSB.supported_device_ids class-attribute
supported_device_ids = {(8263, 991): 'fit-statUSB'}
Functions
busylight_core.Fit_StatUSB.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.Fit_StatUSB.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.Fit_StatUSB.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()
busylight_core.Fit_StatUSB.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Fit_StatUSB.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Fit_StatUSB.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Fit_StatUSB.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Fit_StatUSB.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Fit_StatUSB.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Fit_StatUSB.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Fit_StatUSB.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Fit_StatUSB.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Fit_StatUSB.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Fit_StatUSB.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Fit_StatUSB.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Fit_StatUSB.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/compulab/fit_statusb.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "CompuLab"
busylight_core.Fit_StatUSB.on
on(color, led=0)

Turn on the fit-statUSB with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for fit-statUSB)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/compulab/fit_statusb.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the fit-statUSB with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for fit-statUSB)

    """
    with self.batch_update():
        self.color = color

busylight_core.Blynclight

Blynclight(hardware, *, reset=False, exclusive=True)

Bases: Light

Embrava Blynclight status light controller.

The Blynclight is a USB-connected RGB LED device with additional features like sound playback, volume control, and flashing patterns.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Blynclight.event_loop cached property
event_loop

The default event loop.

busylight_core.Blynclight.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Blynclight.red property writable
red

Red color value.

busylight_core.Blynclight.green property writable
green

Green color value.

busylight_core.Blynclight.blue property writable
blue

Blue color value.

busylight_core.Blynclight.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Blynclight.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Blynclight.hardware instance-attribute
hardware = hardware
busylight_core.Blynclight.path cached property
path

The path to the hardware device.

busylight_core.Blynclight.platform cached property
platform

The discovered operating system platform name.

busylight_core.Blynclight.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Blynclight.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Blynclight.name cached property
name

Return the marketing name of this light.

busylight_core.Blynclight.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Blynclight.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Blynclight.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Blynclight.supported_device_ids class-attribute
supported_device_ids = {
    (11277, 1): "Blynclight",
    (11277, 12): "Blynclight",
    (3667, 9494): "Blynclight",
}
busylight_core.Blynclight.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.Blynclight.struct cached property
struct

Get the binary struct formatter for device communication.

Functions
busylight_core.Blynclight.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.Blynclight.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.Blynclight.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()
busylight_core.Blynclight.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Blynclight.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Blynclight.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Blynclight.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Blynclight.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Blynclight.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Blynclight.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Blynclight.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Blynclight.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Blynclight.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Blynclight.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Blynclight.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Blynclight.on
on(color, led=0)

Turn on the Blynclight with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Blynclight)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/embrava/blynclight.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Blynclight with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Blynclight)

    """
    with self.batch_update():
        self.color = color
busylight_core.Blynclight.dim
dim()

Set the light to dim mode (reduced brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def dim(self) -> None:
    """Set the light to dim mode (reduced brightness)."""
    with self.batch_update():
        self.state.dim = True
busylight_core.Blynclight.bright
bright()

Set the light to bright mode (full brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def bright(self) -> None:
    """Set the light to bright mode (full brightness)."""
    with self.batch_update():
        self.state.dim = False
busylight_core.Blynclight.play_sound
play_sound(music=0, volume=1, repeat=False)

Play a sound on the Blynclight device.

PARAMETER DESCRIPTION
music

Sound ID to play (0-9)

TYPE: int DEFAULT: 0

volume

Volume level (1-10)

TYPE: int DEFAULT: 1

repeat

Whether to repeat the sound continuously

TYPE: bool DEFAULT: False

Source code in src/busylight_core/vendors/embrava/blynclight.py
def play_sound(
    self,
    music: int = 0,
    volume: int = 1,
    repeat: bool = False,
) -> None:
    """Play a sound on the Blynclight device.

    Args:
        music: Sound ID to play (0-9)
        volume: Volume level (1-10)
        repeat: Whether to repeat the sound continuously

    """
    with self.batch_update():
        self.state.repeat = repeat
        self.state.play = True
        self.state.music = music
        self.state.mute = False
        self.state.volume = volume
busylight_core.Blynclight.stop_sound
stop_sound()

Stop playing any currently playing sound.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_sound(self) -> None:
    """Stop playing any currently playing sound."""
    with self.batch_update():
        self.state.play = False
busylight_core.Blynclight.mute
mute()

Mute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def mute(self) -> None:
    """Mute the device sound output."""
    with self.batch_update():
        self.state.mute = True
busylight_core.Blynclight.unmute
unmute()

Unmute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def unmute(self) -> None:
    """Unmute the device sound output."""
    with self.batch_update():
        self.state.mute = False
busylight_core.Blynclight.flash
flash(color, speed=None)

Flash the light with the specified color and speed.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

speed

Flash speed (slow, medium, fast) - defaults to slow

TYPE: FlashSpeed DEFAULT: None

Source code in src/busylight_core/vendors/embrava/blynclight.py
def flash(self, color: tuple[int, int, int], speed: FlashSpeed = None) -> None:
    """Flash the light with the specified color and speed.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        speed: Flash speed (slow, medium, fast) - defaults to slow

    """
    speed = speed or FlashSpeed.slow

    with self.batch_update():
        self.color = color
        self.state.flash = True
        self.state.speed = speed.value
busylight_core.Blynclight.stop_flashing
stop_flashing()

Stop the flashing pattern and return to solid color.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_flashing(self) -> None:
    """Stop the flashing pattern and return to solid color."""
    with self.batch_update():
        self.state.flash = False
busylight_core.Blynclight.reset
reset()

Reset the device to its default state (off, no sound).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def reset(self) -> None:
    """Reset the device to its default state (off, no sound)."""
    self.state.reset()
    self.color = (0, 0, 0)
    self.update()

busylight_core.Blynclight_Mini

Blynclight_Mini(hardware, *, reset=False, exclusive=True)

Bases: Blynclight

Embrava Blynclight Mini status light controller.

A smaller version of the Blynclight with the same functionality as the standard Blynclight device.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Blynclight_Mini.event_loop cached property
event_loop

The default event loop.

busylight_core.Blynclight_Mini.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Blynclight_Mini.red property writable
red

Red color value.

busylight_core.Blynclight_Mini.green property writable
green

Green color value.

busylight_core.Blynclight_Mini.blue property writable
blue

Blue color value.

busylight_core.Blynclight_Mini.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Blynclight_Mini.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Blynclight_Mini.hardware instance-attribute
hardware = hardware
busylight_core.Blynclight_Mini.path cached property
path

The path to the hardware device.

busylight_core.Blynclight_Mini.platform cached property
platform

The discovered operating system platform name.

busylight_core.Blynclight_Mini.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Blynclight_Mini.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Blynclight_Mini.name cached property
name

Return the marketing name of this light.

busylight_core.Blynclight_Mini.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Blynclight_Mini.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Blynclight_Mini.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Blynclight_Mini.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.Blynclight_Mini.struct cached property
struct

Get the binary struct formatter for device communication.

busylight_core.Blynclight_Mini.supported_device_ids class-attribute
supported_device_ids = {
    (11277, 10): "Blynclight Mini",
    (3667, 9495): "Blynclight Mini",
}
Functions
busylight_core.Blynclight_Mini.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.Blynclight_Mini.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.Blynclight_Mini.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()
busylight_core.Blynclight_Mini.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Blynclight_Mini.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Blynclight_Mini.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Blynclight_Mini.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Blynclight_Mini.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Blynclight_Mini.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Blynclight_Mini.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Blynclight_Mini.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Blynclight_Mini.reset
reset()

Reset the device to its default state (off, no sound).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def reset(self) -> None:
    """Reset the device to its default state (off, no sound)."""
    self.state.reset()
    self.color = (0, 0, 0)
    self.update()
busylight_core.Blynclight_Mini.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Blynclight_Mini.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Blynclight_Mini.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Blynclight_Mini.on
on(color, led=0)

Turn on the Blynclight with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Blynclight)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/embrava/blynclight.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Blynclight with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Blynclight)

    """
    with self.batch_update():
        self.color = color
busylight_core.Blynclight_Mini.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Blynclight_Mini.dim
dim()

Set the light to dim mode (reduced brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def dim(self) -> None:
    """Set the light to dim mode (reduced brightness)."""
    with self.batch_update():
        self.state.dim = True
busylight_core.Blynclight_Mini.bright
bright()

Set the light to bright mode (full brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def bright(self) -> None:
    """Set the light to bright mode (full brightness)."""
    with self.batch_update():
        self.state.dim = False
busylight_core.Blynclight_Mini.play_sound
play_sound(music=0, volume=1, repeat=False)

Play a sound on the Blynclight device.

PARAMETER DESCRIPTION
music

Sound ID to play (0-9)

TYPE: int DEFAULT: 0

volume

Volume level (1-10)

TYPE: int DEFAULT: 1

repeat

Whether to repeat the sound continuously

TYPE: bool DEFAULT: False

Source code in src/busylight_core/vendors/embrava/blynclight.py
def play_sound(
    self,
    music: int = 0,
    volume: int = 1,
    repeat: bool = False,
) -> None:
    """Play a sound on the Blynclight device.

    Args:
        music: Sound ID to play (0-9)
        volume: Volume level (1-10)
        repeat: Whether to repeat the sound continuously

    """
    with self.batch_update():
        self.state.repeat = repeat
        self.state.play = True
        self.state.music = music
        self.state.mute = False
        self.state.volume = volume
busylight_core.Blynclight_Mini.stop_sound
stop_sound()

Stop playing any currently playing sound.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_sound(self) -> None:
    """Stop playing any currently playing sound."""
    with self.batch_update():
        self.state.play = False
busylight_core.Blynclight_Mini.mute
mute()

Mute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def mute(self) -> None:
    """Mute the device sound output."""
    with self.batch_update():
        self.state.mute = True
busylight_core.Blynclight_Mini.unmute
unmute()

Unmute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def unmute(self) -> None:
    """Unmute the device sound output."""
    with self.batch_update():
        self.state.mute = False
busylight_core.Blynclight_Mini.flash
flash(color, speed=None)

Flash the light with the specified color and speed.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

speed

Flash speed (slow, medium, fast) - defaults to slow

TYPE: FlashSpeed DEFAULT: None

Source code in src/busylight_core/vendors/embrava/blynclight.py
def flash(self, color: tuple[int, int, int], speed: FlashSpeed = None) -> None:
    """Flash the light with the specified color and speed.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        speed: Flash speed (slow, medium, fast) - defaults to slow

    """
    speed = speed or FlashSpeed.slow

    with self.batch_update():
        self.color = color
        self.state.flash = True
        self.state.speed = speed.value
busylight_core.Blynclight_Mini.stop_flashing
stop_flashing()

Stop the flashing pattern and return to solid color.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_flashing(self) -> None:
    """Stop the flashing pattern and return to solid color."""
    with self.batch_update():
        self.state.flash = False

busylight_core.Blynclight_Plus

Blynclight_Plus(hardware, *, reset=False, exclusive=True)

Bases: Blynclight

Embrava Blynclight Plus status light controller.

An enhanced version of the Blynclight with additional features while maintaining the same basic functionality.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Blynclight_Plus.event_loop cached property
event_loop

The default event loop.

busylight_core.Blynclight_Plus.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Blynclight_Plus.red property writable
red

Red color value.

busylight_core.Blynclight_Plus.green property writable
green

Green color value.

busylight_core.Blynclight_Plus.blue property writable
blue

Blue color value.

busylight_core.Blynclight_Plus.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Blynclight_Plus.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Blynclight_Plus.hardware instance-attribute
hardware = hardware
busylight_core.Blynclight_Plus.path cached property
path

The path to the hardware device.

busylight_core.Blynclight_Plus.platform cached property
platform

The discovered operating system platform name.

busylight_core.Blynclight_Plus.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Blynclight_Plus.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Blynclight_Plus.name cached property
name

Return the marketing name of this light.

busylight_core.Blynclight_Plus.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Blynclight_Plus.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Blynclight_Plus.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Blynclight_Plus.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.Blynclight_Plus.struct cached property
struct

Get the binary struct formatter for device communication.

busylight_core.Blynclight_Plus.supported_device_ids class-attribute
supported_device_ids = {
    (11277, 2): "Blynclight Plus",
    (11277, 16): "Blynclight Plus",
}
Functions
busylight_core.Blynclight_Plus.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.Blynclight_Plus.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.Blynclight_Plus.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()
busylight_core.Blynclight_Plus.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Blynclight_Plus.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Blynclight_Plus.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Blynclight_Plus.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Blynclight_Plus.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Blynclight_Plus.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Blynclight_Plus.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Blynclight_Plus.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Blynclight_Plus.reset
reset()

Reset the device to its default state (off, no sound).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def reset(self) -> None:
    """Reset the device to its default state (off, no sound)."""
    self.state.reset()
    self.color = (0, 0, 0)
    self.update()
busylight_core.Blynclight_Plus.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Blynclight_Plus.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Blynclight_Plus.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Blynclight_Plus.on
on(color, led=0)

Turn on the Blynclight with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Blynclight)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/embrava/blynclight.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Blynclight with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Blynclight)

    """
    with self.batch_update():
        self.color = color
busylight_core.Blynclight_Plus.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Blynclight_Plus.dim
dim()

Set the light to dim mode (reduced brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def dim(self) -> None:
    """Set the light to dim mode (reduced brightness)."""
    with self.batch_update():
        self.state.dim = True
busylight_core.Blynclight_Plus.bright
bright()

Set the light to bright mode (full brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def bright(self) -> None:
    """Set the light to bright mode (full brightness)."""
    with self.batch_update():
        self.state.dim = False
busylight_core.Blynclight_Plus.play_sound
play_sound(music=0, volume=1, repeat=False)

Play a sound on the Blynclight device.

PARAMETER DESCRIPTION
music

Sound ID to play (0-9)

TYPE: int DEFAULT: 0

volume

Volume level (1-10)

TYPE: int DEFAULT: 1

repeat

Whether to repeat the sound continuously

TYPE: bool DEFAULT: False

Source code in src/busylight_core/vendors/embrava/blynclight.py
def play_sound(
    self,
    music: int = 0,
    volume: int = 1,
    repeat: bool = False,
) -> None:
    """Play a sound on the Blynclight device.

    Args:
        music: Sound ID to play (0-9)
        volume: Volume level (1-10)
        repeat: Whether to repeat the sound continuously

    """
    with self.batch_update():
        self.state.repeat = repeat
        self.state.play = True
        self.state.music = music
        self.state.mute = False
        self.state.volume = volume
busylight_core.Blynclight_Plus.stop_sound
stop_sound()

Stop playing any currently playing sound.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_sound(self) -> None:
    """Stop playing any currently playing sound."""
    with self.batch_update():
        self.state.play = False
busylight_core.Blynclight_Plus.mute
mute()

Mute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def mute(self) -> None:
    """Mute the device sound output."""
    with self.batch_update():
        self.state.mute = True
busylight_core.Blynclight_Plus.unmute
unmute()

Unmute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def unmute(self) -> None:
    """Unmute the device sound output."""
    with self.batch_update():
        self.state.mute = False
busylight_core.Blynclight_Plus.flash
flash(color, speed=None)

Flash the light with the specified color and speed.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

speed

Flash speed (slow, medium, fast) - defaults to slow

TYPE: FlashSpeed DEFAULT: None

Source code in src/busylight_core/vendors/embrava/blynclight.py
def flash(self, color: tuple[int, int, int], speed: FlashSpeed = None) -> None:
    """Flash the light with the specified color and speed.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        speed: Flash speed (slow, medium, fast) - defaults to slow

    """
    speed = speed or FlashSpeed.slow

    with self.batch_update():
        self.color = color
        self.state.flash = True
        self.state.speed = speed.value
busylight_core.Blynclight_Plus.stop_flashing
stop_flashing()

Stop the flashing pattern and return to solid color.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_flashing(self) -> None:
    """Stop the flashing pattern and return to solid color."""
    with self.batch_update():
        self.state.flash = False

busylight_core.Busylight

Busylight(hardware, *, reset=False, exclusive=True)

Bases: Light

EPOS Busylight status light controller.

The EPOS Busylight is a USB-connected RGB LED device that provides status indication with multiple LED control capabilities.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Busylight.event_loop cached property
event_loop

The default event loop.

busylight_core.Busylight.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Busylight.red property writable
red

Red color value.

busylight_core.Busylight.green property writable
green

Green color value.

busylight_core.Busylight.blue property writable
blue

Blue color value.

busylight_core.Busylight.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Busylight.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Busylight.hardware instance-attribute
hardware = hardware
busylight_core.Busylight.path cached property
path

The path to the hardware device.

busylight_core.Busylight.platform cached property
platform

The discovered operating system platform name.

busylight_core.Busylight.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Busylight.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Busylight.name cached property
name

Return the marketing name of this light.

busylight_core.Busylight.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Busylight.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Busylight.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Busylight.supported_device_ids class-attribute
supported_device_ids = {(5013, 116): 'Busylight'}
busylight_core.Busylight.state cached property
state

Get the device state manager for controlling LED patterns.

Functions
busylight_core.Busylight.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.Busylight.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.Busylight.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()
busylight_core.Busylight.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Busylight.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Busylight.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Busylight.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Busylight.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Busylight.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Busylight.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Busylight.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Busylight.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Busylight.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Busylight.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Busylight.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/epos/busylight.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "EPOS"
busylight_core.Busylight.on
on(color, led=0)

Turn on the EPOS Busylight with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index for targeting specific LEDs

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/epos/busylight.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the EPOS Busylight with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index for targeting specific LEDs

    """
    self.color = color
    with self.batch_update():
        self.state.set_color(color, led)
busylight_core.Busylight.reset
reset()

Reset the device to its default state.

Source code in src/busylight_core/vendors/epos/busylight.py
def reset(self) -> None:
    """Reset the device to its default state."""
    self.state.reset()
    super().reset()

busylight_core.Busylight_Alpha

Busylight_Alpha(hardware, *, reset=False, exclusive=True)

Bases: Light

Kuando Busylight Alpha status light controller.

The Busylight Alpha is a USB-connected RGB LED device that requires periodic keepalive messages to maintain its connection state.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Busylight_Alpha.event_loop cached property
event_loop

The default event loop.

busylight_core.Busylight_Alpha.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Busylight_Alpha.red property writable
red

Red color value.

busylight_core.Busylight_Alpha.green property writable
green

Green color value.

busylight_core.Busylight_Alpha.blue property writable
blue

Blue color value.

busylight_core.Busylight_Alpha.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Busylight_Alpha.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Busylight_Alpha.hardware instance-attribute
hardware = hardware
busylight_core.Busylight_Alpha.path cached property
path

The path to the hardware device.

busylight_core.Busylight_Alpha.platform cached property
platform

The discovered operating system platform name.

busylight_core.Busylight_Alpha.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Busylight_Alpha.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Busylight_Alpha.name cached property
name

Return the marketing name of this light.

busylight_core.Busylight_Alpha.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Busylight_Alpha.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Busylight_Alpha.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Busylight_Alpha.supported_device_ids class-attribute
supported_device_ids = {
    (1240, 63560): "Busylight Alpha",
    (10171, 15306): "Busylight Alpha",
    (10171, 15307): "Busylight Alpha",
    (10171, 15310): "Busylight Alpha",
}
busylight_core.Busylight_Alpha.state cached property
state

Get the device state manager for controlling light patterns.

Functions
busylight_core.Busylight_Alpha.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.Busylight_Alpha.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.Busylight_Alpha.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()
busylight_core.Busylight_Alpha.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Busylight_Alpha.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Busylight_Alpha.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Busylight_Alpha.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Busylight_Alpha.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Busylight_Alpha.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Busylight_Alpha.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Busylight_Alpha.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Busylight_Alpha.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Busylight_Alpha.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Busylight_Alpha.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Busylight_Alpha.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Busylight_Alpha.on
on(color, led=0)

Turn on the Busylight Alpha with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Busylight Alpha)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/kuando/busylight_alpha.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Busylight Alpha with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Busylight Alpha)

    """
    self.color = color
    with self.batch_update():
        self.state.steps[0].jump(self.color)
    self.add_task("keepalive", _keepalive)
busylight_core.Busylight_Alpha.off
off(led=0)

Turn off the Busylight Alpha.

PARAMETER DESCRIPTION
led

LED index (unused for Busylight Alpha)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/kuando/busylight_alpha.py
def off(self, led: int = 0) -> None:
    """Turn off the Busylight Alpha.

    Args:
        led: LED index (unused for Busylight Alpha)

    """
    self.color = (0, 0, 0)
    with self.batch_update():
        self.state.steps[0].jump(self.color)
    self.cancel_task("keepalive")

busylight_core.Busylight_Omega

Busylight_Omega(hardware, *, reset=False, exclusive=True)

Bases: Busylight_Alpha

Kuando Busylight Omega status light controller.

The Busylight Omega is an enhanced version of the Busylight Alpha with the same functionality and keepalive requirements.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Busylight_Omega.event_loop cached property
event_loop

The default event loop.

busylight_core.Busylight_Omega.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Busylight_Omega.red property writable
red

Red color value.

busylight_core.Busylight_Omega.green property writable
green

Green color value.

busylight_core.Busylight_Omega.blue property writable
blue

Blue color value.

busylight_core.Busylight_Omega.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Busylight_Omega.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Busylight_Omega.hardware instance-attribute
hardware = hardware
busylight_core.Busylight_Omega.path cached property
path

The path to the hardware device.

busylight_core.Busylight_Omega.platform cached property
platform

The discovered operating system platform name.

busylight_core.Busylight_Omega.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Busylight_Omega.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Busylight_Omega.name cached property
name

Return the marketing name of this light.

busylight_core.Busylight_Omega.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Busylight_Omega.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Busylight_Omega.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Busylight_Omega.state cached property
state

Get the device state manager for controlling light patterns.

busylight_core.Busylight_Omega.supported_device_ids class-attribute
supported_device_ids = {
    (10171, 15309): "Busylight Omega",
    (10171, 15311): "Busylight Omega",
}
Functions
busylight_core.Busylight_Omega.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.Busylight_Omega.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.Busylight_Omega.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()
busylight_core.Busylight_Omega.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Busylight_Omega.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Busylight_Omega.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Busylight_Omega.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Busylight_Omega.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Busylight_Omega.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Busylight_Omega.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Busylight_Omega.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Busylight_Omega.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Busylight_Omega.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Busylight_Omega.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Busylight_Omega.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Busylight_Omega.on
on(color, led=0)

Turn on the Busylight Alpha with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Busylight Alpha)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/kuando/busylight_alpha.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Busylight Alpha with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Busylight Alpha)

    """
    self.color = color
    with self.batch_update():
        self.state.steps[0].jump(self.color)
    self.add_task("keepalive", _keepalive)
busylight_core.Busylight_Omega.off
off(led=0)

Turn off the Busylight Alpha.

PARAMETER DESCRIPTION
led

LED index (unused for Busylight Alpha)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/kuando/busylight_alpha.py
def off(self, led: int = 0) -> None:
    """Turn off the Busylight Alpha.

    Args:
        led: LED index (unused for Busylight Alpha)

    """
    self.color = (0, 0, 0)
    with self.batch_update():
        self.state.steps[0].jump(self.color)
    self.cancel_task("keepalive")

busylight_core.Bluetooth

Bluetooth(hardware, *, reset=False, exclusive=True)

Bases: Flag

Luxafor Bluetooth status light controller.

A Bluetooth-enabled version of the Luxafor Flag with the same functionality as the USB-connected Flag device.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Bluetooth.event_loop cached property
event_loop

The default event loop.

busylight_core.Bluetooth.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Bluetooth.red property writable
red

Red color value.

busylight_core.Bluetooth.green property writable
green

Green color value.

busylight_core.Bluetooth.blue property writable
blue

Blue color value.

busylight_core.Bluetooth.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Bluetooth.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Bluetooth.hardware instance-attribute
hardware = hardware
busylight_core.Bluetooth.path cached property
path

The path to the hardware device.

busylight_core.Bluetooth.platform cached property
platform

The discovered operating system platform name.

busylight_core.Bluetooth.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Bluetooth.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Bluetooth.name cached property
name

Return the marketing name of this light.

busylight_core.Bluetooth.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Bluetooth.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Bluetooth.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Bluetooth.state cached property
state

Get the device state manager for controlling LED patterns.

busylight_core.Bluetooth.supported_device_ids class-attribute
supported_device_ids = {(1240, 62322): 'BT'}
Functions
busylight_core.Bluetooth.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.Bluetooth.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.Bluetooth.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()
busylight_core.Bluetooth.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Bluetooth.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Bluetooth.claims classmethod
claims(hardware)

Check if this class can handle the given hardware device.

PARAMETER DESCRIPTION
hardware

Hardware device to check

TYPE: Hardware

RETURNS DESCRIPTION
bool

True if this class can handle the hardware

Source code in src/busylight_core/vendors/luxafor/flag.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Check if this class can handle the given hardware device.

    Args:
        hardware: Hardware device to check

    Returns:
        True if this class can handle the hardware

    """
    if not super().claims(hardware):
        return False

    try:
        product = hardware.product_string.split()[-1].casefold()
    except (KeyError, IndexError) as error:
        logger.debug(f"problem {error} processing {hardware}")
        return False

    return product in [
        value.casefold() for value in cls.supported_device_ids.values()
    ]
busylight_core.Bluetooth.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Bluetooth.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Bluetooth.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Bluetooth.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Bluetooth.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Bluetooth.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Bluetooth.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Bluetooth.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Bluetooth.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Bluetooth.on
on(color, led=0)

Turn on the Luxafor Flag with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (0 for all LEDs, or specific LED number)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/luxafor/flag.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Luxafor Flag with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (0 for all LEDs, or specific LED number)

    """
    with self.batch_update():
        try:
            self.state.leds = LEDS(led)
        except ValueError:
            self.state.leds = LEDS.All
        self.color = color
busylight_core.Bluetooth.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)

busylight_core.BusyTag

BusyTag(hardware, *, reset=False, exclusive=True)

Bases: Light

BusyTag status light controller.

The BusyTag is a wireless status light that uses command strings for communication and supports various lighting patterns.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.BusyTag.event_loop cached property
event_loop

The default event loop.

busylight_core.BusyTag.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.BusyTag.red property writable
red

Red color value.

busylight_core.BusyTag.green property writable
green

Green color value.

busylight_core.BusyTag.blue property writable
blue

Blue color value.

busylight_core.BusyTag.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.BusyTag.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.BusyTag.hardware instance-attribute
hardware = hardware
busylight_core.BusyTag.path cached property
path

The path to the hardware device.

busylight_core.BusyTag.platform cached property
platform

The discovered operating system platform name.

busylight_core.BusyTag.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.BusyTag.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.BusyTag.name cached property
name

Return the marketing name of this light.

busylight_core.BusyTag.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.BusyTag.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.BusyTag.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.BusyTag.supported_device_ids class-attribute
supported_device_ids = {(12346, 33247): 'Busy Tag'}
busylight_core.BusyTag.command property writable
command

Get the current command string for device communication.

Functions
busylight_core.BusyTag.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.BusyTag.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.BusyTag.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()
busylight_core.BusyTag.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.BusyTag.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.BusyTag.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.BusyTag.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.BusyTag.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.BusyTag.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.BusyTag.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.BusyTag.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.BusyTag.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.BusyTag.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.BusyTag.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.BusyTag.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.BusyTag.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/luxafor/busytag.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "Busy Tag"
busylight_core.BusyTag.on
on(color, led=0)

Turn on the BusyTag with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index for specific LED targeting

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/luxafor/busytag.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the BusyTag with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index for specific LED targeting

    """
    with self.batch_update():
        self.color = color
        self.command = Command.solid_color(color, led)

busylight_core.Flag

Flag(hardware, *, reset=False, exclusive=True)

Bases: Light

Luxafor Flag status light controller.

The Luxafor Flag is a USB-connected RGB LED device with multiple individually controllable LEDs arranged in a flag pattern.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Flag.event_loop cached property
event_loop

The default event loop.

busylight_core.Flag.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Flag.red property writable
red

Red color value.

busylight_core.Flag.green property writable
green

Green color value.

busylight_core.Flag.blue property writable
blue

Blue color value.

busylight_core.Flag.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Flag.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Flag.hardware instance-attribute
hardware = hardware
busylight_core.Flag.path cached property
path

The path to the hardware device.

busylight_core.Flag.platform cached property
platform

The discovered operating system platform name.

busylight_core.Flag.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Flag.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Flag.name cached property
name

Return the marketing name of this light.

busylight_core.Flag.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Flag.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Flag.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Flag.supported_device_ids class-attribute
supported_device_ids = {(1240, 62322): 'Flag'}
busylight_core.Flag.state cached property
state

Get the device state manager for controlling LED patterns.

Functions
busylight_core.Flag.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.Flag.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.Flag.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()
busylight_core.Flag.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Flag.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Flag.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Flag.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Flag.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Flag.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Flag.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Flag.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Flag.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Flag.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Flag.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Flag.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Flag.claims classmethod
claims(hardware)

Check if this class can handle the given hardware device.

PARAMETER DESCRIPTION
hardware

Hardware device to check

TYPE: Hardware

RETURNS DESCRIPTION
bool

True if this class can handle the hardware

Source code in src/busylight_core/vendors/luxafor/flag.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Check if this class can handle the given hardware device.

    Args:
        hardware: Hardware device to check

    Returns:
        True if this class can handle the hardware

    """
    if not super().claims(hardware):
        return False

    try:
        product = hardware.product_string.split()[-1].casefold()
    except (KeyError, IndexError) as error:
        logger.debug(f"problem {error} processing {hardware}")
        return False

    return product in [
        value.casefold() for value in cls.supported_device_ids.values()
    ]
busylight_core.Flag.on
on(color, led=0)

Turn on the Luxafor Flag with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (0 for all LEDs, or specific LED number)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/luxafor/flag.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Luxafor Flag with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (0 for all LEDs, or specific LED number)

    """
    with self.batch_update():
        try:
            self.state.leds = LEDS(led)
        except ValueError:
            self.state.leds = LEDS.All
        self.color = color

busylight_core.Mute

Mute(hardware, *, reset=False, exclusive=True)

Bases: Flag

Luxafor Mute status light controller.

A mute button device with status light functionality, combining the Luxafor Flag features with button input capabilities.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Mute.event_loop cached property
event_loop

The default event loop.

busylight_core.Mute.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Mute.red property writable
red

Red color value.

busylight_core.Mute.green property writable
green

Green color value.

busylight_core.Mute.blue property writable
blue

Blue color value.

busylight_core.Mute.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Mute.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Mute.hardware instance-attribute
hardware = hardware
busylight_core.Mute.path cached property
path

The path to the hardware device.

busylight_core.Mute.platform cached property
platform

The discovered operating system platform name.

busylight_core.Mute.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Mute.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Mute.name cached property
name

Return the marketing name of this light.

busylight_core.Mute.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Mute.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Mute.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Mute.state cached property
state

Get the device state manager for controlling LED patterns.

busylight_core.Mute.supported_device_ids class-attribute
supported_device_ids = {(1240, 62322): 'Mute'}
busylight_core.Mute.is_button property
is_button

Check if this device has button functionality.

RETURNS DESCRIPTION
bool

True, as the Mute device has a button

busylight_core.Mute.button_on property
button_on

Check if the mute button is currently pressed.

RETURNS DESCRIPTION
bool

True if the button is pressed, False otherwise

Functions
busylight_core.Mute.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.Mute.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.Mute.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()
busylight_core.Mute.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Mute.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Mute.claims classmethod
claims(hardware)

Check if this class can handle the given hardware device.

PARAMETER DESCRIPTION
hardware

Hardware device to check

TYPE: Hardware

RETURNS DESCRIPTION
bool

True if this class can handle the hardware

Source code in src/busylight_core/vendors/luxafor/flag.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Check if this class can handle the given hardware device.

    Args:
        hardware: Hardware device to check

    Returns:
        True if this class can handle the hardware

    """
    if not super().claims(hardware):
        return False

    try:
        product = hardware.product_string.split()[-1].casefold()
    except (KeyError, IndexError) as error:
        logger.debug(f"problem {error} processing {hardware}")
        return False

    return product in [
        value.casefold() for value in cls.supported_device_ids.values()
    ]
busylight_core.Mute.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Mute.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Mute.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Mute.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Mute.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Mute.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Mute.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Mute.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Mute.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Mute.on
on(color, led=0)

Turn on the Luxafor Flag with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (0 for all LEDs, or specific LED number)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/luxafor/flag.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Luxafor Flag with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (0 for all LEDs, or specific LED number)

    """
    with self.batch_update():
        try:
            self.state.leds = LEDS(led)
        except ValueError:
            self.state.leds = LEDS.All
        self.color = color
busylight_core.Mute.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)

busylight_core.Orb

Orb(hardware, *, reset=False, exclusive=True)

Bases: Flag

Luxafor Orb status light controller.

A spherical-shaped status light with the same functionality as the Luxafor Flag device.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Orb.event_loop cached property
event_loop

The default event loop.

busylight_core.Orb.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Orb.red property writable
red

Red color value.

busylight_core.Orb.green property writable
green

Green color value.

busylight_core.Orb.blue property writable
blue

Blue color value.

busylight_core.Orb.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Orb.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Orb.hardware instance-attribute
hardware = hardware
busylight_core.Orb.path cached property
path

The path to the hardware device.

busylight_core.Orb.platform cached property
platform

The discovered operating system platform name.

busylight_core.Orb.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Orb.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Orb.name cached property
name

Return the marketing name of this light.

busylight_core.Orb.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Orb.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Orb.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Orb.state cached property
state

Get the device state manager for controlling LED patterns.

busylight_core.Orb.supported_device_ids class-attribute
supported_device_ids = {(1240, 62322): 'Orb'}
Functions
busylight_core.Orb.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.Orb.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.Orb.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()
busylight_core.Orb.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Orb.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Orb.claims classmethod
claims(hardware)

Check if this class can handle the given hardware device.

PARAMETER DESCRIPTION
hardware

Hardware device to check

TYPE: Hardware

RETURNS DESCRIPTION
bool

True if this class can handle the hardware

Source code in src/busylight_core/vendors/luxafor/flag.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Check if this class can handle the given hardware device.

    Args:
        hardware: Hardware device to check

    Returns:
        True if this class can handle the hardware

    """
    if not super().claims(hardware):
        return False

    try:
        product = hardware.product_string.split()[-1].casefold()
    except (KeyError, IndexError) as error:
        logger.debug(f"problem {error} processing {hardware}")
        return False

    return product in [
        value.casefold() for value in cls.supported_device_ids.values()
    ]
busylight_core.Orb.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Orb.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Orb.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Orb.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Orb.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Orb.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Orb.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Orb.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Orb.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Orb.on
on(color, led=0)

Turn on the Luxafor Flag with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (0 for all LEDs, or specific LED number)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/luxafor/flag.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Luxafor Flag with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (0 for all LEDs, or specific LED number)

    """
    with self.batch_update():
        try:
            self.state.leds = LEDS(led)
        except ValueError:
            self.state.leds = LEDS.All
        self.color = color
busylight_core.Orb.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)

busylight_core.MuteMe

MuteMe(hardware, *, reset=False, exclusive=True)

Bases: Light

MuteMe status light and button controller.

The MuteMe is a USB-connected RGB LED device with integrated button functionality for mute control in video conferencing applications.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.MuteMe.event_loop cached property
event_loop

The default event loop.

busylight_core.MuteMe.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.MuteMe.red property writable
red

Red color value.

busylight_core.MuteMe.green property writable
green

Green color value.

busylight_core.MuteMe.blue property writable
blue

Blue color value.

busylight_core.MuteMe.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.MuteMe.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.MuteMe.hardware instance-attribute
hardware = hardware
busylight_core.MuteMe.path cached property
path

The path to the hardware device.

busylight_core.MuteMe.platform cached property
platform

The discovered operating system platform name.

busylight_core.MuteMe.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.MuteMe.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.MuteMe.name cached property
name

Return the marketing name of this light.

busylight_core.MuteMe.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.MuteMe.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.MuteMe.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.MuteMe.supported_device_ids class-attribute
supported_device_ids = {
    (5824, 10203): "MuteMe Original",
    (8352, 17114): "MuteMe Original",
}
busylight_core.MuteMe.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.MuteMe.struct cached property
struct

Get the binary struct formatter for device communication.

busylight_core.MuteMe.is_pluggedin property
is_pluggedin

Check if the device is plugged in and responsive.

RETURNS DESCRIPTION
bool

True if device responds to feature report, False otherwise

busylight_core.MuteMe.is_button property
is_button

Check if this device has button functionality.

RETURNS DESCRIPTION
bool

True, as the MuteMe device has a button

busylight_core.MuteMe.button_on property
button_on

Check if the mute button is currently pressed.

RETURNS DESCRIPTION
bool

True if the button is pressed, False otherwise

RAISES DESCRIPTION
NotImplementedError

Button state reading not implemented

Functions
busylight_core.MuteMe.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.MuteMe.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.MuteMe.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()
busylight_core.MuteMe.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.MuteMe.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.MuteMe.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.MuteMe.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.MuteMe.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.MuteMe.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.MuteMe.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.MuteMe.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.MuteMe.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.MuteMe.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.MuteMe.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.MuteMe.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.MuteMe.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/muteme/muteme.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "MuteMe"
busylight_core.MuteMe.on
on(color, led=0)

Turn on the MuteMe with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for MuteMe)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/muteme/muteme.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the MuteMe with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for MuteMe)

    """
    with self.batch_update():
        self.color = color

busylight_core.MuteMe_Mini

MuteMe_Mini(hardware, *, reset=False, exclusive=True)

Bases: MuteMe

MuteMe Mini status light and button controller.

A smaller version of the MuteMe device with the same button and LED functionality as the original MuteMe.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.MuteMe_Mini.event_loop cached property
event_loop

The default event loop.

busylight_core.MuteMe_Mini.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.MuteMe_Mini.red property writable
red

Red color value.

busylight_core.MuteMe_Mini.green property writable
green

Green color value.

busylight_core.MuteMe_Mini.blue property writable
blue

Blue color value.

busylight_core.MuteMe_Mini.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.MuteMe_Mini.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.MuteMe_Mini.hardware instance-attribute
hardware = hardware
busylight_core.MuteMe_Mini.path cached property
path

The path to the hardware device.

busylight_core.MuteMe_Mini.platform cached property
platform

The discovered operating system platform name.

busylight_core.MuteMe_Mini.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.MuteMe_Mini.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.MuteMe_Mini.name cached property
name

Return the marketing name of this light.

busylight_core.MuteMe_Mini.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.MuteMe_Mini.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.MuteMe_Mini.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.MuteMe_Mini.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.MuteMe_Mini.struct cached property
struct

Get the binary struct formatter for device communication.

busylight_core.MuteMe_Mini.is_pluggedin property
is_pluggedin

Check if the device is plugged in and responsive.

RETURNS DESCRIPTION
bool

True if device responds to feature report, False otherwise

busylight_core.MuteMe_Mini.is_button property
is_button

Check if this device has button functionality.

RETURNS DESCRIPTION
bool

True, as the MuteMe device has a button

busylight_core.MuteMe_Mini.button_on property
button_on

Check if the mute button is currently pressed.

RETURNS DESCRIPTION
bool

True if the button is pressed, False otherwise

RAISES DESCRIPTION
NotImplementedError

Button state reading not implemented

busylight_core.MuteMe_Mini.supported_device_ids class-attribute
supported_device_ids = {(8352, 17115): 'MuteMe Mini'}
Functions
busylight_core.MuteMe_Mini.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.MuteMe_Mini.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.MuteMe_Mini.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()
busylight_core.MuteMe_Mini.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/muteme/muteme.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "MuteMe"
busylight_core.MuteMe_Mini.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.MuteMe_Mini.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.MuteMe_Mini.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.MuteMe_Mini.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.MuteMe_Mini.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.MuteMe_Mini.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.MuteMe_Mini.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.MuteMe_Mini.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.MuteMe_Mini.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.MuteMe_Mini.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.MuteMe_Mini.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.MuteMe_Mini.on
on(color, led=0)

Turn on the MuteMe with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for MuteMe)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/muteme/muteme.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the MuteMe with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for MuteMe)

    """
    with self.batch_update():
        self.color = color
busylight_core.MuteMe_Mini.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)

busylight_core.MuteSync

MuteSync(hardware, *, reset=False, exclusive=True)

Bases: Light

MuteSync status light and button controller.

The MuteSync is a USB-connected device that combines button functionality with status light capabilities for meeting control.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.MuteSync.event_loop cached property
event_loop

The default event loop.

busylight_core.MuteSync.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.MuteSync.red property writable
red

Red color value.

busylight_core.MuteSync.green property writable
green

Green color value.

busylight_core.MuteSync.blue property writable
blue

Blue color value.

busylight_core.MuteSync.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.MuteSync.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.MuteSync.hardware instance-attribute
hardware = hardware
busylight_core.MuteSync.path cached property
path

The path to the hardware device.

busylight_core.MuteSync.platform cached property
platform

The discovered operating system platform name.

busylight_core.MuteSync.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.MuteSync.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.MuteSync.name cached property
name

Return the marketing name of this light.

busylight_core.MuteSync.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.MuteSync.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.MuteSync.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.MuteSync.supported_device_ids class-attribute
supported_device_ids = {(4292, 60000): 'MuteSync Button'}
busylight_core.MuteSync.is_button property
is_button

Check if this device has button functionality.

RETURNS DESCRIPTION
bool

True, as the MuteSync device has a button

busylight_core.MuteSync.button_on property
button_on

Check if the mute button is currently pressed.

RETURNS DESCRIPTION
bool

Always False in current implementation

Functions
busylight_core.MuteSync.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.MuteSync.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.MuteSync.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()
busylight_core.MuteSync.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.MuteSync.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.MuteSync.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.MuteSync.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.MuteSync.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.MuteSync.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.MuteSync.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.MuteSync.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.MuteSync.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.MuteSync.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.MuteSync.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.MuteSync.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/muteme/mutesync.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "MuteSync"
busylight_core.MuteSync.claims classmethod
claims(hardware)

Return True if the hardware describes a MuteSync Button.

Source code in src/busylight_core/vendors/muteme/mutesync.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware describes a MuteSync Button."""
    # Addresses issue #356 where MuteSync claims another hardware with
    # a SiliconLabs CP2102 USB to Serial controller that is not a MuteSync
    # hardware.

    claim = super().claims(hardware)

    vendor = cls.vendor().lower()

    try:
        manufacturer = vendor in hardware.manufacturer_string.lower()
    except AttributeError:
        manufacturer = False

    try:
        product = vendor in hardware.product_string.lower()
    except AttributeError:
        product = False

    return claim and (product or manufacturer)
busylight_core.MuteSync.on
on(color, led=0)

Turn on the MuteSync with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for MuteSync)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/muteme/mutesync.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the MuteSync with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for MuteSync)

    """
    with self.batch_update():
        self.color = color

busylight_core.Status_Indicator

Status_Indicator(hardware, *, reset=False, exclusive=True)

Bases: Blynclight

Plantronics Status Indicator status light controller.

A Plantronics-branded version of the Blynclight device with identical functionality to the Embrava Blynclight.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Status_Indicator.event_loop cached property
event_loop

The default event loop.

busylight_core.Status_Indicator.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Status_Indicator.red property writable
red

Red color value.

busylight_core.Status_Indicator.green property writable
green

Green color value.

busylight_core.Status_Indicator.blue property writable
blue

Blue color value.

busylight_core.Status_Indicator.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Status_Indicator.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Status_Indicator.hardware instance-attribute
hardware = hardware
busylight_core.Status_Indicator.path cached property
path

The path to the hardware device.

busylight_core.Status_Indicator.platform cached property
platform

The discovered operating system platform name.

busylight_core.Status_Indicator.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Status_Indicator.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Status_Indicator.name cached property
name

Return the marketing name of this light.

busylight_core.Status_Indicator.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Status_Indicator.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Status_Indicator.write_strategy property
write_strategy

Return the write method used by this light.

busylight_core.Status_Indicator.state cached property
state

Get the device state manager for controlling light behavior.

busylight_core.Status_Indicator.struct cached property
struct

Get the binary struct formatter for device communication.

busylight_core.Status_Indicator.supported_device_ids class-attribute
supported_device_ids = {(1151, 53253): 'Status Indicator'}
Functions
busylight_core.Status_Indicator.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.Status_Indicator.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.Status_Indicator.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()
busylight_core.Status_Indicator.vendor cached classmethod
vendor()

Return the vendor name in title case.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def vendor(cls) -> str:
    """Return the vendor name in title case."""
    return cls.__module__.split(".")[-2].title()
busylight_core.Status_Indicator.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Status_Indicator.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Status_Indicator.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Status_Indicator.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Status_Indicator.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Status_Indicator.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Status_Indicator.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Status_Indicator.reset
reset()

Reset the device to its default state (off, no sound).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def reset(self) -> None:
    """Reset the device to its default state (off, no sound)."""
    self.state.reset()
    self.color = (0, 0, 0)
    self.update()
busylight_core.Status_Indicator.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Status_Indicator.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Status_Indicator.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Status_Indicator.on
on(color, led=0)

Turn on the Blynclight with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index (unused for Blynclight)

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/embrava/blynclight.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Blynclight with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index (unused for Blynclight)

    """
    with self.batch_update():
        self.color = color
busylight_core.Status_Indicator.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Status_Indicator.dim
dim()

Set the light to dim mode (reduced brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def dim(self) -> None:
    """Set the light to dim mode (reduced brightness)."""
    with self.batch_update():
        self.state.dim = True
busylight_core.Status_Indicator.bright
bright()

Set the light to bright mode (full brightness).

Source code in src/busylight_core/vendors/embrava/blynclight.py
def bright(self) -> None:
    """Set the light to bright mode (full brightness)."""
    with self.batch_update():
        self.state.dim = False
busylight_core.Status_Indicator.play_sound
play_sound(music=0, volume=1, repeat=False)

Play a sound on the Blynclight device.

PARAMETER DESCRIPTION
music

Sound ID to play (0-9)

TYPE: int DEFAULT: 0

volume

Volume level (1-10)

TYPE: int DEFAULT: 1

repeat

Whether to repeat the sound continuously

TYPE: bool DEFAULT: False

Source code in src/busylight_core/vendors/embrava/blynclight.py
def play_sound(
    self,
    music: int = 0,
    volume: int = 1,
    repeat: bool = False,
) -> None:
    """Play a sound on the Blynclight device.

    Args:
        music: Sound ID to play (0-9)
        volume: Volume level (1-10)
        repeat: Whether to repeat the sound continuously

    """
    with self.batch_update():
        self.state.repeat = repeat
        self.state.play = True
        self.state.music = music
        self.state.mute = False
        self.state.volume = volume
busylight_core.Status_Indicator.stop_sound
stop_sound()

Stop playing any currently playing sound.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_sound(self) -> None:
    """Stop playing any currently playing sound."""
    with self.batch_update():
        self.state.play = False
busylight_core.Status_Indicator.mute
mute()

Mute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def mute(self) -> None:
    """Mute the device sound output."""
    with self.batch_update():
        self.state.mute = True
busylight_core.Status_Indicator.unmute
unmute()

Unmute the device sound output.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def unmute(self) -> None:
    """Unmute the device sound output."""
    with self.batch_update():
        self.state.mute = False
busylight_core.Status_Indicator.flash
flash(color, speed=None)

Flash the light with the specified color and speed.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

speed

Flash speed (slow, medium, fast) - defaults to slow

TYPE: FlashSpeed DEFAULT: None

Source code in src/busylight_core/vendors/embrava/blynclight.py
def flash(self, color: tuple[int, int, int], speed: FlashSpeed = None) -> None:
    """Flash the light with the specified color and speed.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        speed: Flash speed (slow, medium, fast) - defaults to slow

    """
    speed = speed or FlashSpeed.slow

    with self.batch_update():
        self.color = color
        self.state.flash = True
        self.state.speed = speed.value
busylight_core.Status_Indicator.stop_flashing
stop_flashing()

Stop the flashing pattern and return to solid color.

Source code in src/busylight_core/vendors/embrava/blynclight.py
def stop_flashing(self) -> None:
    """Stop the flashing pattern and return to solid color."""
    with self.batch_update():
        self.state.flash = False

busylight_core.Blink1

Blink1(hardware, *, reset=False, exclusive=True)

Bases: Light

ThingM Blink(1) status light controller.

The Blink(1) is a USB-connected RGB LED device that uses feature reports for communication and supports various effects.

Initialize a Light with the given hardware information.

:param: reset - bool - reset the hardware to a known state :param: exclusive - bool - acquire exclusive access to the hardware

  • HardwareUnsupportedError: if the given Hardware is not supported by this class.
Source code in src/busylight_core/light.py
def __init__(
    self,
    hardware: Hardware,
    *,
    reset: bool = False,
    exclusive: bool = True,
) -> None:
    """Initialize a Light with the given hardware information.

    :param: reset - bool - reset the hardware to a known state
    :param: exclusive - bool - acquire exclusive access to the hardware

    Raises:
    - HardwareUnsupportedError: if the given Hardware is not supported by this
      class.

    """
    if not self.__class__.claims(hardware):
        raise HardwareUnsupportedError(hardware)

    self.hardware = hardware
    self._reset = reset
    self._exclusive = exclusive

    if exclusive:
        self.hardware.acquire()

    if reset:
        self.reset()
Attributes
busylight_core.Blink1.event_loop cached property
event_loop

The default event loop.

busylight_core.Blink1.tasks cached property
tasks

Active tasks that are associated with this class.

busylight_core.Blink1.red property writable
red

Red color value.

busylight_core.Blink1.green property writable
green

Green color value.

busylight_core.Blink1.blue property writable
blue

Blue color value.

busylight_core.Blink1.color property writable
color

A tuple of red, green, and blue color values.

busylight_core.Blink1.is_lit property
is_lit

True if any color value is greater than 0.

busylight_core.Blink1.hardware instance-attribute
hardware = hardware
busylight_core.Blink1.path cached property
path

The path to the hardware device.

busylight_core.Blink1.platform cached property
platform

The discovered operating system platform name.

busylight_core.Blink1.exclusive property
exclusive

Return True if the light has exclusive access to the hardware.

busylight_core.Blink1.sort_key cached property
sort_key

Return a tuple used for sorting lights.

The tuple consists of: - vendor name in lowercase - device name in lowercase - hardware path

busylight_core.Blink1.name cached property
name

Return the marketing name of this light.

busylight_core.Blink1.hex property
hex

Return the hexadecimal representation of the light's state.

busylight_core.Blink1.read_strategy property
read_strategy

Return the read method used by this light.

busylight_core.Blink1.supported_device_ids class-attribute
supported_device_ids = {(10168, 493): 'Blink(1)'}
busylight_core.Blink1.state cached property
state

Get the device state manager for controlling LED patterns.

busylight_core.Blink1.write_strategy property
write_strategy

Get the write strategy for communicating with the device.

RETURNS DESCRIPTION
Callable

The hardware's send_feature_report method

Functions
busylight_core.Blink1.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.Blink1.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.Blink1.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()
busylight_core.Blink1.unique_device_names cached classmethod
unique_device_names()

Return a list of unique device names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def unique_device_names(cls) -> list[str]:
    """Return a list of unique device names."""
    try:
        return sorted(set(cls.supported_device_ids.values()))
    except AttributeError:
        return []
busylight_core.Blink1.claims classmethod
claims(hardware)

Return True if the hardware is claimed by this class.

Source code in src/busylight_core/light.py
@classmethod
def claims(cls, hardware: Hardware) -> bool:
    """Return True if the hardware is claimed by this class."""
    try:
        return hardware.device_id in cls.supported_device_ids
    except TypeError:
        return False
busylight_core.Blink1.subclasses cached classmethod
subclasses()

Return a list of all subclasses of this class.

Source code in src/busylight_core/light.py
@classmethod
@cache
def subclasses(cls) -> list[type[Light]]:
    """Return a list of all subclasses of this class."""
    subclasses = []

    if cls != Light:
        subclasses.append(cls)

    for subclass in cls.__subclasses__():
        subclasses.extend(subclass.subclasses())

    return sorted(subclasses, key=lambda s: s.__module__)
busylight_core.Blink1.supported_lights cached classmethod
supported_lights()

Return a dictionary of supported lights by vendor.

Keys are vendor names, values are a list of product names.

Source code in src/busylight_core/light.py
@classmethod
@lru_cache(maxsize=1)
def supported_lights(cls) -> dict[str, list[str]]:
    """Return a dictionary of supported lights by vendor.

    Keys are vendor names, values are a list of product names.
    """
    supported_lights: dict[str, list[str]] = {}

    for subclass in cls.subclasses():
        names = supported_lights.setdefault(subclass.vendor(), [])
        names.extend(subclass.unique_device_names())

    return supported_lights
busylight_core.Blink1.available_lights classmethod
available_lights()

Return a dictionary of available hardware by type.

Keys are Light subclasses, values are a list of Hardware instances.

Source code in src/busylight_core/light.py
@classmethod
def available_lights(cls) -> dict[type[Light], list[Hardware]]:
    """Return a dictionary of available hardware by type.

    Keys are Light subclasses, values are a list of Hardware instances.
    """
    available_lights: dict[type[Light], list[Hardware]] = {}

    for hardware in Hardware.enumerate():
        if cls != Light:
            if cls.claims(hardware):
                logger.debug(f"{cls.__name__} claims {hardware}")
                claimed = available_lights.setdefault(cls, [])
                claimed.append(hardware)
        else:
            for subclass in cls.subclasses():
                if subclass.claims(hardware):
                    logger.debug(f"{subclass.__name__} claims {hardware}")
                    claimed = available_lights.setdefault(subclass, [])
                    claimed.append(hardware)

    return available_lights
busylight_core.Blink1.all_lights classmethod
all_lights(*, reset=True, exclusive=True)

Return a list of all lights ready for use.

Source code in src/busylight_core/light.py
@classmethod
def all_lights(cls, *, reset: bool = True, exclusive: bool = True) -> list[Light]:
    """Return a list of all lights ready for use."""
    lights: list[Light] = []

    for subclass, devices in cls.available_lights().items():
        lights.extend(
            [
                subclass(device, reset=reset, exclusive=exclusive)
                for device in devices
            ]
        )

    return lights
busylight_core.Blink1.first_light classmethod
first_light(*, reset=True, exclusive=True)

Return the first unused light ready for use.

Raises: - NoLightsFoundError: if no lights are available.

Source code in src/busylight_core/light.py
@classmethod
def first_light(cls, *, reset: bool = True, exclusive: bool = True) -> Light:
    """Return the first unused light ready for use.

    Raises:
    - NoLightsFoundError: if no lights are available.

    """
    for subclass, devices in cls.available_lights().items():
        for device in devices:
            try:
                return subclass(device, reset=reset, exclusive=exclusive)
            except Exception as error:
                logger.info(f"Failed to acquire {device}: {error}")
                raise

    raise NoLightsFoundError
busylight_core.Blink1.reset
reset()

Turn the light off and cancel associated asynchronous tasks.

Source code in src/busylight_core/light.py
def reset(self) -> None:
    """Turn the light off and cancel associated asynchronous tasks."""
    self.off()
    self.cancel_tasks()
busylight_core.Blink1.exclusive_access
exclusive_access()

Manage exclusive access to the light.

If the device is not acquired in exclusive mode, it will be acquired and released automatically.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def exclusive_access(self) -> Generator[None, None, None]:
    """Manage exclusive access to the light.

    If the device is not acquired in exclusive mode, it will be
    acquired and released automatically.
    """
    if not self._exclusive:
        self.hardware.acquire()

    yield

    if not self._exclusive:
        self.hardware.release()
busylight_core.Blink1.update
update()

Obtain the current state of the light and writes it to the device.

Raises: - LightUnavailableError

Source code in src/busylight_core/light.py
def update(self) -> None:
    """Obtain the current state of the light and writes it to the device.

    Raises:
    - LightUnavailableError

    """
    state = bytes(self)

    match self.platform:
        case "Windows_10":
            state = bytes([0]) + state
        case "Darwin" | "Linux" | "Windows_11":
            pass
        case _:
            logger.info(f"Unsupported OS {self.platform}, hoping for the best.")

    with self.exclusive_access():
        logger.debug(f"{self.name} payload {state.hex(':')}")

        try:
            self.write_strategy(state)
        except Exception as error:
            logger.error(f"{self}: {error}")
            raise LightUnavailableError(self) from None
busylight_core.Blink1.batch_update
batch_update()

Update the software state of the light on exit.

Source code in src/busylight_core/light.py
@contextlib.contextmanager
def batch_update(self) -> Generator[None, None, None]:
    """Update the software state of the light on exit."""
    yield
    self.update()
busylight_core.Blink1.off
off(led=0)

Deactivate the light.

Source code in src/busylight_core/light.py
def off(self, led: int = 0) -> None:
    """Deactivate the light."""
    self.on((0, 0, 0), led)
busylight_core.Blink1.vendor staticmethod
vendor()

Return the vendor name for this device.

Source code in src/busylight_core/vendors/thingm/blink1.py
@staticmethod
def vendor() -> str:
    """Return the vendor name for this device."""
    return "ThingM"
busylight_core.Blink1.on
on(color, led=0)

Turn on the Blink(1) with the specified color.

PARAMETER DESCRIPTION
color

RGB color tuple (red, green, blue) with values 0-255

TYPE: tuple[int, int, int]

led

LED index for targeting specific LEDs

TYPE: int DEFAULT: 0

Source code in src/busylight_core/vendors/thingm/blink1.py
def on(self, color: tuple[int, int, int], led: int = 0) -> None:
    """Turn on the Blink(1) with the specified color.

    Args:
        color: RGB color tuple (red, green, blue) with values 0-255
        led: LED index for targeting specific LEDs

    """
    self.color = color
    with self.batch_update():
        self.state.fade_to_color(self.color, leds=LEDS(led))