jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-25 22:27:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Re-org emulated_hue and fix google home (#4708)

Browse Source
This commit is contained in:
Paulus Schoutsen 2016-12-04 10:57:48 -08:00 committed by GitHub
parent 1b35f0878e
commit a9be6c36f1
8 changed files with 1170 additions and 995 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

566
homeassistant/components/emulated_hue.py
View File

@ -1,566 +0,0 @@
"""
Support for local control of entities by emulating the Phillips Hue bridge.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/emulated_hue/
"""
import asyncio
import threading
import socket
import logging
import os
import select
from aiohttp import web
import voluptuous as vol
from homeassistant import util, core
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME, SERVICE_TURN_OFF, SERVICE_TURN_ON,
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP,
STATE_ON, STATE_OFF, HTTP_BAD_REQUEST, HTTP_NOT_FOUND,
)
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_SUPPORTED_FEATURES, SUPPORT_BRIGHTNESS
)
from homeassistant.components.http import (
HomeAssistantView, HomeAssistantWSGI
)
import homeassistant.helpers.config_validation as cv
DOMAIN = 'emulated_hue'
_LOGGER = logging.getLogger(__name__)
CONF_HOST_IP = 'host_ip'
CONF_LISTEN_PORT = 'listen_port'
CONF_OFF_MAPS_TO_ON_DOMAINS = 'off_maps_to_on_domains'
CONF_EXPOSE_BY_DEFAULT = 'expose_by_default'
CONF_EXPOSED_DOMAINS = 'exposed_domains'
ATTR_EMULATED_HUE = 'emulated_hue'
ATTR_EMULATED_HUE_NAME = 'emulated_hue_name'
DEFAULT_LISTEN_PORT = 8300
DEFAULT_OFF_MAPS_TO_ON_DOMAINS = ['script', 'scene']
DEFAULT_EXPOSE_BY_DEFAULT = True
DEFAULT_EXPOSED_DOMAINS = [
'switch', 'light', 'group', 'input_boolean', 'media_player', 'fan'
]
HUE_API_STATE_ON = 'on'
HUE_API_STATE_BRI = 'bri'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_HOST_IP): cv.string,
vol.Optional(CONF_LISTEN_PORT, default=DEFAULT_LISTEN_PORT):
vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)),
vol.Optional(CONF_OFF_MAPS_TO_ON_DOMAINS): cv.ensure_list,
vol.Optional(CONF_EXPOSE_BY_DEFAULT): cv.boolean,
vol.Optional(CONF_EXPOSED_DOMAINS): cv.ensure_list
})
}, extra=vol.ALLOW_EXTRA)
def setup(hass, yaml_config):
"""Activate the emulated_hue component."""
config = Config(yaml_config)
server = HomeAssistantWSGI(
hass,
development=False,
server_host=config.host_ip_addr,
server_port=config.listen_port,
api_password=None,
ssl_certificate=None,
ssl_key=None,
cors_origins=None,
use_x_forwarded_for=False,
trusted_networks=[],
login_threshold=0,
is_ban_enabled=False
)
server.register_view(DescriptionXmlView(config))
server.register_view(HueUsernameView)
server.register_view(HueLightsView(config))
upnp_listener = UPNPResponderThread(
config.host_ip_addr, config.listen_port)
@asyncio.coroutine
def stop_emulated_hue_bridge(event):
"""Stop the emulated hue bridge."""
upnp_listener.stop()
yield from server.stop()
@asyncio.coroutine
def start_emulated_hue_bridge(event):
"""Start the emulated hue bridge."""
upnp_listener.start()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP,
stop_emulated_hue_bridge)
yield from server.start()
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_emulated_hue_bridge)
return True
class Config(object):
"""Holds configuration variables for the emulated hue bridge."""
def __init__(self, yaml_config):
"""Initialize the instance."""
conf = yaml_config.get(DOMAIN, {})
# Get the IP address that will be passed to the Echo during discovery
self.host_ip_addr = conf.get(CONF_HOST_IP)
if self.host_ip_addr is None:
self.host_ip_addr = util.get_local_ip()
_LOGGER.warning(
"Listen IP address not specified, auto-detected address is %s",
self.host_ip_addr)
# Get the port that the Hue bridge will listen on
self.listen_port = conf.get(CONF_LISTEN_PORT)
if not isinstance(self.listen_port, int):
self.listen_port = DEFAULT_LISTEN_PORT
_LOGGER.warning(
"Listen port not specified, defaulting to %s",
self.listen_port)
# Get domains that cause both "on" and "off" commands to map to "on"
# This is primarily useful for things like scenes or scripts, which
# don't really have a concept of being off
self.off_maps_to_on_domains = conf.get(CONF_OFF_MAPS_TO_ON_DOMAINS)
if not isinstance(self.off_maps_to_on_domains, list):
self.off_maps_to_on_domains = DEFAULT_OFF_MAPS_TO_ON_DOMAINS
# Get whether or not entities should be exposed by default, or if only
# explicitly marked ones will be exposed
self.expose_by_default = conf.get(
CONF_EXPOSE_BY_DEFAULT, DEFAULT_EXPOSE_BY_DEFAULT)
# Get domains that are exposed by default when expose_by_default is
# True
self.exposed_domains = conf.get(
CONF_EXPOSED_DOMAINS, DEFAULT_EXPOSED_DOMAINS)
class DescriptionXmlView(HomeAssistantView):
"""Handles requests for the description.xml file."""
url = '/description.xml'
name = 'description:xml'
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
@core.callback
def get(self, request):
"""Handle a GET request."""
xml_template = """<?xml version="1.0" encoding="UTF-8" ?>
<root xmlns="urn:schemas-upnp-org:device-1-0">
<specVersion>
<major>1</major>
<minor>0</minor>
</specVersion>
<URLBase>http://{0}:{1}/</URLBase>
<device>
<deviceType>urn:schemas-upnp-org:device:Basic:1</deviceType>
<friendlyName>HASS Bridge ({0})</friendlyName>
<manufacturer>Royal Philips Electronics</manufacturer>
<manufacturerURL>http://www.philips.com</manufacturerURL>
<modelDescription>Philips hue Personal Wireless Lighting</modelDescription>
<modelName>Philips hue bridge 2015</modelName>
<modelNumber>BSB002</modelNumber>
<modelURL>http://www.meethue.com</modelURL>
<serialNumber>1234</serialNumber>
<UDN>uuid:2f402f80-da50-11e1-9b23-001788255acc</UDN>
</device>
</root>
"""
resp_text = xml_template.format(
self.config.host_ip_addr, self.config.listen_port)
return web.Response(text=resp_text, content_type='text/xml')
class HueUsernameView(HomeAssistantView):
"""Handle requests to create a username for the emulated hue bridge."""
url = '/api'
name = 'hue:api'
extra_urls = ['/api/']
requires_auth = False
@asyncio.coroutine
def post(self, request):
"""Handle a POST request."""
try:
data = yield from request.json()
except ValueError:
return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)
if 'devicetype' not in data:
return self.json_message('devicetype not specified',
HTTP_BAD_REQUEST)
return self.json([{'success': {'username': '12345678901234567890'}}])
class HueLightsView(HomeAssistantView):
"""Handle requests for getting and setting info about entities."""
url = '/api/{username}/lights'
name = 'api:username:lights'
extra_urls = ['/api/{username}/lights/{entity_id}',
'/api/{username}/lights/{entity_id}/state']
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
self.cached_states = {}
@core.callback
def get(self, request, username, entity_id=None):
"""Handle a GET request."""
hass = request.app['hass']
if entity_id is None:
return self.async_get_lights_list(hass)
if not request.path.endswith('state'):
return self.async_get_light_state(hass, entity_id)
return web.Response(text="Method not allowed", status=405)
@asyncio.coroutine
def put(self, request, username, entity_id=None):
"""Handle a PUT request."""
hass = request.app['hass']
if not request.path.endswith('state'):
return web.Response(text="Method not allowed", status=405)
if entity_id and hass.states.get(entity_id) is None:
return self.json_message('Entity not found', HTTP_NOT_FOUND)
try:
json_data = yield from request.json()
except ValueError:
return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)
result = yield from self.async_put_light_state(hass, json_data,
entity_id)
return result
@core.callback
def async_get_lights_list(self, hass):
"""Process a request to get the list of available lights."""
json_response = {}
for entity in hass.states.async_all():
if self.is_entity_exposed(entity):
json_response[entity.entity_id] = entity_to_json(entity)
return self.json(json_response)
@core.callback
def async_get_light_state(self, hass, entity_id):
"""Process a request to get the state of an individual light."""
entity = hass.states.get(entity_id)
if entity is None or not self.is_entity_exposed(entity):
return web.Response(text="Entity not found", status=404)
cached_state = self.cached_states.get(entity_id, None)
if cached_state is None:
final_state = entity.state == STATE_ON
final_brightness = entity.attributes.get(
ATTR_BRIGHTNESS, 255 if final_state else 0)
else:
final_state, final_brightness = cached_state
json_response = entity_to_json(entity, final_state, final_brightness)
return self.json(json_response)
@asyncio.coroutine
def async_put_light_state(self, hass, request_json, entity_id):
"""Process a request to set the state of an individual light."""
config = self.config
# Retrieve the entity from the state machine
entity = hass.states.get(entity_id)
if entity is None:
return web.Response(text="Entity not found", status=404)
if not self.is_entity_exposed(entity):
return web.Response(text="Entity not found", status=404)
# Parse the request into requested "on" status and brightness
parsed = parse_hue_api_put_light_body(request_json, entity)
if parsed is None:
return web.Response(text="Bad request", status=400)
result, brightness = parsed
# Convert the resulting "on" status into the service we need to call
service = SERVICE_TURN_ON if result else SERVICE_TURN_OFF
# Construct what we need to send to the service
data = {ATTR_ENTITY_ID: entity_id}
# If the requested entity is a script add some variables
if entity.domain.lower() == "script":
data['variables'] = {
'requested_state': STATE_ON if result else STATE_OFF
}
if brightness is not None:
data['variables']['requested_level'] = brightness
elif brightness is not None:
data[ATTR_BRIGHTNESS] = brightness
if entity.domain.lower() in config.off_maps_to_on_domains:
# Map the off command to on
service = SERVICE_TURN_ON
# Caching is required because things like scripts and scenes won't
# report as "off" to Alexa if an "off" command is received, because
# they'll map to "on". Thus, instead of reporting its actual
# status, we report what Alexa will want to see, which is the same
# as the actual requested command.
self.cached_states[entity_id] = (result, brightness)
# Perform the requested action
yield from hass.services.async_call(core.DOMAIN, service, data,
blocking=True)
json_response = \
[create_hue_success_response(entity_id, HUE_API_STATE_ON, result)]
if brightness is not None:
json_response.append(create_hue_success_response(
entity_id, HUE_API_STATE_BRI, brightness))
return self.json(json_response)
def is_entity_exposed(self, entity):
"""Determine if an entity should be exposed on the emulated bridge.
Async friendly.
"""
config = self.config
if entity.attributes.get('view') is not None:
# Ignore entities that are views
return False
domain = entity.domain.lower()
explicit_expose = entity.attributes.get(ATTR_EMULATED_HUE, None)
domain_exposed_by_default = \
config.expose_by_default and domain in config.exposed_domains
# Expose an entity if the entity's domain is exposed by default and
# the configuration doesn't explicitly exclude it from being
# exposed, or if the entity is explicitly exposed
is_default_exposed = \
domain_exposed_by_default and explicit_expose is not False
return is_default_exposed or explicit_expose
def parse_hue_api_put_light_body(request_json, entity):
"""Parse the body of a request to change the state of a light."""
if HUE_API_STATE_ON in request_json:
if not isinstance(request_json[HUE_API_STATE_ON], bool):
return None
if request_json['on']:
# Echo requested device be turned on
brightness = None
report_brightness = False
result = True
else:
# Echo requested device be turned off
brightness = None
report_brightness = False
result = False
if HUE_API_STATE_BRI in request_json:
# Make sure the entity actually supports brightness
entity_features = entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if (entity_features & SUPPORT_BRIGHTNESS) == SUPPORT_BRIGHTNESS:
try:
# Clamp brightness from 0 to 255
brightness = \
max(0, min(int(request_json[HUE_API_STATE_BRI]), 255))
except ValueError:
return None
report_brightness = True
result = (brightness > 0)
elif entity.domain.lower() == "script":
# Convert 0-255 to 0-100
level = int(request_json[HUE_API_STATE_BRI]) / 255 * 100
brightness = round(level)
report_brightness = True
result = True
return (result, brightness) if report_brightness else (result, None)
def entity_to_json(entity, is_on=None, brightness=None):
"""Convert an entity to its Hue bridge JSON representation."""
if is_on is None:
is_on = entity.state == STATE_ON
if brightness is None:
brightness = 255 if is_on else 0
name = entity.attributes.get(
ATTR_EMULATED_HUE_NAME, entity.attributes[ATTR_FRIENDLY_NAME])
return {
'state':
{
HUE_API_STATE_ON: is_on,
HUE_API_STATE_BRI: brightness,
'reachable': True
},
'type': 'Dimmable light',
'name': name,
'modelid': 'HASS123',
'uniqueid': entity.entity_id,
'swversion': '123'
}
def create_hue_success_response(entity_id, attr, value):
"""Create a success response for an attribute set on a light."""
success_key = '/lights/{}/state/{}'.format(entity_id, attr)
return {'success': {success_key: value}}
class UPNPResponderThread(threading.Thread):
"""Handle responding to UPNP/SSDP discovery requests."""
_interrupted = False
def __init__(self, host_ip_addr, listen_port):
"""Initialize the class."""
threading.Thread.__init__(self)
self.host_ip_addr = host_ip_addr
self.listen_port = listen_port
# Note that the double newline at the end of
# this string is required per the SSDP spec
resp_template = """HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{0}:{1}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/0.1
ST: urn:schemas-upnp-org:device:basic:1
USN: uuid:Socket-1_0-221438K0100073::urn:schemas-upnp-org:device:basic:1
"""
self.upnp_response = resp_template.format(host_ip_addr, listen_port) \
.replace("\n", "\r\n") \
.encode('utf-8')
# Set up a pipe for signaling to the receiver that it's time to
# shutdown. Essentially, we place the SSDP socket into nonblocking
# mode and use select() to wait for data to arrive on either the SSDP
# socket or the pipe. If data arrives on either one, select() returns
# and tells us which filenos have data ready to read.
#
# When we want to stop the responder, we write data to the pipe, which
# causes the select() to return and indicate that said pipe has data
# ready to be read, which indicates to us that the responder needs to
# be shutdown.
self._interrupted_read_pipe, self._interrupted_write_pipe = os.pipe()
def run(self):
"""Run the server."""
# Listen for UDP port 1900 packets sent to SSDP multicast address
ssdp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ssdp_socket.setblocking(False)
# Required for receiving multicast
ssdp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ssdp_socket.setsockopt(
socket.SOL_IP,
socket.IP_MULTICAST_IF,
socket.inet_aton(self.host_ip_addr))
ssdp_socket.setsockopt(
socket.SOL_IP,
socket.IP_ADD_MEMBERSHIP,
socket.inet_aton("239.255.255.250") +
socket.inet_aton(self.host_ip_addr))
ssdp_socket.bind(("239.255.255.250", 1900))
while True:
if self._interrupted:
clean_socket_close(ssdp_socket)
return
try:
read, _, _ = select.select(
[self._interrupted_read_pipe, ssdp_socket], [],
[ssdp_socket])
if self._interrupted_read_pipe in read:
# Implies self._interrupted is True
clean_socket_close(ssdp_socket)
return
elif ssdp_socket in read:
data, addr = ssdp_socket.recvfrom(1024)
else:
continue
except socket.error as ex:
if self._interrupted:
clean_socket_close(ssdp_socket)
return
_LOGGER.error("UPNP Responder socket exception occured: %s",
ex.__str__)
if "M-SEARCH" in data.decode('utf-8'):
# SSDP M-SEARCH method received, respond to it with our info
resp_socket = socket.socket(
socket.AF_INET, socket.SOCK_DGRAM)
resp_socket.sendto(self.upnp_response, addr)
resp_socket.close()
def stop(self):
"""Stop the server."""
# Request for server
self._interrupted = True
os.write(self._interrupted_write_pipe, bytes([0]))
self.join()
def clean_socket_close(sock):
"""Close a socket connection and logs its closure."""
_LOGGER.info("UPNP responder shutting down.")
sock.close()

198
homeassistant/components/emulated_hue/__init__.py Normal file
View File

@ -0,0 +1,198 @@
"""
Support for local control of entities by emulating the Phillips Hue bridge.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/emulated_hue/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant import util
from homeassistant.const import (
EVENT_HOMEASSISTANT_START, EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.components.http import HomeAssistantWSGI
import homeassistant.helpers.config_validation as cv
from .hue_api import (
HueUsernameView, HueAllLightsStateView, HueOneLightStateView,
HueOneLightChangeView)
from .upnp import DescriptionXmlView, UPNPResponderThread
DOMAIN = 'emulated_hue'
_LOGGER = logging.getLogger(__name__)
CONF_HOST_IP = 'host_ip'
CONF_LISTEN_PORT = 'listen_port'
CONF_OFF_MAPS_TO_ON_DOMAINS = 'off_maps_to_on_domains'
CONF_EXPOSE_BY_DEFAULT = 'expose_by_default'
CONF_EXPOSED_DOMAINS = 'exposed_domains'
CONF_TYPE = 'type'
TYPE_ALEXA = 'alexa'
TYPE_GOOGLE = 'google_home'
DEFAULT_LISTEN_PORT = 8300
DEFAULT_OFF_MAPS_TO_ON_DOMAINS = ['script', 'scene']
DEFAULT_EXPOSE_BY_DEFAULT = True
DEFAULT_EXPOSED_DOMAINS = [
'switch', 'light', 'group', 'input_boolean', 'media_player', 'fan'
]
DEFAULT_TYPE = TYPE_ALEXA
CONFIG_SCHEMA = vol.Schema({
DOMAIN: vol.Schema({
vol.Optional(CONF_HOST_IP): cv.string,
vol.Optional(CONF_LISTEN_PORT, default=DEFAULT_LISTEN_PORT):
vol.All(vol.Coerce(int), vol.Range(min=1, max=65535)),
vol.Optional(CONF_OFF_MAPS_TO_ON_DOMAINS): cv.ensure_list,
vol.Optional(CONF_EXPOSE_BY_DEFAULT): cv.boolean,
vol.Optional(CONF_EXPOSED_DOMAINS): cv.ensure_list,
vol.Optional(CONF_TYPE, default=DEFAULT_TYPE):
vol.Any(TYPE_ALEXA, TYPE_GOOGLE)
})
}, extra=vol.ALLOW_EXTRA)
ATTR_EMULATED_HUE = 'emulated_hue'
def setup(hass, yaml_config):
"""Activate the emulated_hue component."""
config = Config(yaml_config.get(DOMAIN, {}))
server = HomeAssistantWSGI(
hass,
development=False,
server_host=config.host_ip_addr,
server_port=config.listen_port,
api_password=None,
ssl_certificate=None,
ssl_key=None,
cors_origins=None,
use_x_forwarded_for=False,
trusted_networks=[],
login_threshold=0,
is_ban_enabled=False
)
server.register_view(DescriptionXmlView(config))
server.register_view(HueUsernameView)
server.register_view(HueAllLightsStateView(config))
server.register_view(HueOneLightStateView(config))
server.register_view(HueOneLightChangeView(config))
upnp_listener = UPNPResponderThread(
config.host_ip_addr, config.listen_port)
@asyncio.coroutine
def stop_emulated_hue_bridge(event):
"""Stop the emulated hue bridge."""
upnp_listener.stop()
yield from server.stop()
@asyncio.coroutine
def start_emulated_hue_bridge(event):
"""Start the emulated hue bridge."""
upnp_listener.start()
yield from server.start()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP,
stop_emulated_hue_bridge)
hass.bus.listen_once(EVENT_HOMEASSISTANT_START, start_emulated_hue_bridge)
return True
class Config(object):
"""Holds configuration variables for the emulated hue bridge."""
def __init__(self, conf):
"""Initialize the instance."""
self.type = conf.get(CONF_TYPE)
self.numbers = {}
self.cached_states = {}
# Get the IP address that will be passed to the Echo during discovery
self.host_ip_addr = conf.get(CONF_HOST_IP)
if self.host_ip_addr is None:
self.host_ip_addr = util.get_local_ip()
_LOGGER.warning(
"Listen IP address not specified, auto-detected address is %s",
self.host_ip_addr)
# Get the port that the Hue bridge will listen on
self.listen_port = conf.get(CONF_LISTEN_PORT)
if not isinstance(self.listen_port, int):
self.listen_port = DEFAULT_LISTEN_PORT
_LOGGER.warning(
"Listen port not specified, defaulting to %s",
self.listen_port)
if self.type == TYPE_GOOGLE and self.listen_port != 80:
_LOGGER.warning('When targetting Google Home, listening port has '
'to be port 80')
# Get domains that cause both "on" and "off" commands to map to "on"
# This is primarily useful for things like scenes or scripts, which
# don't really have a concept of being off
self.off_maps_to_on_domains = conf.get(CONF_OFF_MAPS_TO_ON_DOMAINS)
if not isinstance(self.off_maps_to_on_domains, list):
self.off_maps_to_on_domains = DEFAULT_OFF_MAPS_TO_ON_DOMAINS
# Get whether or not entities should be exposed by default, or if only
# explicitly marked ones will be exposed
self.expose_by_default = conf.get(
CONF_EXPOSE_BY_DEFAULT, DEFAULT_EXPOSE_BY_DEFAULT)
# Get domains that are exposed by default when expose_by_default is
# True
self.exposed_domains = conf.get(
CONF_EXPOSED_DOMAINS, DEFAULT_EXPOSED_DOMAINS)
def entity_id_to_number(self, entity_id):
"""Get a unique number for the entity id."""
if self.type == TYPE_ALEXA:
return entity_id
# Google Home
for number, ent_id in self.numbers.items():
if entity_id == ent_id:
return number
number = str(len(self.numbers) + 1)
self.numbers[number] = entity_id
return number
def number_to_entity_id(self, number):
"""Convert unique number to entity id."""
if self.type == TYPE_ALEXA:
return number
# Google Home
assert isinstance(number, str)
return self.numbers.get(number)
def is_entity_exposed(self, entity):
"""Determine if an entity should be exposed on the emulated bridge.
Async friendly.
"""
if entity.attributes.get('view') is not None:
# Ignore entities that are views
return False
domain = entity.domain.lower()
explicit_expose = entity.attributes.get(ATTR_EMULATED_HUE, None)
domain_exposed_by_default = \
self.expose_by_default and domain in self.exposed_domains
# Expose an entity if the entity's domain is exposed by default and
# the configuration doesn't explicitly exclude it from being
# exposed, or if the entity is explicitly exposed
is_default_exposed = \
domain_exposed_by_default and explicit_expose is not False
return is_default_exposed or explicit_expose

275
homeassistant/components/emulated_hue/hue_api.py Normal file
View File

@ -0,0 +1,275 @@
"""Provides a Hue API to control Home Assistant."""
import asyncio
import logging
from aiohttp import web
from homeassistant import core
from homeassistant.const import (
ATTR_ENTITY_ID, ATTR_FRIENDLY_NAME, SERVICE_TURN_OFF, SERVICE_TURN_ON,
STATE_ON, STATE_OFF, HTTP_BAD_REQUEST, HTTP_NOT_FOUND,
)
from homeassistant.components.light import (
ATTR_BRIGHTNESS, ATTR_SUPPORTED_FEATURES, SUPPORT_BRIGHTNESS
)
from homeassistant.components.http import HomeAssistantView
_LOGGER = logging.getLogger(__name__)
ATTR_EMULATED_HUE = 'emulated_hue'
ATTR_EMULATED_HUE_NAME = 'emulated_hue_name'
HUE_API_STATE_ON = 'on'
HUE_API_STATE_BRI = 'bri'
class HueUsernameView(HomeAssistantView):
"""Handle requests to create a username for the emulated hue bridge."""
url = '/api'
name = 'emulated_hue:api:create_username'
extra_urls = ['/api/']
requires_auth = False
@asyncio.coroutine
def post(self, request):
"""Handle a POST request."""
try:
data = yield from request.json()
except ValueError:
return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)
if 'devicetype' not in data:
return self.json_message('devicetype not specified',
HTTP_BAD_REQUEST)
return self.json([{'success': {'username': '12345678901234567890'}}])
class HueAllLightsStateView(HomeAssistantView):
"""Handle requests for getting and setting info about entities."""
url = '/api/{username}/lights'
name = 'emulated_hue:lights:state'
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
@core.callback
def get(self, request, username):
"""Process a request to get the list of available lights."""
hass = request.app['hass']
json_response = {}
for entity in hass.states.async_all():
if self.config.is_entity_exposed(entity):
number = self.config.entity_id_to_number(entity.entity_id)
json_response[number] = entity_to_json(entity)
return self.json(json_response)
class HueOneLightStateView(HomeAssistantView):
"""Handle requests for getting and setting info about entities."""
url = '/api/{username}/lights/{entity_id}'
name = 'emulated_hue:light:state'
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
@core.callback
def get(self, request, username, entity_id=None):
"""Process a request to get the state of an individual light."""
hass = request.app['hass']
entity_id = self.config.number_to_entity_id(entity_id)
entity = hass.states.get(entity_id)
if entity is None:
_LOGGER.error('Entity not found: %s', entity_id)
return web.Response(text="Entity not found", status=404)
if not self.config.is_entity_exposed(entity):
_LOGGER.error('Entity not exposed: %s', entity_id)
return web.Response(text="Entity not exposed", status=404)
cached_state = self.config.cached_states.get(entity_id, None)
if cached_state is None:
final_state = entity.state == STATE_ON
final_brightness = entity.attributes.get(
ATTR_BRIGHTNESS, 255 if final_state else 0)
else:
final_state, final_brightness = cached_state
json_response = entity_to_json(entity, final_state, final_brightness)
return self.json(json_response)
class HueOneLightChangeView(HomeAssistantView):
"""Handle requests for getting and setting info about entities."""
url = '/api/{username}/lights/{entity_number}/state'
name = 'emulated_hue:light:state'
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
@asyncio.coroutine
def put(self, request, username, entity_number):
"""Process a request to set the state of an individual light."""
config = self.config
hass = request.app['hass']
entity_id = config.number_to_entity_id(entity_number)
if entity_id is None:
_LOGGER.error('Unknown entity number: %s', entity_number)
return self.json_message('Entity not found', HTTP_NOT_FOUND)
entity = hass.states.get(entity_id)
if entity is None:
_LOGGER.error('Entity not found: %s', entity_id)
return self.json_message('Entity not found', HTTP_NOT_FOUND)
if not config.is_entity_exposed(entity):
_LOGGER.error('Entity not exposed: %s', entity_id)
return web.Response(text="Entity not exposed", status=404)
try:
request_json = yield from request.json()
except ValueError:
_LOGGER.error('Received invalid json')
return self.json_message('Invalid JSON', HTTP_BAD_REQUEST)
# Parse the request into requested "on" status and brightness
parsed = parse_hue_api_put_light_body(request_json, entity)
if parsed is None:
_LOGGER.error('Unable to parse data: %s', request_json)
return web.Response(text="Bad request", status=400)
result, brightness = parsed
# Convert the resulting "on" status into the service we need to call
service = SERVICE_TURN_ON if result else SERVICE_TURN_OFF
# Construct what we need to send to the service
data = {ATTR_ENTITY_ID: entity_id}
# If the requested entity is a script add some variables
if entity.domain == "script":
data['variables'] = {
'requested_state': STATE_ON if result else STATE_OFF
}
if brightness is not None:
data['variables']['requested_level'] = brightness
elif brightness is not None:
data[ATTR_BRIGHTNESS] = brightness
if entity.domain in config.off_maps_to_on_domains:
# Map the off command to on
service = SERVICE_TURN_ON
# Caching is required because things like scripts and scenes won't
# report as "off" to Alexa if an "off" command is received, because
# they'll map to "on". Thus, instead of reporting its actual
# status, we report what Alexa will want to see, which is the same
# as the actual requested command.
config.cached_states[entity_id] = (result, brightness)
# Perform the requested action
yield from hass.services.async_call(core.DOMAIN, service, data,
blocking=True)
json_response = \
[create_hue_success_response(entity_id, HUE_API_STATE_ON, result)]
if brightness is not None:
json_response.append(create_hue_success_response(
entity_id, HUE_API_STATE_BRI, brightness))
return self.json(json_response)
def parse_hue_api_put_light_body(request_json, entity):
"""Parse the body of a request to change the state of a light."""
if HUE_API_STATE_ON in request_json:
if not isinstance(request_json[HUE_API_STATE_ON], bool):
return None
if request_json['on']:
# Echo requested device be turned on
brightness = None
report_brightness = False
result = True
else:
# Echo requested device be turned off
brightness = None
report_brightness = False
result = False
if HUE_API_STATE_BRI in request_json:
# Make sure the entity actually supports brightness
entity_features = entity.attributes.get(ATTR_SUPPORTED_FEATURES, 0)
if (entity_features & SUPPORT_BRIGHTNESS) == SUPPORT_BRIGHTNESS:
try:
# Clamp brightness from 0 to 255
brightness = \
max(0, min(int(request_json[HUE_API_STATE_BRI]), 255))
except ValueError:
return None
report_brightness = True
result = (brightness > 0)
elif entity.domain.lower() == "script":
# Convert 0-255 to 0-100
level = int(request_json[HUE_API_STATE_BRI]) / 255 * 100
brightness = round(level)
report_brightness = True
result = True
return (result, brightness) if report_brightness else (result, None)
def entity_to_json(entity, is_on=None, brightness=None):
"""Convert an entity to its Hue bridge JSON representation."""
if is_on is None:
is_on = entity.state == STATE_ON
if brightness is None:
brightness = 255 if is_on else 0
name = entity.attributes.get(
ATTR_EMULATED_HUE_NAME, entity.attributes[ATTR_FRIENDLY_NAME])
return {
'state':
{
HUE_API_STATE_ON: is_on,
HUE_API_STATE_BRI: brightness,
'reachable': True
},
'type': 'Dimmable light',
'name': name,
'modelid': 'HASS123',
'uniqueid': entity.entity_id,
'swversion': '123'
}
def create_hue_success_response(entity_id, attr, value):
"""Create a success response for an attribute set on a light."""
success_key = '/lights/{}/state/{}'.format(entity_id, attr)
return {'success': {success_key: value}}

166
homeassistant/components/emulated_hue/upnp.py Normal file
View File

@ -0,0 +1,166 @@
"""Provides a UPNP discovery method that mimicks Hue hubs."""
import threading
import socket
import logging
import os
import select
from aiohttp import web
from homeassistant import core
from homeassistant.components.http import HomeAssistantView
_LOGGER = logging.getLogger(__name__)
class DescriptionXmlView(HomeAssistantView):
"""Handles requests for the description.xml file."""
url = '/description.xml'
name = 'description:xml'
requires_auth = False
def __init__(self, config):
"""Initialize the instance of the view."""
self.config = config
@core.callback
def get(self, request):
"""Handle a GET request."""
xml_template = """<?xml version="1.0" encoding="UTF-8" ?>
<root xmlns="urn:schemas-upnp-org:device-1-0">
<specVersion>
<major>1</major>
<minor>0</minor>
</specVersion>
<URLBase>http://{0}:{1}/</URLBase>
<device>
<deviceType>urn:schemas-upnp-org:device:Basic:1</deviceType>
<friendlyName>HASS Bridge ({0})</friendlyName>
<manufacturer>Royal Philips Electronics</manufacturer>
<manufacturerURL>http://www.philips.com</manufacturerURL>
<modelDescription>Philips hue Personal Wireless Lighting</modelDescription>
<modelName>Philips hue bridge 2015</modelName>
<modelNumber>BSB002</modelNumber>
<modelURL>http://www.meethue.com</modelURL>
<serialNumber>1234</serialNumber>
<UDN>uuid:2f402f80-da50-11e1-9b23-001788255acc</UDN>
</device>
</root>
"""
resp_text = xml_template.format(
self.config.host_ip_addr, self.config.listen_port)
return web.Response(text=resp_text, content_type='text/xml')
class UPNPResponderThread(threading.Thread):
"""Handle responding to UPNP/SSDP discovery requests."""
_interrupted = False
def __init__(self, host_ip_addr, listen_port):
"""Initialize the class."""
threading.Thread.__init__(self)
self.host_ip_addr = host_ip_addr
self.listen_port = listen_port
# Note that the double newline at the end of
# this string is required per the SSDP spec
resp_template = """HTTP/1.1 200 OK
CACHE-CONTROL: max-age=60
EXT:
LOCATION: http://{0}:{1}/description.xml
SERVER: FreeRTOS/6.0.5, UPnP/1.0, IpBridge/0.1
ST: urn:schemas-upnp-org:device:basic:1
USN: uuid:Socket-1_0-221438K0100073::urn:schemas-upnp-org:device:basic:1
"""
self.upnp_response = resp_template.format(host_ip_addr, listen_port) \
.replace("\n", "\r\n") \
.encode('utf-8')
# Set up a pipe for signaling to the receiver that it's time to
# shutdown. Essentially, we place the SSDP socket into nonblocking
# mode and use select() to wait for data to arrive on either the SSDP
# socket or the pipe. If data arrives on either one, select() returns
# and tells us which filenos have data ready to read.
#
# When we want to stop the responder, we write data to the pipe, which
# causes the select() to return and indicate that said pipe has data
# ready to be read, which indicates to us that the responder needs to
# be shutdown.
self._interrupted_read_pipe, self._interrupted_write_pipe = os.pipe()
def run(self):
"""Run the server."""
# Listen for UDP port 1900 packets sent to SSDP multicast address
ssdp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
ssdp_socket.setblocking(False)
# Required for receiving multicast
ssdp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
ssdp_socket.setsockopt(
socket.SOL_IP,
socket.IP_MULTICAST_IF,
socket.inet_aton(self.host_ip_addr))
ssdp_socket.setsockopt(
socket.SOL_IP,
socket.IP_ADD_MEMBERSHIP,
socket.inet_aton("239.255.255.250") +
socket.inet_aton(self.host_ip_addr))
ssdp_socket.bind(("239.255.255.250", 1900))
while True:
if self._interrupted:
clean_socket_close(ssdp_socket)
return
try:
read, _, _ = select.select(
[self._interrupted_read_pipe, ssdp_socket], [],
[ssdp_socket])
if self._interrupted_read_pipe in read:
# Implies self._interrupted is True
clean_socket_close(ssdp_socket)
return
elif ssdp_socket in read:
data, addr = ssdp_socket.recvfrom(1024)
else:
continue
except socket.error as ex:
if self._interrupted:
clean_socket_close(ssdp_socket)
return
_LOGGER.error("UPNP Responder socket exception occured: %s",
ex.__str__)
if "M-SEARCH" in data.decode('utf-8'):
# SSDP M-SEARCH method received, respond to it with our info
resp_socket = socket.socket(
socket.AF_INET, socket.SOCK_DGRAM)
resp_socket.sendto(self.upnp_response, addr)
resp_socket.close()
def stop(self):
"""Stop the server."""
# Request for server
self._interrupted = True
os.write(self._interrupted_write_pipe, bytes([0]))
self.join()
def clean_socket_close(sock):
"""Close a socket connection and logs its closure."""
_LOGGER.info("UPNP responder shutting down.")
sock.close()

1
tests/components/emulated_hue/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Tests for emulated_hue."""

784
tests/components/test_emulated_hue.py → tests/components/emulated_hue/test_hue_api.py Executable file → Normal file
View File

@ -1,429 +1,355 @@
"""The tests for the emulated Hue component.""" """The tests for the emulated Hue component."""
import json import json
import unittest import unittest
import requests from unittest.mock import patch
import requests
from homeassistant import bootstrap, const, core
import homeassistant.components as core_components from homeassistant import bootstrap, const, core
from homeassistant.components import emulated_hue, http, light, script import homeassistant.components as core_components
from homeassistant.const import STATE_ON, STATE_OFF from homeassistant.components import emulated_hue, http, light, script
from homeassistant.components.emulated_hue import ( from homeassistant.const import STATE_ON, STATE_OFF
HUE_API_STATE_ON, HUE_API_STATE_BRI) from homeassistant.components.emulated_hue.hue_api import (
from homeassistant.util.async import run_coroutine_threadsafe HUE_API_STATE_ON, HUE_API_STATE_BRI)
from homeassistant.util.async import run_coroutine_threadsafe
from tests.common import get_test_instance_port, get_test_home_assistant
from tests.common import get_test_instance_port, get_test_home_assistant
HTTP_SERVER_PORT = get_test_instance_port()
BRIDGE_SERVER_PORT = get_test_instance_port() HTTP_SERVER_PORT = get_test_instance_port()
BRIDGE_SERVER_PORT = get_test_instance_port()
BRIDGE_URL_BASE = 'http://127.0.0.1:{}'.format(BRIDGE_SERVER_PORT) + '{}'
JSON_HEADERS = {const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON} BRIDGE_URL_BASE = 'http://127.0.0.1:{}'.format(BRIDGE_SERVER_PORT) + '{}'
JSON_HEADERS = {const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON}
def setup_hass_instance(emulated_hue_config):
"""Set up the Home Assistant instance to test.""" class TestEmulatedHueExposedByDefault(unittest.TestCase):
hass = get_test_home_assistant() """Test class for emulated hue component."""
# We need to do this to get access to homeassistant/turn_(on,off) @classmethod
run_coroutine_threadsafe( def setUpClass(cls):
core_components.async_setup(hass, {core.DOMAIN: {}}), hass.loop """Setup the class."""
).result() cls.hass = hass = get_test_home_assistant()
bootstrap.setup_component( # We need to do this to get access to homeassistant/turn_(on,off)
hass, http.DOMAIN, run_coroutine_threadsafe(
{http.DOMAIN: {http.CONF_SERVER_PORT: HTTP_SERVER_PORT}}) core_components.async_setup(hass, {core.DOMAIN: {}}), hass.loop
).result()
bootstrap.setup_component(hass, emulated_hue.DOMAIN, emulated_hue_config)
bootstrap.setup_component(
return hass hass, http.DOMAIN,
{http.DOMAIN: {http.CONF_SERVER_PORT: HTTP_SERVER_PORT}})
def start_hass_instance(hass): with patch('homeassistant.components'
"""Start the Home Assistant instance to test.""" '.emulated_hue.UPNPResponderThread'):
hass.start() bootstrap.setup_component(hass, emulated_hue.DOMAIN, {
emulated_hue.DOMAIN: {
emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT,
class TestEmulatedHue(unittest.TestCase): emulated_hue.CONF_EXPOSE_BY_DEFAULT: True
"""Test the emulated Hue component.""" }
})
hass = None
bootstrap.setup_component(cls.hass, light.DOMAIN, {
@classmethod 'light': [
def setUpClass(cls): {
"""Setup the class.""" 'platform': 'demo',
cls.hass = setup_hass_instance({ }
emulated_hue.DOMAIN: { ]
emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT })
}})
bootstrap.setup_component(cls.hass, script.DOMAIN, {
start_hass_instance(cls.hass) 'script': {
'set_kitchen_light': {
@classmethod 'sequence': [
def tearDownClass(cls): {
"""Stop the class.""" 'service_template':
cls.hass.stop() "light.turn_{{ requested_state }}",
'data_template': {
def test_description_xml(self): 'entity_id': 'light.kitchen_lights',
"""Test the description.""" 'brightness': "{{ requested_level }}"
import xml.etree.ElementTree as ET }
}
result = requests.get( ]
BRIDGE_URL_BASE.format('/description.xml'), timeout=5) }
}
self.assertEqual(result.status_code, 200) })
self.assertTrue('text/xml' in result.headers['content-type'])
cls.hass.start()
# Make sure the XML is parsable
# pylint: disable=bare-except # Kitchen light is explicitly excluded from being exposed
try: kitchen_light_entity = cls.hass.states.get('light.kitchen_lights')
ET.fromstring(result.text) attrs = dict(kitchen_light_entity.attributes)
except: attrs[emulated_hue.ATTR_EMULATED_HUE] = False
self.fail('description.xml is not valid XML!') cls.hass.states.set(
kitchen_light_entity.entity_id, kitchen_light_entity.state,
def test_create_username(self): attributes=attrs)
"""Test the creation of an username."""
request_json = {'devicetype': 'my_device'} # Expose the script
script_entity = cls.hass.states.get('script.set_kitchen_light')
result = requests.post( attrs = dict(script_entity.attributes)
BRIDGE_URL_BASE.format('/api'), data=json.dumps(request_json), attrs[emulated_hue.ATTR_EMULATED_HUE] = True
timeout=5) cls.hass.states.set(
script_entity.entity_id, script_entity.state, attributes=attrs
self.assertEqual(result.status_code, 200) )
self.assertTrue('application/json' in result.headers['content-type'])
@classmethod
resp_json = result.json() def tearDownClass(cls):
success_json = resp_json[0] """Stop the class."""
cls.hass.stop()
self.assertTrue('success' in success_json)
self.assertTrue('username' in success_json['success']) def test_discover_lights(self):
"""Test the discovery of lights."""
def test_valid_username_request(self): result = requests.get(
"""Test request with a valid username.""" BRIDGE_URL_BASE.format('/api/username/lights'), timeout=5)
request_json = {'invalid_key': 'my_device'}
self.assertEqual(result.status_code, 200)
result = requests.post( self.assertTrue('application/json' in result.headers['content-type'])
BRIDGE_URL_BASE.format('/api'), data=json.dumps(request_json),
timeout=5) result_json = result.json()
self.assertEqual(result.status_code, 400) # Make sure the lights we added to the config are there
self.assertTrue('light.ceiling_lights' in result_json)
self.assertTrue('light.bed_light' in result_json)
class TestEmulatedHueExposedByDefault(unittest.TestCase): self.assertTrue('script.set_kitchen_light' in result_json)
"""Test class for emulated hue component.""" self.assertTrue('light.kitchen_lights' not in result_json)
@classmethod def test_get_light_state(self):
def setUpClass(cls): """Test the getting of light state."""
"""Setup the class.""" # Turn office light on and set to 127 brightness
cls.hass = setup_hass_instance({ self.hass.services.call(
emulated_hue.DOMAIN: { light.DOMAIN, const.SERVICE_TURN_ON,
emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT, {
emulated_hue.CONF_EXPOSE_BY_DEFAULT: True const.ATTR_ENTITY_ID: 'light.ceiling_lights',
} light.ATTR_BRIGHTNESS: 127
}) },
blocking=True)
bootstrap.setup_component(cls.hass, light.DOMAIN, {
'light': [ office_json = self.perform_get_light_state('light.ceiling_lights', 200)
{
'platform': 'demo', self.assertEqual(office_json['state'][HUE_API_STATE_ON], True)
} self.assertEqual(office_json['state'][HUE_API_STATE_BRI], 127)
]
}) # Turn bedroom light off
self.hass.services.call(
bootstrap.setup_component(cls.hass, script.DOMAIN, { light.DOMAIN, const.SERVICE_TURN_OFF,
'script': { {
'set_kitchen_light': { const.ATTR_ENTITY_ID: 'light.bed_light'
'sequence': [ },
{ blocking=True)
'service_template':
"light.turn_{{ requested_state }}", bedroom_json = self.perform_get_light_state('light.bed_light', 200)
'data_template': {
'entity_id': 'light.kitchen_lights', self.assertEqual(bedroom_json['state'][HUE_API_STATE_ON], False)
'brightness': "{{ requested_level }}" self.assertEqual(bedroom_json['state'][HUE_API_STATE_BRI], 0)
}
} # Make sure kitchen light isn't accessible
] kitchen_url = '/api/username/lights/{}'.format('light.kitchen_lights')
} kitchen_result = requests.get(
} BRIDGE_URL_BASE.format(kitchen_url), timeout=5)
})
self.assertEqual(kitchen_result.status_code, 404)
start_hass_instance(cls.hass)
def test_put_light_state(self):
# Kitchen light is explicitly excluded from being exposed """Test the seeting of light states."""
kitchen_light_entity = cls.hass.states.get('light.kitchen_lights') self.perform_put_test_on_ceiling_lights()
attrs = dict(kitchen_light_entity.attributes)
attrs[emulated_hue.ATTR_EMULATED_HUE] = False # Turn the bedroom light on first
cls.hass.states.set( self.hass.services.call(
kitchen_light_entity.entity_id, kitchen_light_entity.state, light.DOMAIN, const.SERVICE_TURN_ON,
attributes=attrs) {const.ATTR_ENTITY_ID: 'light.bed_light',
light.ATTR_BRIGHTNESS: 153},
# Expose the script blocking=True)
script_entity = cls.hass.states.get('script.set_kitchen_light')
attrs = dict(script_entity.attributes) bed_light = self.hass.states.get('light.bed_light')
attrs[emulated_hue.ATTR_EMULATED_HUE] = True self.assertEqual(bed_light.state, STATE_ON)
cls.hass.states.set( self.assertEqual(bed_light.attributes[light.ATTR_BRIGHTNESS], 153)
script_entity.entity_id, script_entity.state, attributes=attrs
) # Go through the API to turn it off
bedroom_result = self.perform_put_light_state(
@classmethod 'light.bed_light', False)
def tearDownClass(cls):
"""Stop the class.""" bedroom_result_json = bedroom_result.json()
cls.hass.stop()
self.assertEqual(bedroom_result.status_code, 200)
def test_discover_lights(self): self.assertTrue(
"""Test the discovery of lights.""" 'application/json' in bedroom_result.headers['content-type'])
result = requests.get(
BRIDGE_URL_BASE.format('/api/username/lights'), timeout=5) self.assertEqual(len(bedroom_result_json), 1)
self.assertEqual(result.status_code, 200) # Check to make sure the state changed
self.assertTrue('application/json' in result.headers['content-type']) bed_light = self.hass.states.get('light.bed_light')
self.assertEqual(bed_light.state, STATE_OFF)
result_json = result.json()
# Make sure we can't change the kitchen light state
# Make sure the lights we added to the config are there kitchen_result = self.perform_put_light_state(
self.assertTrue('light.ceiling_lights' in result_json) 'light.kitchen_light', True)
self.assertTrue('light.bed_light' in result_json) self.assertEqual(kitchen_result.status_code, 404)
self.assertTrue('script.set_kitchen_light' in result_json)
self.assertTrue('light.kitchen_lights' not in result_json) def test_put_light_state_script(self):
"""Test the setting of script variables."""
def test_get_light_state(self): # Turn the kitchen light off first
"""Test the getting of light state.""" self.hass.services.call(
# Turn office light on and set to 127 brightness light.DOMAIN, const.SERVICE_TURN_OFF,
self.hass.services.call( {const.ATTR_ENTITY_ID: 'light.kitchen_lights'},
light.DOMAIN, const.SERVICE_TURN_ON, blocking=True)
{
const.ATTR_ENTITY_ID: 'light.ceiling_lights', # Emulated hue converts 0-100% to 0-255.
light.ATTR_BRIGHTNESS: 127 level = 23
}, brightness = round(level * 255 / 100)
blocking=True)
script_result = self.perform_put_light_state(
office_json = self.perform_get_light_state('light.ceiling_lights', 200) 'script.set_kitchen_light', True, brightness)
self.assertEqual(office_json['state'][HUE_API_STATE_ON], True) script_result_json = script_result.json()
self.assertEqual(office_json['state'][HUE_API_STATE_BRI], 127)
self.assertEqual(script_result.status_code, 200)
# Turn bedroom light off self.assertEqual(len(script_result_json), 2)
self.hass.services.call(
light.DOMAIN, const.SERVICE_TURN_OFF, # Wait until script is complete before continuing
{ self.hass.block_till_done()
const.ATTR_ENTITY_ID: 'light.bed_light'
}, kitchen_light = self.hass.states.get('light.kitchen_lights')
blocking=True) self.assertEqual(kitchen_light.state, 'on')
self.assertEqual(
bedroom_json = self.perform_get_light_state('light.bed_light', 200) kitchen_light.attributes[light.ATTR_BRIGHTNESS],
level)
self.assertEqual(bedroom_json['state'][HUE_API_STATE_ON], False)
self.assertEqual(bedroom_json['state'][HUE_API_STATE_BRI], 0) # pylint: disable=invalid-name
def test_put_with_form_urlencoded_content_type(self):
# Make sure kitchen light isn't accessible """Test the form with urlencoded content."""
kitchen_url = '/api/username/lights/{}'.format('light.kitchen_lights') # Needed for Alexa
kitchen_result = requests.get( self.perform_put_test_on_ceiling_lights(
BRIDGE_URL_BASE.format(kitchen_url), timeout=5) 'application/x-www-form-urlencoded')
self.assertEqual(kitchen_result.status_code, 404) # Make sure we fail gracefully when we can't parse the data
data = {'key1': 'value1', 'key2': 'value2'}
def test_put_light_state(self): result = requests.put(
"""Test the seeting of light states.""" BRIDGE_URL_BASE.format(
self.perform_put_test_on_ceiling_lights() '/api/username/lights/{}/state'.format(
'light.ceiling_lights')), data=data)
# Turn the bedroom light on first
self.hass.services.call( self.assertEqual(result.status_code, 400)
light.DOMAIN, const.SERVICE_TURN_ON,
{const.ATTR_ENTITY_ID: 'light.bed_light', def test_entity_not_found(self):
light.ATTR_BRIGHTNESS: 153}, """Test for entity which are not found."""
blocking=True) result = requests.get(
BRIDGE_URL_BASE.format(
bed_light = self.hass.states.get('light.bed_light') '/api/username/lights/{}'.format("not.existant_entity")),
self.assertEqual(bed_light.state, STATE_ON) timeout=5)
self.assertEqual(bed_light.attributes[light.ATTR_BRIGHTNESS], 153)
self.assertEqual(result.status_code, 404)
# Go through the API to turn it off
bedroom_result = self.perform_put_light_state( result = requests.put(
'light.bed_light', False) BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format("non.existant_entity")),
bedroom_result_json = bedroom_result.json() timeout=5)
self.assertEqual(bedroom_result.status_code, 200) self.assertEqual(result.status_code, 404)
self.assertTrue(
'application/json' in bedroom_result.headers['content-type']) def test_allowed_methods(self):
"""Test the allowed methods."""
self.assertEqual(len(bedroom_result_json), 1) result = requests.get(
BRIDGE_URL_BASE.format(
# Check to make sure the state changed '/api/username/lights/{}/state'.format(
bed_light = self.hass.states.get('light.bed_light') "light.ceiling_lights")))
self.assertEqual(bed_light.state, STATE_OFF)
self.assertEqual(result.status_code, 405)
# Make sure we can't change the kitchen light state
kitchen_result = self.perform_put_light_state( result = requests.put(
'light.kitchen_light', True) BRIDGE_URL_BASE.format(
self.assertEqual(kitchen_result.status_code, 404) '/api/username/lights/{}'.format("light.ceiling_lights")),
data={'key1': 'value1'})
def test_put_light_state_script(self):
"""Test the setting of script variables.""" self.assertEqual(result.status_code, 405)
# Turn the kitchen light off first
self.hass.services.call( result = requests.put(
light.DOMAIN, const.SERVICE_TURN_OFF, BRIDGE_URL_BASE.format('/api/username/lights'),
{const.ATTR_ENTITY_ID: 'light.kitchen_lights'}, data={'key1': 'value1'})
blocking=True)
self.assertEqual(result.status_code, 405)
# Emulated hue converts 0-100% to 0-255.
level = 23 def test_proper_put_state_request(self):
brightness = round(level * 255 / 100) """Test the request to set the state."""
# Test proper on value parsing
script_result = self.perform_put_light_state( result = requests.put(
'script.set_kitchen_light', True, brightness) BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format(
script_result_json = script_result.json() 'light.ceiling_lights')),
data=json.dumps({HUE_API_STATE_ON: 1234}))
self.assertEqual(script_result.status_code, 200)
self.assertEqual(len(script_result_json), 2) self.assertEqual(result.status_code, 400)
# Wait until script is complete before continuing # Test proper brightness value parsing
self.hass.block_till_done() result = requests.put(
BRIDGE_URL_BASE.format(
kitchen_light = self.hass.states.get('light.kitchen_lights') '/api/username/lights/{}/state'.format(
self.assertEqual(kitchen_light.state, 'on') 'light.ceiling_lights')), data=json.dumps({
self.assertEqual( HUE_API_STATE_ON: True,
kitchen_light.attributes[light.ATTR_BRIGHTNESS], HUE_API_STATE_BRI: 'Hello world!'
level) }))
# pylint: disable=invalid-name self.assertEqual(result.status_code, 400)
def test_put_with_form_urlencoded_content_type(self):
"""Test the form with urlencoded content.""" # pylint: disable=invalid-name
# Needed for Alexa def perform_put_test_on_ceiling_lights(self,
self.perform_put_test_on_ceiling_lights( content_type='application/json'):
'application/x-www-form-urlencoded') """Test the setting of a light."""
# Turn the office light off first
# Make sure we fail gracefully when we can't parse the data self.hass.services.call(
data = {'key1': 'value1', 'key2': 'value2'} light.DOMAIN, const.SERVICE_TURN_OFF,
result = requests.put( {const.ATTR_ENTITY_ID: 'light.ceiling_lights'},
BRIDGE_URL_BASE.format( blocking=True)
'/api/username/lights/{}/state'.format(
'light.ceiling_lights')), data=data) ceiling_lights = self.hass.states.get('light.ceiling_lights')
self.assertEqual(ceiling_lights.state, STATE_OFF)
self.assertEqual(result.status_code, 400)
# Go through the API to turn it on
def test_entity_not_found(self): office_result = self.perform_put_light_state(
"""Test for entity which are not found.""" 'light.ceiling_lights', True, 56, content_type)
result = requests.get(
BRIDGE_URL_BASE.format( office_result_json = office_result.json()
'/api/username/lights/{}'.format("not.existant_entity")),
timeout=5) self.assertEqual(office_result.status_code, 200)
self.assertTrue(
self.assertEqual(result.status_code, 404) 'application/json' in office_result.headers['content-type'])
result = requests.put( self.assertEqual(len(office_result_json), 2)
BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format("non.existant_entity")), # Check to make sure the state changed
timeout=5) ceiling_lights = self.hass.states.get('light.ceiling_lights')
self.assertEqual(ceiling_lights.state, STATE_ON)
self.assertEqual(result.status_code, 404) self.assertEqual(ceiling_lights.attributes[light.ATTR_BRIGHTNESS], 56)
def test_allowed_methods(self): def perform_get_light_state(self, entity_id, expected_status):
"""Test the allowed methods.""" """Test the gettting of a light state."""
result = requests.get( result = requests.get(
BRIDGE_URL_BASE.format( BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format( '/api/username/lights/{}'.format(entity_id)), timeout=5)
"light.ceiling_lights")))
self.assertEqual(result.status_code, expected_status)
self.assertEqual(result.status_code, 405)
if expected_status == 200:
result = requests.put( self.assertTrue(
BRIDGE_URL_BASE.format( 'application/json' in result.headers['content-type'])
'/api/username/lights/{}'.format("light.ceiling_lights")),
data={'key1': 'value1'}) return result.json()
self.assertEqual(result.status_code, 405) return None
result = requests.put( # pylint: disable=no-self-use
BRIDGE_URL_BASE.format('/api/username/lights'), def perform_put_light_state(self, entity_id, is_on, brightness=None,
data={'key1': 'value1'}) content_type='application/json'):
"""Test the setting of a light state."""
self.assertEqual(result.status_code, 405) url = BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format(entity_id))
def test_proper_put_state_request(self):
"""Test the request to set the state.""" req_headers = {'Content-Type': content_type}
# Test proper on value parsing
result = requests.put( data = {HUE_API_STATE_ON: is_on}
BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format( if brightness is not None:
'light.ceiling_lights')), data[HUE_API_STATE_BRI] = brightness
data=json.dumps({HUE_API_STATE_ON: 1234}))
result = requests.put(
self.assertEqual(result.status_code, 400) url, data=json.dumps(data), timeout=5, headers=req_headers)
# Test proper brightness value parsing return result
result = requests.put(
BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format(
'light.ceiling_lights')), data=json.dumps({
HUE_API_STATE_ON: True,
HUE_API_STATE_BRI: 'Hello world!'
}))
self.assertEqual(result.status_code, 400)
# pylint: disable=invalid-name
def perform_put_test_on_ceiling_lights(self,
content_type='application/json'):
"""Test the setting of a light."""
# Turn the office light off first
self.hass.services.call(
light.DOMAIN, const.SERVICE_TURN_OFF,
{const.ATTR_ENTITY_ID: 'light.ceiling_lights'},
blocking=True)
ceiling_lights = self.hass.states.get('light.ceiling_lights')
self.assertEqual(ceiling_lights.state, STATE_OFF)
# Go through the API to turn it on
office_result = self.perform_put_light_state(
'light.ceiling_lights', True, 56, content_type)
office_result_json = office_result.json()
self.assertEqual(office_result.status_code, 200)
self.assertTrue(
'application/json' in office_result.headers['content-type'])
self.assertEqual(len(office_result_json), 2)
# Check to make sure the state changed
ceiling_lights = self.hass.states.get('light.ceiling_lights')
self.assertEqual(ceiling_lights.state, STATE_ON)
self.assertEqual(ceiling_lights.attributes[light.ATTR_BRIGHTNESS], 56)
def perform_get_light_state(self, entity_id, expected_status):
"""Test the gettting of a light state."""
result = requests.get(
BRIDGE_URL_BASE.format(
'/api/username/lights/{}'.format(entity_id)), timeout=5)
self.assertEqual(result.status_code, expected_status)
if expected_status == 200:
self.assertTrue(
'application/json' in result.headers['content-type'])
return result.json()
return None
# pylint: disable=no-self-use
def perform_put_light_state(self, entity_id, is_on, brightness=None,
content_type='application/json'):
"""Test the setting of a light state."""
url = BRIDGE_URL_BASE.format(
'/api/username/lights/{}/state'.format(entity_id))
req_headers = {'Content-Type': content_type}
data = {HUE_API_STATE_ON: is_on}
if brightness is not None:
data[HUE_API_STATE_BRI] = brightness
result = requests.put(
url, data=json.dumps(data), timeout=5, headers=req_headers)
return result

55
tests/components/emulated_hue/test_init.py Executable file
View File

@ -0,0 +1,55 @@
from unittest.mock import patch
from homeassistant.components.emulated_hue import Config, _LOGGER
def test_config_google_home_entity_id_to_number():
"""Test config adheres to the type."""
conf = Config({
'type': 'google_home'
})
number = conf.entity_id_to_number('light.test')
assert number == '1'
number = conf.entity_id_to_number('light.test')
assert number == '1'
number = conf.entity_id_to_number('light.test2')
assert number == '2'
entity_id = conf.number_to_entity_id('1')
assert entity_id == 'light.test'
def test_config_alexa_entity_id_to_number():
"""Test config adheres to the type."""
conf = Config({
'type': 'alexa'
})
number = conf.entity_id_to_number('light.test')
assert number == 'light.test'
number = conf.entity_id_to_number('light.test')
assert number == 'light.test'
number = conf.entity_id_to_number('light.test2')
assert number == 'light.test2'
entity_id = conf.number_to_entity_id('light.test')
assert entity_id == 'light.test'
def test_warning_config_google_home_listen_port():
"""Test we warn when non-default port is used for Google Home."""
with patch.object(_LOGGER, 'warning') as mock_warn:
Config({
'type': 'google_home',
'host_ip': '123.123.123.123',
'listen_port': 8300
})
assert mock_warn.called
assert mock_warn.mock_calls[0][1][0] == \
"When targetting Google Home, listening port has to be port 80"

120
tests/components/emulated_hue/test_upnp.py Normal file
View File

@ -0,0 +1,120 @@
"""The tests for the emulated Hue component."""
import json
import unittest
from unittest.mock import patch
import requests
from homeassistant import bootstrap, const, core
import homeassistant.components as core_components
from homeassistant.components import emulated_hue, http
from homeassistant.util.async import run_coroutine_threadsafe
from tests.common import get_test_instance_port, get_test_home_assistant
HTTP_SERVER_PORT = get_test_instance_port()
BRIDGE_SERVER_PORT = get_test_instance_port()
BRIDGE_URL_BASE = 'http://127.0.0.1:{}'.format(BRIDGE_SERVER_PORT) + '{}'
JSON_HEADERS = {const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON}
def setup_hass_instance(emulated_hue_config):
"""Set up the Home Assistant instance to test."""
hass = get_test_home_assistant()
# We need to do this to get access to homeassistant/turn_(on,off)
run_coroutine_threadsafe(
core_components.async_setup(hass, {core.DOMAIN: {}}), hass.loop
).result()
bootstrap.setup_component(
hass, http.DOMAIN,
{http.DOMAIN: {http.CONF_SERVER_PORT: HTTP_SERVER_PORT}})
bootstrap.setup_component(hass, emulated_hue.DOMAIN, emulated_hue_config)
return hass
def start_hass_instance(hass):
"""Start the Home Assistant instance to test."""
hass.start()
class TestEmulatedHue(unittest.TestCase):
"""Test the emulated Hue component."""
hass = None
@classmethod
def setUpClass(cls):
"""Setup the class."""
cls.hass = hass = get_test_home_assistant()
# We need to do this to get access to homeassistant/turn_(on,off)
run_coroutine_threadsafe(
core_components.async_setup(hass, {core.DOMAIN: {}}), hass.loop
).result()
bootstrap.setup_component(
hass, http.DOMAIN,
{http.DOMAIN: {http.CONF_SERVER_PORT: HTTP_SERVER_PORT}})
with patch('homeassistant.components'
'.emulated_hue.UPNPResponderThread'):
bootstrap.setup_component(hass, emulated_hue.DOMAIN, {
emulated_hue.DOMAIN: {
emulated_hue.CONF_LISTEN_PORT: BRIDGE_SERVER_PORT
}})
cls.hass.start()
@classmethod
def tearDownClass(cls):
"""Stop the class."""
cls.hass.stop()
def test_description_xml(self):
"""Test the description."""
import xml.etree.ElementTree as ET
result = requests.get(
BRIDGE_URL_BASE.format('/description.xml'), timeout=5)
self.assertEqual(result.status_code, 200)
self.assertTrue('text/xml' in result.headers['content-type'])
# Make sure the XML is parsable
# pylint: disable=bare-except
try:
ET.fromstring(result.text)
except:
self.fail('description.xml is not valid XML!')
def test_create_username(self):
"""Test the creation of an username."""
request_json = {'devicetype': 'my_device'}
result = requests.post(
BRIDGE_URL_BASE.format('/api'), data=json.dumps(request_json),
timeout=5)
self.assertEqual(result.status_code, 200)
self.assertTrue('application/json' in result.headers['content-type'])
resp_json = result.json()
success_json = resp_json[0]
self.assertTrue('success' in success_json)
self.assertTrue('username' in success_json['success'])
def test_valid_username_request(self):
"""Test request with a valid username."""
request_json = {'invalid_key': 'my_device'}
result = requests.post(
BRIDGE_URL_BASE.format('/api'), data=json.dumps(request_json),
timeout=5)
self.assertEqual(result.status_code, 400)