light.py

Go to the documentation of this file.

"""Support for the Philips Hue lights."""
 
from __future__ import annotations
 
import asyncio
from datetime import timedelta
from functools import partial
import logging
import random
 
import aiohue
 
from homeassistant.components.light import (
    ATTR_BRIGHTNESS,
    ATTR_COLOR_TEMP,
    ATTR_EFFECT,
    ATTR_FLASH,
    ATTR_HS_COLOR,
    ATTR_TRANSITION,
    EFFECT_COLORLOOP,
    EFFECT_RANDOM,
    FLASH_LONG,
    FLASH_SHORT,
    ColorMode,
    LightEntity,
    LightEntityFeature,
    filter_supported_color_modes,
)
from homeassistant.core import callback
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import (
    CoordinatorEntity,
    DataUpdateCoordinator,
    UpdateFailed,
)
from homeassistant.util import color
 
from ..bridge import HueBridge
from ..const import (
    CONF_ALLOW_HUE_GROUPS,
    CONF_ALLOW_UNREACHABLE,
    DEFAULT_ALLOW_HUE_GROUPS,
    DEFAULT_ALLOW_UNREACHABLE,
    DOMAIN as HUE_DOMAIN,
    GROUP_TYPE_ENTERTAINMENT,
    GROUP_TYPE_LIGHT_GROUP,
    GROUP_TYPE_LIGHT_SOURCE,
    GROUP_TYPE_LUMINAIRE,
    GROUP_TYPE_ROOM,
    GROUP_TYPE_ZONE,
    REQUEST_REFRESH_DELAY,
)
from .helpers import remove_devices
 
SCAN_INTERVAL = timedelta(seconds=5)
 
LOGGER = logging.getLogger(__name__)
 
COLOR_MODES_HUE_ON_OFF = {ColorMode.ONOFF}
COLOR_MODES_HUE_DIMMABLE = {ColorMode.BRIGHTNESS}
COLOR_MODES_HUE_COLOR_TEMP = {ColorMode.COLOR_TEMP}
COLOR_MODES_HUE_COLOR = {ColorMode.HS}
COLOR_MODES_HUE_EXTENDED = {ColorMode.COLOR_TEMP, ColorMode.HS}
 
COLOR_MODES_HUE = {
    "Extended color light": COLOR_MODES_HUE_EXTENDED,
    "Color light": COLOR_MODES_HUE_COLOR,
    "Dimmable light": COLOR_MODES_HUE_DIMMABLE,
    "On/Off plug-in unit": COLOR_MODES_HUE_ON_OFF,
    "Color temperature light": COLOR_MODES_HUE_COLOR_TEMP,
}
 
SUPPORT_HUE_ON_OFF = LightEntityFeature.FLASH | LightEntityFeature.TRANSITION
SUPPORT_HUE_DIMMABLE = SUPPORT_HUE_ON_OFF
SUPPORT_HUE_COLOR_TEMP = SUPPORT_HUE_DIMMABLE
SUPPORT_HUE_COLOR = SUPPORT_HUE_DIMMABLE | LightEntityFeature.EFFECT
SUPPORT_HUE_EXTENDED = SUPPORT_HUE_COLOR_TEMP | SUPPORT_HUE_COLOR
 
SUPPORT_HUE = {
    "Extended color light": SUPPORT_HUE_EXTENDED,
    "Color light": SUPPORT_HUE_COLOR,
    "Dimmable light": SUPPORT_HUE_DIMMABLE,
    "On/Off plug-in unit": SUPPORT_HUE_ON_OFF,
    "Color temperature light": SUPPORT_HUE_COLOR_TEMP,
}
 
ATTR_IS_HUE_GROUP = "is_hue_group"
GAMUT_TYPE_UNAVAILABLE = "None"
# Minimum Hue Bridge API version to support groups
# 1.4.0 introduced extended group info
# 1.12 introduced the state object for groups
# 1.13 introduced "any_on" to group state objects
GROUP_MIN_API_VERSION = (1, 13, 0)
 
 
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
    """Old way of setting up Hue lights.
 
    Can only be called when a user accidentally mentions hue platform in their
    config. But even in that case it would have been ignored.
    """
 
 
def create_light(item_class, coordinator, bridge, is_group, rooms, api, item_id):
    """Create the light."""
    api_item = api[item_id]
 
    if is_group:
        supported_color_modes = set()
        supported_features = LightEntityFeature(0)
        for light_id in api_item.lights:
            if light_id not in bridge.api.lights:
                continue
            light = bridge.api.lights[light_id]
            supported_features |= SUPPORT_HUE.get(light.type, SUPPORT_HUE_EXTENDED)
            supported_color_modes.update(
                COLOR_MODES_HUE.get(light.type, COLOR_MODES_HUE_EXTENDED)
            )
        supported_features = supported_features or SUPPORT_HUE_EXTENDED
        supported_color_modes = supported_color_modes or COLOR_MODES_HUE_EXTENDED
        supported_color_modes = filter_supported_color_modes(supported_color_modes)
    else:
        supported_color_modes = COLOR_MODES_HUE.get(
            api_item.type, COLOR_MODES_HUE_EXTENDED
        )
        supported_features = SUPPORT_HUE.get(api_item.type, SUPPORT_HUE_EXTENDED)
    return item_class(
        coordinator,
        bridge,
        is_group,
        api_item,
        supported_color_modes,
        supported_features,
        rooms,
    )
 
 
async def async_setup_entry(hass, config_entry, async_add_entities):
    """Set up the Hue lights from a config entry."""
    bridge: HueBridge = hass.data[HUE_DOMAIN][config_entry.entry_id]
    api_version = tuple(int(v) for v in bridge.api.config.apiversion.split("."))
    rooms = {}
 
    allow_groups = config_entry.options.get(
        CONF_ALLOW_HUE_GROUPS, DEFAULT_ALLOW_HUE_GROUPS
    )
    supports_groups = api_version >= GROUP_MIN_API_VERSION
    if allow_groups and not supports_groups:
        LOGGER.warning("Please update your Hue bridge to support groups")
 
    light_coordinator = DataUpdateCoordinator(
        hass,
        LOGGER,
        name="light",
        update_method=partial(async_safe_fetch, bridge, bridge.api.lights.update),
        update_interval=SCAN_INTERVAL,
        request_refresh_debouncer=Debouncer(
            bridge.hass, LOGGER, cooldown=REQUEST_REFRESH_DELAY, immediate=True
        ),
    )
 
    # First do a refresh to see if we can reach the hub.
    # Otherwise we will declare not ready.
    await light_coordinator.async_refresh()
 
    if not light_coordinator.last_update_success:
        raise PlatformNotReady
 
    if not supports_groups:
        update_lights_without_group_support = partial(
            async_update_items,
            bridge,
            bridge.api.lights,
            {},
            async_add_entities,
            partial(create_light, HueLight, light_coordinator, bridge, False, rooms),
            None,
        )
        # We add a listener after fetching the data, so manually trigger listener
        bridge.reset_jobs.append(
            light_coordinator.async_add_listener(update_lights_without_group_support)
        )
        return
 
    group_coordinator = DataUpdateCoordinator(
        hass,
        LOGGER,
        name="group",
        update_method=partial(async_safe_fetch, bridge, bridge.api.groups.update),
        update_interval=SCAN_INTERVAL,
        request_refresh_debouncer=Debouncer(
            bridge.hass, LOGGER, cooldown=REQUEST_REFRESH_DELAY, immediate=True
        ),
    )
 
    if allow_groups:
        update_groups = partial(
            async_update_items,
            bridge,
            bridge.api.groups,
            {},
            async_add_entities,
            partial(create_light, HueLight, group_coordinator, bridge, True, None),
            None,
        )
 
        bridge.reset_jobs.append(group_coordinator.async_add_listener(update_groups))
 
    cancel_update_rooms_listener = None
 
    @callback
    def _async_update_rooms():
        """Update rooms."""
        nonlocal cancel_update_rooms_listener
        rooms.clear()
        for item_id in bridge.api.groups:
            group = bridge.api.groups[item_id]
            if group.type not in [GROUP_TYPE_ROOM, GROUP_TYPE_ZONE]:
                continue
            for light_id in group.lights:
                rooms[light_id] = group.name
 
        # Once we do a rooms update, we cancel the listener
        # until the next time lights are added
        bridge.reset_jobs.remove(cancel_update_rooms_listener)
        cancel_update_rooms_listener()
        cancel_update_rooms_listener = None
 
    @callback
    def _setup_rooms_listener():
        nonlocal cancel_update_rooms_listener
        if cancel_update_rooms_listener is not None:
            # If there are new lights added before _async_update_rooms
            # is called we should not add another listener
            return
 
        cancel_update_rooms_listener = group_coordinator.async_add_listener(
            _async_update_rooms
        )
        bridge.reset_jobs.append(cancel_update_rooms_listener)
 
    _setup_rooms_listener()
    await group_coordinator.async_refresh()
 
    update_lights_with_group_support = partial(
        async_update_items,
        bridge,
        bridge.api.lights,
        {},
        async_add_entities,
        partial(create_light, HueLight, light_coordinator, bridge, False, rooms),
        _setup_rooms_listener,
    )
    # We add a listener after fetching the data, so manually trigger listener
    bridge.reset_jobs.append(
        light_coordinator.async_add_listener(update_lights_with_group_support)
    )
    update_lights_with_group_support()
 
 
async def async_safe_fetch(bridge, fetch_method):
    """Safely fetch data."""
    try:
        async with asyncio.timeout(4):
            return await bridge.async_request_call(fetch_method)
    except aiohue.Unauthorized as err:
        await bridge.handle_unauthorized_error()
        raise UpdateFailed("Unauthorized") from err
    except aiohue.AiohueException as err:
        raise UpdateFailed(f"Hue error: {err}") from err
 
 
@callback
def async_update_items(
    bridge, api, current, async_add_entities, create_item, new_items_callback
):
    """Update items."""
    new_items = []
 
    for item_id in api:
        if item_id in current:
            continue
 
        current[item_id] = create_item(api, item_id)
        new_items.append(current[item_id])
 
    bridge.hass.async_create_task(remove_devices(bridge, api, current))
 
    if new_items:
        # This is currently used to setup the listener to update rooms
        if new_items_callback:
            new_items_callback()
        async_add_entities(new_items)
 
 
def hue_brightness_to_hass(value):
    """Convert hue brightness 1..254 to hass format 0..255."""
    return min(255, round((value / 254) * 255))
 
 
def hass_to_hue_brightness(value):
    """Convert hass brightness 0..255 to hue 1..254 scale."""
    return max(1, round((value / 255) * 254))
 
 
# pylint: disable-next=hass-enforce-class-module
class HueLight(CoordinatorEntity, LightEntity):
    """Representation of a Hue light."""
 
    def __init__(
        self,
        coordinator,
        bridge,
        is_group,
        light,
        supported_color_modes,
        supported_features,
        rooms,
    ):
        """Initialize the light."""
        super().__init__(coordinator)
        self._attr_supported_color_modes_attr_supported_color_modes = supported_color_modes
        self._attr_supported_features_attr_supported_features = supported_features
        self.lightlight = light
        self.bridgebridge = bridge
        self.is_groupis_group = is_group
        self._rooms_rooms = rooms
        self.allow_unreachableallow_unreachable = self.bridgebridge.config_entry.options.get(
            CONF_ALLOW_UNREACHABLE, DEFAULT_ALLOW_UNREACHABLE
        )
 
        self._fixed_color_mode_fixed_color_mode = None
        if len(supported_color_modes) == 1:
            self._fixed_color_mode_fixed_color_mode = next(iter(supported_color_modes))
        else:
            assert supported_color_modes == {ColorMode.COLOR_TEMP, ColorMode.HS}
 
        if is_group:
            self.is_osramis_osram = False
            self.is_philipsis_philips = False
            self.is_innris_innr = False
            self.is_ewelinkis_ewelink = False
            self.is_livarnois_livarno = False
            self.is_s31litezbis_s31litezb = False
            self.gamut_typgamut_typ = GAMUT_TYPE_UNAVAILABLE
            self.gamutgamut = None
        else:
            self.is_osramis_osram = light.manufacturername == "OSRAM"
            self.is_philipsis_philips = light.manufacturername == "Philips"
            self.is_innris_innr = light.manufacturername == "innr"
            self.is_ewelinkis_ewelink = light.manufacturername == "eWeLink"
            self.is_livarnois_livarno = light.manufacturername.startswith("_TZ3000_")
            self.is_s31litezbis_s31litezb = light.modelid == "S31 Lite zb"
            self.gamut_typgamut_typ = self.lightlight.colorgamuttype
            self.gamutgamut = self.lightlight.colorgamut
            LOGGER.debug("Color gamut of %s: %s", self.namenamename, str(self.gamutgamut))
            if self.lightlight.swupdatestate == "readytoinstall":
                err = (
                    "Please check for software updates of the %s "
                    "bulb in the Philips Hue App."
                )
                LOGGER.warning(err, self.namenamename)
            if self.gamutgamut and not color.check_valid_gamut(self.gamutgamut):
                err = "Color gamut of %s: %s, not valid, setting gamut to None."
                LOGGER.debug(err, self.namenamename, str(self.gamutgamut))
                self.gamut_typgamut_typ = GAMUT_TYPE_UNAVAILABLE
                self.gamutgamut = None
 
    @property
    def unique_id(self):
        """Return the unique ID of this Hue light."""
        unique_id = self.lightlight.uniqueid
        if not unique_id and self.is_groupis_group:
            unique_id = self.lightlight.id
 
        return unique_id
 
    @property
    def device_id(self):
        """Return the ID of this Hue light."""
        return self.unique_idunique_idunique_id
 
    @property
    def name(self):
        """Return the name of the Hue light."""
        return self.lightlight.name
 
    @property
    def brightness(self):
        """Return the brightness of this light between 0..255."""
        if self.is_groupis_group:
            bri = self.lightlight.action.get("bri")
        else:
            bri = self.lightlight.state.get("bri")
 
        if bri is None:
            return bri
 
        return hue_brightness_to_hass(bri)
 
    @property
    def color_mode(self) -> str:
        """Return the color mode of the light."""
        if self._fixed_color_mode_fixed_color_mode:
            return self._fixed_color_mode_fixed_color_mode
 
        # The light supports both hs/xy and white with adjustabe color_temperature
        mode = self._color_mode_color_mode
        if mode in ("xy", "hs"):
            return ColorMode.HS
 
        return ColorMode.COLOR_TEMP
 
    @property
    def _color_mode(self):
        """Return the hue color mode."""
        if self.is_groupis_group:
            return self.lightlight.action.get("colormode")
        return self.lightlight.state.get("colormode")
 
    @property
    def hs_color(self):
        """Return the hs color value."""
        mode = self._color_mode_color_mode
        source = self.lightlight.action if self.is_groupis_group else self.lightlight.state
 
        if mode in ("xy", "hs") and "xy" in source:
            return color.color_xy_to_hs(*source["xy"], self.gamutgamut)
 
        return None
 
    @property
    def color_temp(self):
        """Return the CT color value."""
        # Don't return color temperature unless in color temperature mode
        if self._color_mode_color_mode != "ct":
            return None
 
        if self.is_groupis_group:
            return self.lightlight.action.get("ct")
        return self.lightlight.state.get("ct")
 
    @property
    def min_mireds(self):
        """Return the coldest color_temp that this light supports."""
        if self.is_groupis_group:
            return super().min_mireds
 
        min_mireds = self.lightlight.controlcapabilities.get("ct", {}).get("min")
 
        # We filter out '0' too, which can be incorrectly reported by 3rd party buls
        if not min_mireds:
            return super().min_mireds
 
        return min_mireds
 
    @property
    def max_mireds(self):
        """Return the warmest color_temp that this light supports."""
        if self.is_groupis_group:
            return super().max_mireds
        if self.is_livarnois_livarno:
            return 500
 
        max_mireds = self.lightlight.controlcapabilities.get("ct", {}).get("max")
 
        if not max_mireds:
            return super().max_mireds
 
        return max_mireds
 
    @property
    def is_on(self):
        """Return true if device is on."""
        if self.is_groupis_group:
            return self.lightlight.state["any_on"]
        return self.lightlight.state["on"]
 
    @property
    def available(self):
        """Return if light is available."""
        return self.coordinator.last_update_success and (
            self.is_groupis_group or self.allow_unreachableallow_unreachable or self.lightlight.state["reachable"]
        )
 
    @property
    def effect(self):
        """Return the current effect."""
        return self.lightlight.state.get("effect", None)
 
    @property
    def effect_list(self):
        """Return the list of supported effects."""
        if self.is_osramis_osram:
            return [EFFECT_RANDOM]
        return [EFFECT_COLORLOOP, EFFECT_RANDOM]
 
    @property
    def device_info(self) -> DeviceInfo | None:
        """Return the device info."""
        if self.lightlight.type in (
            GROUP_TYPE_ENTERTAINMENT,
            GROUP_TYPE_LIGHT_GROUP,
            GROUP_TYPE_ROOM,
            GROUP_TYPE_LUMINAIRE,
            GROUP_TYPE_LIGHT_SOURCE,
            GROUP_TYPE_ZONE,
        ):
            return None
 
        suggested_area = None
        if self._rooms_rooms and self.lightlight.id in self._rooms_rooms:
            suggested_area = self._rooms_rooms[self.lightlight.id]
 
        return DeviceInfo(
            identifiers={(HUE_DOMAIN, self.device_iddevice_id)},
            manufacturer=self.lightlight.manufacturername,
            # productname added in Hue Bridge API 1.24
            # (published 03/05/2018)
            model=self.lightlight.productname or self.lightlight.modelid,
            name=self.namenamename,
            sw_version=self.lightlight.swversion,
            suggested_area=suggested_area,
            via_device=(HUE_DOMAIN, self.bridgebridge.api.config.bridgeid),
        )
 
    async def async_turn_on(self, **kwargs):
        """Turn the specified or all lights on."""
        command = {"on": True}
 
        if ATTR_TRANSITION in kwargs:
            command["transitiontime"] = int(kwargs[ATTR_TRANSITION] * 10)
 
        if ATTR_HS_COLOR in kwargs:
            if self.is_osramis_osram:
                command["hue"] = int(kwargs[ATTR_HS_COLOR][0] / 360 * 65535)
                command["sat"] = int(kwargs[ATTR_HS_COLOR][1] / 100 * 255)
            else:
                # Philips hue bulb models respond differently to hue/sat
                # requests, so we convert to XY first to ensure a consistent
                # color.
                xy_color = color.color_hs_to_xy(*kwargs[ATTR_HS_COLOR], self.gamutgamut)
                command["xy"] = xy_color
        elif ATTR_COLOR_TEMP in kwargs:
            temp = kwargs[ATTR_COLOR_TEMP]
            command["ct"] = max(self.min_miredsmin_miredsmin_mireds, min(temp, self.max_miredsmax_miredsmax_mireds))
 
        if ATTR_BRIGHTNESS in kwargs:
            command["bri"] = hass_to_hue_brightness(kwargs[ATTR_BRIGHTNESS])
 
        flash = kwargs.get(ATTR_FLASH)
 
        if flash == FLASH_LONG:
            command["alert"] = "lselect"
            del command["on"]
        elif flash == FLASH_SHORT:
            command["alert"] = "select"
            del command["on"]
        elif (
            not self.is_innris_innr
            and not self.is_ewelinkis_ewelink
            and not self.is_livarnois_livarno
            and not self.is_s31litezbis_s31litezb
        ):
            command["alert"] = "none"
 
        if ATTR_EFFECT in kwargs:
            effect = kwargs[ATTR_EFFECT]
            if effect == EFFECT_COLORLOOP:
                command["effect"] = "colorloop"
            elif effect == EFFECT_RANDOM:
                command["hue"] = random.randrange(0, 65535)
                command["sat"] = random.randrange(150, 254)
            else:
                command["effect"] = "none"
 
        if self.is_groupis_group:
            await self.bridgebridge.async_request_call(self.lightlight.set_action, **command)
        else:
            await self.bridgebridge.async_request_call(self.lightlight.set_state, **command)
 
        await self.coordinator.async_request_refresh()
 
    async def async_turn_off(self, **kwargs):
        """Turn the specified or all lights off."""
        command = {"on": False}
 
        if ATTR_TRANSITION in kwargs:
            command["transitiontime"] = int(kwargs[ATTR_TRANSITION] * 10)
 
        flash = kwargs.get(ATTR_FLASH)
 
        if flash == FLASH_LONG:
            command["alert"] = "lselect"
            del command["on"]
        elif flash == FLASH_SHORT:
            command["alert"] = "select"
            del command["on"]
        elif not self.is_innris_innr and not self.is_livarnois_livarno:
            command["alert"] = "none"
 
        if self.is_groupis_group:
            await self.bridgebridge.async_request_call(self.lightlight.set_action, **command)
        else:
            await self.bridgebridge.async_request_call(self.lightlight.set_state, **command)
 
        await self.coordinator.async_request_refresh()
 
    @property
    def extra_state_attributes(self):
        """Return the device state attributes."""
        if not self.is_groupis_group:
            return {}
        return {ATTR_IS_HUE_GROUP: self.is_groupis_group}