huey.py

from typing import Dict, List

import HueBLE
from HueBLE import HueBleLight
from bleak import BleakScanner

# noinspection PyBroadException
def temperature_to_mireds_or_kelvin(value: float) -> int:
    if value == 0:
        return 0

    return round(1_000_000 / value)

class Huey:
    lights: Dict[str, HueBleLight] = {}
    max_connections = 8
    max_attempts = 3
    connection_timeout = 120
    wait_timeout = 240

    async def start(self, light_addresses : List[str]) -> None:
        print("Registering devices ... ")
        for light_address in light_addresses:
            device = await BleakScanner.find_device_by_address(light_address)
            light = HueBLE.HueBleLight(device)

            print(f"Registering light: {device.address}")
            self.lights[device.address] = light

            await light.poll_state(self.connection_timeout, False)
            await light.disconnect()

        num_connected_lights = 0
        for light in self.lights.values():
            if num_connected_lights >= self.max_connections:
                break

            num_connected_lights += 1

            if light.connected:
                continue

            print(f"Connecting to light: {light.address}")
            await light.connect(self.max_attempts, self.connection_timeout, self.wait_timeout, False)

    async def poll(self, address):
        print(f"Polling {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return

        light = self.lights[address]
        await light.poll_state(self.connection_timeout, False)

    def get_light_by_address(self, address):
        if address not in self.lights:
            return None

        light = self.lights[address]

        x = light.colour_xy[0]
        y = light.colour_xy[1]

        temperature = temperature_to_mireds_or_kelvin(light.colour_temp)
        min_temperature = temperature_to_mireds_or_kelvin(light.maximum_mireds)
        max_temperature = temperature_to_mireds_or_kelvin(light.minimum_mireds)

        is_in_temperature_mode = temperature >= min_temperature

        return {
            'address': light.address,
            'name': light.name,
            'is_in_temperature_mode': is_in_temperature_mode,
            'is_available': light.available,
            'is_powered_on': light.power_state,
            'brightness': light.brightness,
            'temperature': temperature,
            'min_temperature': min_temperature,
            'max_temperature': max_temperature,
            'color': {
                'x': x,
                'y': y
            }
        }

    def get_list(self) -> list:
        data = []

        for light in self.lights.values():
            data.append(self.get_light_by_address(light.address))

        return data

    async def set_power(self, address: str, new_state: bool, retry: bool = True) -> bool:
        print(f"Want to toggle power state for {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return False

        await self.connect(address)

        light = self.lights[address]

        print(f"Setting power for {address} to {new_state} ... ")

        try:
            await light.set_power(new_state)
        except Exception:
            if retry:
                await self.lights[address].reconnect()
                return await self.set_power(address, new_state, False)
            else:
                print('Unable to set power state')
                return False

        return True

    async def set_brightness(self, address: str, new_brightness: int, retry: bool = True) -> bool:
        print(f"Want to change brightness for {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return False

        await self.connect(address)

        light = self.lights[address]

        print(f"Setting brightness for {address} to {new_brightness} ... ")

        try:
            await light.set_brightness(new_brightness)
        except Exception:
            if retry:
                await self.lights[address].reconnect()
                return await self.set_brightness(address, new_brightness, False)
            else:
                print('Unable to set brightness')
                return False

        return True

    async def set_temp(self, address: str, temp: int, retry: bool = True) -> bool:
        print(f"Want to change temp for {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return False

        await self.connect(address)

        light = self.lights[address]

        print(f"Setting color temperature for {address} to {temp} ... ")

        mireds = temperature_to_mireds_or_kelvin(temp)

        try:
            await light.set_colour_temp(mireds)
        except Exception:
            if retry:
                await self.lights[address].reconnect()
                return await self.set_temp(address, temp, False)
            else:
                print('Unable to set color temperature')
                return False

        return True

    async def set_color(self, address: str, x: float, y: float, retry: bool = True) -> bool:
        print(f"Want to change color for {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return False

        await self.connect(address)

        light = self.lights[address]

        print(f"Setting color for {address} to {x}, {y}... ")

        try:
            await light.set_colour_xy(x, y)
        except Exception:
            if retry:
                await self.lights[address].reconnect()
                return await self.set_color(address, x, y, False)
            else:
                print('Unable to set color')
                return False

        return True

    async def set_name(self, address: str, name: str, retry: bool = True) -> bool:
        print(f"Want to change name for {address} ... ")
        if address not in self.lights:
            print(f"Unknown device: {address}")
            return False

        await self.connect(address)

        light = self.lights[address]

        print(f"Setting name {address} to {name}... ")

        try:
            await light.set_light_name(name)
        except Exception:
            if retry:
                await self.lights[address].reconnect()
                return await self.set_name(address, name, False)
            else:
                print('Unable to set name')
                return False

        return True

    async def disconnect_all(self):
        for light in self.lights.values():
            if not light.connected:
                continue
            print(f"Disconnecting: {light.address} ...")
            await light.disconnect()

    connecting_to = None
    async def connect(self, address: str):
        if self.connecting_to == address:
            return

        if address not in self.lights:
            print(f"Unknown device: {address}")
            return

        light = self.lights[address]
        if light.connected:
            print(f"Already connected to: {address}")
            return

        self.connecting_to = address

        connected_lights = []
        for other_light in self.lights.values():
            if light is other_light:
                continue

            if not other_light.connected:
                continue

            connected_lights.append(other_light)

        if len(connected_lights) >= self.max_connections:
            await connected_lights[0].disconnect()

        print(f"Establishing connection: {light.address} ...")
        await light.connect(self.max_attempts, self.connection_timeout, self.wait_timeout, False)
        await light.poll_state(self.connection_timeout, False)

        self.connecting_to = None