"""
homeassistant.components.light
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Provides functionality to interact with lights.
"""
import logging
from datetime import datetime, timedelta
import homeassistant as ha
import homeassistant.util as util
import homeassistant.components.group as group
DOMAIN = "light"
GROUP_NAME_ALL_LIGHTS = 'all_lights'
ENTITY_ID_ALL_LIGHTS = group.ENTITY_ID_FORMAT.format(
GROUP_NAME_ALL_LIGHTS)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
MIN_TIME_BETWEEN_SCANS = timedelta(seconds=10)
def is_on(statemachine, entity_id=None):
""" Returns if the lights are on based on the statemachine. """
entity_id = entity_id or ENTITY_ID_ALL_LIGHTS
return statemachine.is_state(entity_id, ha.STATE_ON)
def turn_on(bus, entity_id=None, transition_seconds=None):
""" Turns all or specified light on. """
data = {}
if entity_id:
data["light_id"] = ha.split_entity_id(entity_id)[1]
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN, ha.SERVICE_TURN_ON, data)
def turn_off(bus, entity_id=None, transition_seconds=None):
""" Turns all or specified light off. """
data = {}
if entity_id:
data["light_id"] = ha.split_entity_id(entity_id)[1]
if transition_seconds:
data["transition_seconds"] = transition_seconds
bus.call_service(DOMAIN, ha.SERVICE_TURN_OFF, data)
def setup(bus, statemachine, light_control):
""" Exposes light control via statemachine and services. """
logger = logging.getLogger(__name__)
def update_light_state(time): # pylint: disable=unused-argument
""" Track the state of the lights. """
try:
should_update = datetime.now() - update_light_state.last_updated \
> MIN_TIME_BETWEEN_SCANS
except AttributeError: # if last_updated does not exist
should_update = True
if should_update:
logger.info("Updating light status")
update_light_state.last_updated = datetime.now()
status = {light_id: light_control.is_light_on(light_id)
for light_id in light_control.light_ids}
for light_id, state in status.items():
entity_id = ENTITY_ID_FORMAT.format(light_id)
new_state = ha.STATE_ON if state else ha.STATE_OFF
statemachine.set_state(entity_id, new_state)
ha.track_time_change(bus, update_light_state, second=[0, 30])
update_light_state(None)
# Track the all lights state
entity_ids = [ENTITY_ID_FORMAT.format(light_id) for light_id
in light_control.light_ids]
group.setup(bus, statemachine, GROUP_NAME_ALL_LIGHTS, entity_ids)
def handle_light_service(service):
""" Hande a turn light on or off service call. """
light_id = service.data.get("light_id", None)
transition_seconds = service.data.get("transition_seconds", None)
if service.service == ha.SERVICE_TURN_ON:
light_control.turn_light_on(light_id, transition_seconds)
else:
light_control.turn_light_off(light_id, transition_seconds)
update_light_state(None)
# Listen for light on and light off events
bus.register_service(DOMAIN, ha.SERVICE_TURN_ON,
handle_light_service)
bus.register_service(DOMAIN, ha.SERVICE_TURN_OFF,
handle_light_service)
return True
class HueLightControl(object):
""" Class to interface with the Hue light system. """
def __init__(self, host=None):
logger = logging.getLogger(__name__)
try:
import phue
import socket
except ImportError:
logger.exception(
"HueLightControl:Error while importing dependency phue.")
self.success_init = False
self._light_map = {}
return
try:
self._bridge = phue.Bridge(host)
except socket.error: # Error connecting using Phue
logger.exception((
"HueLightControl:Error while connecting to the bridge. "
"Is phue registered?"))
self.success_init = False
self._light_map = {}
return
self._light_map = {util.slugify(light.name): light for light
in self._bridge.get_light_objects()}
if not self._light_map:
logger.error("HueLightControl:Could not find any lights. ")
self.success_init = False
else:
self.success_init = True
@property
def light_ids(self):
""" Return a list of light ids. """
return self._light_map.keys()
def is_light_on(self, light_id=None):
""" Returns if specified or all light are on. """
if not light_id:
return any(
True for light_id in self._light_map.keys()
if self.is_light_on(light_id))
else:
light_id = self._convert_id(light_id)
if not light_id: # Not valid light_id submitted
return False
state = self._bridge.get_light(light_id)
try:
return state['state']['reachable'] and state['state']['on']
except KeyError:
# If key 'state', 'reachable' or 'on' not exists.
return False
def turn_light_on(self, light_id=None, transition_seconds=None):
""" Turn the specified or all lights on. """
self._turn_light(True, light_id, transition_seconds)
def turn_light_off(self, light_id=None, transition_seconds=None):
""" Turn the specified or all lights off. """
self._turn_light(False, light_id, transition_seconds)
def _turn_light(self, turn, light_id=None, transition_seconds=None):
""" Helper method to turn lights on or off. """
if light_id:
light_id = self._convert_id(light_id)
if not light_id: # Not valid light id submitted
return
else:
light_id = [light.light_id for light in self._light_map.values()]
command = {'on': True, 'xy': [0.5119, 0.4147], 'bri': 164} if turn \
else {'on': False}
if transition_seconds:
# Transition time is in 1/10th seconds and cannot exceed
# 900 seconds.
command['transitiontime'] = min(9000, transition_seconds * 10)
self._bridge.set_light(light_id, command)
def _convert_id(self, light_id):
""" Returns internal light id to be used with phue. """
try:
return self._light_map[light_id].light_id
except KeyError: # if light_id is not a valid key
return None