Busylight omega

busylight_core.vendors.kuando.busylight_omega

Busylight Omega Support

Classes

busylight_core.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.Busylight_Omega.supported_device_ids class-attribute

supported_device_ids = {
    (10171, 15309): "Busylight Omega",
    (10171, 15311): "Busylight Omega",
}

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.event_loop cached property

event_loop

The default event loop.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.tasks cached property

tasks

Active tasks that are associated with this class.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.red property writable

red

Red color value.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.green property writable

green

Green color value.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.blue property writable

blue

Blue color value.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.color property writable

color

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

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.is_lit property

is_lit

True if any color value is greater than 0.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.hardware instance-attribute

hardware = hardware

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.path cached property

path

The path to the hardware device.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.platform cached property

platform

The discovered operating system platform name.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.exclusive property

exclusive

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

busylight_core.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.Busylight_Omega.name cached property

name

Return the marketing name of this light.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.hex property

hex

Return the hexadecimal representation of the light's state.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.read_strategy property

read_strategy

Return the read method used by this light.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.write_strategy property

write_strategy

Return the write method used by this light.

busylight_core.vendors.kuando.busylight_omega.Busylight_Omega.state cached property

state

Get the device state manager for controlling light patterns.

Functions

busylight_core.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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.vendors.kuando.busylight_omega.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")