core/homeassistant/components/light/hue.py

"""
Support for Hue lights.

For more details about this platform, please refer to the documentation at
https://home-assistant.io/components/light.hue/
"""
import json
import logging
import os
import random
import socket
from datetime import timedelta

import voluptuous as vol

import homeassistant.util as util
import homeassistant.util.color as color_util
from homeassistant.components.light import (
    ATTR_BRIGHTNESS, ATTR_COLOR_TEMP, ATTR_EFFECT, ATTR_FLASH, ATTR_RGB_COLOR,
    ATTR_TRANSITION, ATTR_XY_COLOR, EFFECT_COLORLOOP, EFFECT_RANDOM,
    FLASH_LONG, FLASH_SHORT, SUPPORT_BRIGHTNESS, SUPPORT_COLOR_TEMP,
    SUPPORT_EFFECT, SUPPORT_FLASH, SUPPORT_RGB_COLOR, SUPPORT_TRANSITION,
    SUPPORT_XY_COLOR, Light, PLATFORM_SCHEMA)
from homeassistant.config import load_yaml_config_file
from homeassistant.const import (CONF_FILENAME, CONF_HOST, DEVICE_DEFAULT_NAME)
from homeassistant.components.emulated_hue import ATTR_EMULATED_HUE
import homeassistant.helpers.config_validation as cv

REQUIREMENTS = ['phue==1.0']

# Track previously setup bridges
_CONFIGURED_BRIDGES = {}
# Map ip to request id for configuring
_CONFIGURING = {}
_LOGGER = logging.getLogger(__name__)

CONF_ALLOW_UNREACHABLE = 'allow_unreachable'

DEFAULT_ALLOW_UNREACHABLE = False
DOMAIN = "light"
SERVICE_HUE_SCENE = "hue_activate_scene"

MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
MIN_TIME_BETWEEN_FORCED_SCANS = timedelta(milliseconds=100)

PHUE_CONFIG_FILE = 'phue.conf'

SUPPORT_HUE_ON_OFF = (SUPPORT_FLASH | SUPPORT_TRANSITION)
SUPPORT_HUE_DIMMABLE = (SUPPORT_HUE_ON_OFF | SUPPORT_BRIGHTNESS)
SUPPORT_HUE_COLOR_TEMP = (SUPPORT_HUE_DIMMABLE | SUPPORT_COLOR_TEMP)
SUPPORT_HUE_COLOR = (SUPPORT_HUE_DIMMABLE | SUPPORT_EFFECT |
                     SUPPORT_RGB_COLOR | SUPPORT_XY_COLOR)
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
    }

CONF_ALLOW_IN_EMULATED_HUE = "allow_in_emulated_hue"
DEFAULT_ALLOW_IN_EMULATED_HUE = True

CONF_ALLOW_HUE_GROUPS = "allow_hue_groups"
DEFAULT_ALLOW_HUE_GROUPS = True

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
    vol.Optional(CONF_HOST): cv.string,
    vol.Optional(CONF_ALLOW_UNREACHABLE): cv.boolean,
    vol.Optional(CONF_FILENAME): cv.string,
    vol.Optional(CONF_ALLOW_IN_EMULATED_HUE): cv.boolean,
    vol.Optional(CONF_ALLOW_HUE_GROUPS,
                 default=DEFAULT_ALLOW_HUE_GROUPS): cv.boolean,
})

ATTR_GROUP_NAME = "group_name"
ATTR_SCENE_NAME = "scene_name"
SCENE_SCHEMA = vol.Schema({
    vol.Required(ATTR_GROUP_NAME): cv.string,
    vol.Required(ATTR_SCENE_NAME): cv.string,
})

ATTR_IS_HUE_GROUP = "is_hue_group"
GROUP_NAME_ALL_HUE_LIGHTS = "All Hue Lights"


def _find_host_from_config(hass, filename=PHUE_CONFIG_FILE):
    """Attempt to detect host based on existing configuration."""
    path = hass.config.path(filename)

    if not os.path.isfile(path):
        return None

    try:
        with open(path) as inp:
            return next(json.loads(''.join(inp)).keys().__iter__())
    except (ValueError, AttributeError, StopIteration):
        # ValueError if can't parse as JSON
        # AttributeError if JSON value is not a dict
        # StopIteration if no keys
        return None


def setup_platform(hass, config, add_devices, discovery_info=None):
    """Set up the Hue lights."""
    # Default needed in case of discovery
    filename = config.get(CONF_FILENAME, PHUE_CONFIG_FILE)
    allow_unreachable = config.get(CONF_ALLOW_UNREACHABLE,
                                   DEFAULT_ALLOW_UNREACHABLE)
    allow_in_emulated_hue = config.get(CONF_ALLOW_IN_EMULATED_HUE,
                                       DEFAULT_ALLOW_IN_EMULATED_HUE)
    allow_hue_groups = config.get(CONF_ALLOW_HUE_GROUPS)

    if discovery_info is not None:
        if "HASS Bridge" in discovery_info.get('name', ''):
            _LOGGER.info("Emulated hue found, will not add")
            return False

        host = discovery_info.get('host')
    else:
        host = config.get(CONF_HOST, None)

        if host is None:
            host = _find_host_from_config(hass, filename)

        if host is None:
            _LOGGER.error("No host found in configuration")
            return False

    # Only act if we are not already configuring this host
    if host in _CONFIGURING or \
            socket.gethostbyname(host) in _CONFIGURED_BRIDGES:
        return

    setup_bridge(host, hass, add_devices, filename, allow_unreachable,
                 allow_in_emulated_hue, allow_hue_groups)


def setup_bridge(host, hass, add_devices, filename, allow_unreachable,
                 allow_in_emulated_hue, allow_hue_groups):
    """Set up a phue bridge based on host parameter."""
    import phue

    try:
        bridge = phue.Bridge(
            host,
            config_file_path=hass.config.path(filename))
    except ConnectionRefusedError:  # Wrong host was given
        _LOGGER.error("Error connecting to the Hue bridge at %s", host)

        return

    except phue.PhueRegistrationException:
        _LOGGER.warning("Connected to Hue at %s but not registered.", host)

        request_configuration(host, hass, add_devices, filename,
                              allow_unreachable, allow_in_emulated_hue,
                              allow_hue_groups)

        return

    # If we came here and configuring this host, mark as done
    if host in _CONFIGURING:
        request_id = _CONFIGURING.pop(host)
        configurator = hass.components.configurator
        configurator.request_done(request_id)

    lights = {}
    lightgroups = {}
    skip_groups = not allow_hue_groups

    @util.Throttle(MIN_TIME_BETWEEN_SCANS, MIN_TIME_BETWEEN_FORCED_SCANS)
    def update_lights():
        """Update the Hue light objects with latest info from the bridge."""
        nonlocal skip_groups

        try:
            api = bridge.get_api()
        except phue.PhueRequestTimeout:
            _LOGGER.warning("Timeout trying to reach the bridge")
            return
        except ConnectionRefusedError:
            _LOGGER.error("The bridge refused the connection")
            return
        except socket.error:
            # socket.error when we cannot reach Hue
            _LOGGER.exception("Cannot reach the bridge")
            return

        api_lights = api.get('lights')

        if not isinstance(api_lights, dict):
            _LOGGER.error("Got unexpected result from Hue API")
            return

        if skip_groups:
            api_groups = {}
        else:
            api_groups = api.get('groups')

        if not isinstance(api_groups, dict):
            _LOGGER.error("Got unexpected result from Hue API")
            return

        if not skip_groups:
            # Group ID 0 is a special group in the hub for all lights, but it
            # is not returned by get_api() so explicity get it and include it.
            # See https://developers.meethue.com/documentation/
            #               groups-api#21_get_all_groups
            _LOGGER.debug("Getting group 0 from bridge")
            all_lights = bridge.get_group(0)
            if not isinstance(all_lights, dict):
                _LOGGER.error("Got unexpected result from Hue API for group 0")
                return
            # Hue hub returns name of group 0 as "Group 0", so rename
            # for ease of use in HA.
            all_lights['name'] = GROUP_NAME_ALL_HUE_LIGHTS
            api_groups["0"] = all_lights

        new_lights = []

        api_name = api.get('config').get('name')
        if api_name in ('RaspBee-GW', 'deCONZ-GW'):
            bridge_type = 'deconz'
        else:
            bridge_type = 'hue'

        for light_id, info in api_lights.items():
            if light_id not in lights:
                lights[light_id] = HueLight(int(light_id), info,
                                            bridge, update_lights,
                                            bridge_type, allow_unreachable,
                                            allow_in_emulated_hue)
                new_lights.append(lights[light_id])
            else:
                lights[light_id].info = info
                lights[light_id].schedule_update_ha_state()

        for lightgroup_id, info in api_groups.items():
            if 'state' not in info:
                _LOGGER.warning("Group info does not contain state. "
                                "Please update your hub.")
                skip_groups = True
                break

            if lightgroup_id not in lightgroups:
                lightgroups[lightgroup_id] = HueLight(
                    int(lightgroup_id), info, bridge, update_lights,
                    bridge_type, allow_unreachable, allow_in_emulated_hue,
                    True)
                new_lights.append(lightgroups[lightgroup_id])
            else:
                lightgroups[lightgroup_id].info = info
                lightgroups[lightgroup_id].schedule_update_ha_state()

        if new_lights:
            add_devices(new_lights)

    _CONFIGURED_BRIDGES[socket.gethostbyname(host)] = True

    # create a service for calling run_scene directly on the bridge,
    # used to simplify automation rules.
    def hue_activate_scene(call):
        """Service to call directly directly into bridge to set scenes."""
        group_name = call.data[ATTR_GROUP_NAME]
        scene_name = call.data[ATTR_SCENE_NAME]
        bridge.run_scene(group_name, scene_name)

    descriptions = load_yaml_config_file(
        os.path.join(os.path.dirname(__file__), 'services.yaml'))
    hass.services.register(DOMAIN, SERVICE_HUE_SCENE, hue_activate_scene,
                           descriptions.get(SERVICE_HUE_SCENE),
                           schema=SCENE_SCHEMA)

    update_lights()


def request_configuration(host, hass, add_devices, filename,
                          allow_unreachable, allow_in_emulated_hue,
                          allow_hue_groups):
    """Request configuration steps from the user."""
    configurator = hass.components.configurator

    # We got an error if this method is called while we are configuring
    if host in _CONFIGURING:
        configurator.notify_errors(
            _CONFIGURING[host], "Failed to register, please try again.")

        return

    # pylint: disable=unused-argument
    def hue_configuration_callback(data):
        """Set up actions to do when our configuration callback is called."""
        setup_bridge(host, hass, add_devices, filename, allow_unreachable,
                     allow_in_emulated_hue, allow_hue_groups)

    _CONFIGURING[host] = configurator.request_config(
        "Philips Hue", hue_configuration_callback,
        description=("Press the button on the bridge to register Philips Hue "
                     "with Home Assistant."),
        entity_picture="/static/images/logo_philips_hue.png",
        description_image="/static/images/config_philips_hue.jpg",
        submit_caption="I have pressed the button"
    )


class HueLight(Light):
    """Representation of a Hue light."""

    def __init__(self, light_id, info, bridge, update_lights,
                 bridge_type, allow_unreachable, allow_in_emulated_hue,
                 is_group=False):
        """Initialize the light."""
        self.light_id = light_id
        self.info = info
        self.bridge = bridge
        self.update_lights = update_lights
        self.bridge_type = bridge_type
        self.allow_unreachable = allow_unreachable
        self.is_group = is_group
        self.allow_in_emulated_hue = allow_in_emulated_hue

        if is_group:
            self._command_func = self.bridge.set_group
        else:
            self._command_func = self.bridge.set_light

    @property
    def unique_id(self):
        """Return the ID of this Hue light."""
        lid = self.info.get('uniqueid')

        if lid is None:
            default_type = 'Group' if self.is_group else 'Light'
            ltype = self.info.get('type', default_type)
            lid = '{}.{}.{}'.format(self.name, ltype, self.light_id)

        return '{}.{}'.format(self.__class__, lid)

    @property
    def name(self):
        """Return the name of the Hue light."""
        return self.info.get('name', DEVICE_DEFAULT_NAME)

    @property
    def brightness(self):
        """Return the brightness of this light between 0..255."""
        if self.is_group:
            return self.info['action'].get('bri')
        return self.info['state'].get('bri')

    @property
    def xy_color(self):
        """Return the XY color value."""
        if self.is_group:
            return self.info['action'].get('xy')
        return self.info['state'].get('xy')

    @property
    def color_temp(self):
        """Return the CT color value."""
        if self.is_group:
            return self.info['action'].get('ct')
        return self.info['state'].get('ct')

    @property
    def is_on(self):
        """Return true if device is on."""
        if self.is_group:
            return self.info['state']['any_on']
        elif self.allow_unreachable:
            return self.info['state']['on']
        return self.info['state']['reachable'] and \
            self.info['state']['on']

    @property
    def supported_features(self):
        """Flag supported features."""
        return SUPPORT_HUE.get(self.info.get('type'), SUPPORT_HUE_EXTENDED)

    @property
    def effect_list(self):
        """Return the list of supported effects."""
        return [EFFECT_COLORLOOP, EFFECT_RANDOM]

    def 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_XY_COLOR in kwargs:
            if self.info.get('manufacturername') == "OSRAM":
                hue, sat = color_util.color_xy_to_hs(*kwargs[ATTR_XY_COLOR])
                command['hue'] = hue
                command['sat'] = sat
            else:
                command['xy'] = kwargs[ATTR_XY_COLOR]
        elif ATTR_RGB_COLOR in kwargs:
            if self.info.get('manufacturername') == "OSRAM":
                hsv = color_util.color_RGB_to_hsv(
                    *(int(val) for val in kwargs[ATTR_RGB_COLOR]))
                command['hue'] = hsv[0]
                command['sat'] = hsv[1]
                command['bri'] = hsv[2]
            else:
                xyb = color_util.color_RGB_to_xy(
                    *(int(val) for val in kwargs[ATTR_RGB_COLOR]))
                command['xy'] = xyb[0], xyb[1]
                command['bri'] = xyb[2]
        elif ATTR_COLOR_TEMP in kwargs:
            temp = kwargs[ATTR_COLOR_TEMP]
            command['ct'] = max(self.min_mireds, min(temp, self.max_mireds))

        if ATTR_BRIGHTNESS in kwargs:
            command['bri'] = 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 self.bridge_type == 'hue':
            command['alert'] = 'none'

        effect = kwargs.get(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)
        elif (self.bridge_type == 'hue' and
              self.info.get('manufacturername') == 'Philips'):
            command['effect'] = 'none'

        self._command_func(self.light_id, command)

    def 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 self.bridge_type == 'hue':
            command['alert'] = 'none'

        self._command_func(self.light_id, command)

    def update(self):
        """Synchronize state with bridge."""
        self.update_lights(no_throttle=True)

    @property
    def device_state_attributes(self):
        """Return the device state attributes."""
        attributes = {}
        if not self.allow_in_emulated_hue:
            attributes[ATTR_EMULATED_HUE] = self.allow_in_emulated_hue
        if self.is_group:
            attributes[ATTR_IS_HUE_GROUP] = self.is_group
        return attributes