evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Break up Alexa per functionality (#9400)

Browse source 
* Break up Alexa per functionality

* Lint

* Lint
This commit is contained in:
Paulus Schoutsen 2017-09-12 12:24:44 -07:00 • committed by Pascal Vizeli
parent 29b62f814f
commit 05192e678e
7 changed files with 272 additions and 185 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
homeassistant/components/alexa/__init__.py Normal file
View file

@ -0,0 +1,52 @@
"""
Support for Alexa skill service end point.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/alexa/
"""
import asyncio
import logging
import voluptuous as vol
from homeassistant.helpers import config_validation as cv
from .const import (
DOMAIN, CONF_UID, CONF_TITLE, CONF_AUDIO, CONF_TEXT, CONF_DISPLAY_URL)
from . import flash_briefings, intent
_LOGGER = logging.getLogger(__name__)
DEPENDENCIES = ['http']
CONF_FLASH_BRIEFINGS = 'flash_briefings'
CONFIG_SCHEMA = vol.Schema({
DOMAIN: {
CONF_FLASH_BRIEFINGS: {
cv.string: vol.All(cv.ensure_list, [{
vol.Optional(CONF_UID): cv.string,
vol.Required(CONF_TITLE): cv.template,
vol.Optional(CONF_AUDIO): cv.template,
vol.Required(CONF_TEXT, default=""): cv.template,
vol.Optional(CONF_DISPLAY_URL): cv.template,
}]),
}
}
}, extra=vol.ALLOW_EXTRA)
@asyncio.coroutine
def async_setup(hass, config):
"""Activate Alexa component."""
config = config.get(DOMAIN, {})
flash_briefings_config = config.get(CONF_FLASH_BRIEFINGS)
intent.async_setup(hass)
if flash_briefings_config:
flash_briefings.async_setup(hass, flash_briefings_config)
return True

18
homeassistant/components/alexa/const.py Normal file
View file

@ -0,0 +1,18 @@
"""Constants for the Alexa integration."""
DOMAIN = 'alexa'
# Flash briefing constants
CONF_UID = 'uid'
CONF_TITLE = 'title'
CONF_AUDIO = 'audio'
CONF_TEXT = 'text'
CONF_DISPLAY_URL = 'display_url'
ATTR_UID = 'uid'
ATTR_UPDATE_DATE = 'updateDate'
ATTR_TITLE_TEXT = 'titleText'
ATTR_STREAM_URL = 'streamUrl'
ATTR_MAIN_TEXT = 'mainText'
ATTR_REDIRECTION_URL = 'redirectionURL'
DATE_FORMAT = '%Y-%m-%dT%H:%M:%S.0Z'

96
homeassistant/components/alexa/flash_briefings.py Normal file
View file

@ -0,0 +1,96 @@
"""
Support for Alexa skill service end point.
For more details about this component, please refer to the documentation at
https://home-assistant.io/components/alexa/
"""
import copy
import logging
from datetime import datetime
import uuid
from homeassistant.core import callback
from homeassistant.helpers import template
from homeassistant.components import http
from .const import (
CONF_UID, CONF_TITLE, CONF_AUDIO, CONF_TEXT, CONF_DISPLAY_URL, ATTR_UID,
ATTR_UPDATE_DATE, ATTR_TITLE_TEXT, ATTR_STREAM_URL, ATTR_MAIN_TEXT,
ATTR_REDIRECTION_URL, DATE_FORMAT)
_LOGGER = logging.getLogger(__name__)
FLASH_BRIEFINGS_API_ENDPOINT = '/api/alexa/flash_briefings/{briefing_id}'
@callback
def async_setup(hass, flash_briefing_config):
"""Activate Alexa component."""
hass.http.register_view(
AlexaFlashBriefingView(hass, flash_briefing_config))
class AlexaFlashBriefingView(http.HomeAssistantView):
"""Handle Alexa Flash Briefing skill requests."""
url = FLASH_BRIEFINGS_API_ENDPOINT
name = 'api:alexa:flash_briefings'
def __init__(self, hass, flash_briefings):
"""Initialize Alexa view."""
super().__init__()
self.flash_briefings = copy.deepcopy(flash_briefings)
template.attach(hass, self.flash_briefings)
@callback
def get(self, request, briefing_id):
"""Handle Alexa Flash Briefing request."""
_LOGGER.debug('Received Alexa flash briefing request for: %s',
briefing_id)
if self.flash_briefings.get(briefing_id) is None:
err = 'No configured Alexa flash briefing was found for: %s'
_LOGGER.error(err, briefing_id)
return b'', 404
briefing = []
for item in self.flash_briefings.get(briefing_id, []):
output = {}
if item.get(CONF_TITLE) is not None:
if isinstance(item.get(CONF_TITLE), template.Template):
output[ATTR_TITLE_TEXT] = item[CONF_TITLE].async_render()
else:
output[ATTR_TITLE_TEXT] = item.get(CONF_TITLE)
if item.get(CONF_TEXT) is not None:
if isinstance(item.get(CONF_TEXT), template.Template):
output[ATTR_MAIN_TEXT] = item[CONF_TEXT].async_render()
else:
output[ATTR_MAIN_TEXT] = item.get(CONF_TEXT)
uid = item.get(CONF_UID)
if uid is None:
uid = str(uuid.uuid4())
output[ATTR_UID] = uid
if item.get(CONF_AUDIO) is not None:
if isinstance(item.get(CONF_AUDIO), template.Template):
output[ATTR_STREAM_URL] = item[CONF_AUDIO].async_render()
else:
output[ATTR_STREAM_URL] = item.get(CONF_AUDIO)
if item.get(CONF_DISPLAY_URL) is not None:
if isinstance(item.get(CONF_DISPLAY_URL),
template.Template):
output[ATTR_REDIRECTION_URL] = \
item[CONF_DISPLAY_URL].async_render()
else:
output[ATTR_REDIRECTION_URL] = item.get(CONF_DISPLAY_URL)
output[ATTR_UPDATE_DATE] = datetime.now().strftime(DATE_FORMAT)
briefing.append(output)
return self.json(briefing)

126
homeassistant/components/alexa.py → homeassistant/components/alexa/intent.py
View file

@ -5,52 +5,19 @@ For more details about this component, please refer to the documentation at
https://home-assistant.io/components/alexa/
"""
import asyncio
import copy
import enum
import logging
import uuid
from datetime import datetime
import voluptuous as vol
from homeassistant.core import callback
from homeassistant.const import HTTP_BAD_REQUEST
from homeassistant.helpers import intent, template, config_validation as cv
from homeassistant.helpers import intent
from homeassistant.components import http
_LOGGER = logging.getLogger(__name__)
from .const import DOMAIN
INTENTS_API_ENDPOINT = '/api/alexa'
FLASH_BRIEFINGS_API_ENDPOINT = '/api/alexa/flash_briefings/{briefing_id}'
CONF_ACTION = 'action'
CONF_CARD = 'card'
CONF_INTENTS = 'intents'
CONF_SPEECH = 'speech'
CONF_TYPE = 'type'
CONF_TITLE = 'title'
CONF_CONTENT = 'content'
CONF_TEXT = 'text'
CONF_FLASH_BRIEFINGS = 'flash_briefings'
CONF_UID = 'uid'
CONF_TITLE = 'title'
CONF_AUDIO = 'audio'
CONF_TEXT = 'text'
CONF_DISPLAY_URL = 'display_url'
ATTR_UID = 'uid'
ATTR_UPDATE_DATE = 'updateDate'
ATTR_TITLE_TEXT = 'titleText'
ATTR_STREAM_URL = 'streamUrl'
ATTR_MAIN_TEXT = 'mainText'
ATTR_REDIRECTION_URL = 'redirectionURL'
DATE_FORMAT = '%Y-%m-%dT%H:%M:%S.0Z'
DOMAIN = 'alexa'
DEPENDENCIES = ['http']
_LOGGER = logging.getLogger(__name__)
class SpeechType(enum.Enum):
@ -73,30 +40,10 @@ class CardType(enum.Enum):
link_account = "LinkAccount"
CONFIG_SCHEMA = vol.Schema({
DOMAIN: {
CONF_FLASH_BRIEFINGS: {
cv.string: vol.All(cv.ensure_list, [{
vol.Required(CONF_UID, default=str(uuid.uuid4())): cv.string,
vol.Required(CONF_TITLE): cv.template,
vol.Optional(CONF_AUDIO): cv.template,
vol.Required(CONF_TEXT, default=""): cv.template,
vol.Optional(CONF_DISPLAY_URL): cv.template,
}]),
}
}
}, extra=vol.ALLOW_EXTRA)
@asyncio.coroutine
def async_setup(hass, config):
@callback
def async_setup(hass):
"""Activate Alexa component."""
flash_briefings = config[DOMAIN].get(CONF_FLASH_BRIEFINGS, {})
hass.http.register_view(AlexaIntentsView)
hass.http.register_view(AlexaFlashBriefingView(hass, flash_briefings))
return True
class AlexaIntentsView(http.HomeAssistantView):
@ -255,66 +202,3 @@ class AlexaResponse(object):
'sessionAttributes': self.session_attributes,
'response': response,
}
class AlexaFlashBriefingView(http.HomeAssistantView):
"""Handle Alexa Flash Briefing skill requests."""
url = FLASH_BRIEFINGS_API_ENDPOINT
name = 'api:alexa:flash_briefings'
def __init__(self, hass, flash_briefings):
"""Initialize Alexa view."""
super().__init__()
self.flash_briefings = copy.deepcopy(flash_briefings)
template.attach(hass, self.flash_briefings)
@callback
def get(self, request, briefing_id):
"""Handle Alexa Flash Briefing request."""
_LOGGER.debug('Received Alexa flash briefing request for: %s',
briefing_id)
if self.flash_briefings.get(briefing_id) is None:
err = 'No configured Alexa flash briefing was found for: %s'
_LOGGER.error(err, briefing_id)
return b'', 404
briefing = []
for item in self.flash_briefings.get(briefing_id, []):
output = {}
if item.get(CONF_TITLE) is not None:
if isinstance(item.get(CONF_TITLE), template.Template):
output[ATTR_TITLE_TEXT] = item[CONF_TITLE].async_render()
else:
output[ATTR_TITLE_TEXT] = item.get(CONF_TITLE)
if item.get(CONF_TEXT) is not None:
if isinstance(item.get(CONF_TEXT), template.Template):
output[ATTR_MAIN_TEXT] = item[CONF_TEXT].async_render()
else:
output[ATTR_MAIN_TEXT] = item.get(CONF_TEXT)
if item.get(CONF_UID) is not None:
output[ATTR_UID] = item.get(CONF_UID)
if item.get(CONF_AUDIO) is not None:
if isinstance(item.get(CONF_AUDIO), template.Template):
output[ATTR_STREAM_URL] = item[CONF_AUDIO].async_render()
else:
output[ATTR_STREAM_URL] = item.get(CONF_AUDIO)
if item.get(CONF_DISPLAY_URL) is not None:
if isinstance(item.get(CONF_DISPLAY_URL),
template.Template):
output[ATTR_REDIRECTION_URL] = \
item[CONF_DISPLAY_URL].async_render()
else:
output[ATTR_REDIRECTION_URL] = item.get(CONF_DISPLAY_URL)
output[ATTR_UPDATE_DATE] = datetime.now().strftime(DATE_FORMAT)
briefing.append(output)
return self.json(briefing)

1
tests/components/alexa/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Tests for the Alexa integration."""

98
tests/components/alexa/test_flash_briefings.py Normal file
View file

@ -0,0 +1,98 @@
"""The tests for the Alexa component."""
# pylint: disable=protected-access
import asyncio
import datetime
import pytest
from homeassistant.core import callback
from homeassistant.setup import async_setup_component
from homeassistant.components import alexa
from homeassistant.components.alexa import const
SESSION_ID = "amzn1.echo-api.session.0000000-0000-0000-0000-00000000000"
APPLICATION_ID = "amzn1.echo-sdk-ams.app.000000-d0ed-0000-ad00-000000d00ebe"
REQUEST_ID = "amzn1.echo-api.request.0000000-0000-0000-0000-00000000000"
# pylint: disable=invalid-name
calls = []
NPR_NEWS_MP3_URL = "https://pd.npr.org/anon.npr-mp3/npr/news/newscast.mp3"
@pytest.fixture
def alexa_client(loop, hass, test_client):
"""Initialize a Home Assistant server for testing this module."""
@callback
def mock_service(call):
calls.append(call)
hass.services.async_register("test", "alexa", mock_service)
assert loop.run_until_complete(async_setup_component(hass, alexa.DOMAIN, {
# Key is here to verify we allow other keys in config too
"homeassistant": {},
"alexa": {
"flash_briefings": {
"weather": [
{"title": "Weekly forecast",
"text": "This week it will be sunny."},
{"title": "Current conditions",
"text": "Currently it is 80 degrees fahrenheit."}
],
"news_audio": {
"title": "NPR",
"audio": NPR_NEWS_MP3_URL,
"display_url": "https://npr.org",
"uid": "uuid"
}
},
}
}))
return loop.run_until_complete(test_client(hass.http.app))
def _flash_briefing_req(client, briefing_id):
return client.get(
"/api/alexa/flash_briefings/{}".format(briefing_id))
@asyncio.coroutine
def test_flash_briefing_invalid_id(alexa_client):
"""Test an invalid Flash Briefing ID."""
req = yield from _flash_briefing_req(alexa_client, 10000)
assert req.status == 404
text = yield from req.text()
assert text == ''
@asyncio.coroutine
def test_flash_briefing_date_from_str(alexa_client):
"""Test the response has a valid date parsed from string."""
req = yield from _flash_briefing_req(alexa_client, "weather")
assert req.status == 200
data = yield from req.json()
assert isinstance(datetime.datetime.strptime(data[0].get(
const.ATTR_UPDATE_DATE), const.DATE_FORMAT), datetime.datetime)
@asyncio.coroutine
def test_flash_briefing_valid(alexa_client):
"""Test the response is valid."""
data = [{
"titleText": "NPR",
"redirectionURL": "https://npr.org",
"streamUrl": NPR_NEWS_MP3_URL,
"mainText": "",
"uid": "uuid",
"updateDate": '2016-10-10T19:51:42.0Z'
}]
req = yield from _flash_briefing_req(alexa_client, "news_audio")
assert req.status == 200
json = yield from req.json()
assert isinstance(datetime.datetime.strptime(json[0].get(
const.ATTR_UPDATE_DATE), const.DATE_FORMAT), datetime.datetime)
json[0].pop(const.ATTR_UPDATE_DATE)
data[0].pop(const.ATTR_UPDATE_DATE)
assert json == data

66
tests/components/test_alexa.py → tests/components/alexa/test_intent.py
View file

@ -2,13 +2,13 @@
# pylint: disable=protected-access
import asyncio
import json
import datetime
import pytest
from homeassistant.core import callback
from homeassistant.setup import async_setup_component
from homeassistant.components import alexa
from homeassistant.components.alexa import intent
SESSION_ID = "amzn1.echo-api.session.0000000-0000-0000-0000-00000000000"
APPLICATION_ID = "amzn1.echo-sdk-ams.app.000000-d0ed-0000-ad00-000000d00ebe"
@ -32,22 +32,6 @@ def alexa_client(loop, hass, test_client):
assert loop.run_until_complete(async_setup_component(hass, alexa.DOMAIN, {
# Key is here to verify we allow other keys in config too
"homeassistant": {},
"alexa": {
"flash_briefings": {
"weather": [
{"title": "Weekly forecast",
"text": "This week it will be sunny."},
{"title": "Current conditions",
"text": "Currently it is 80 degrees fahrenheit."}
],
"news_audio": {
"title": "NPR",
"audio": NPR_NEWS_MP3_URL,
"display_url": "https://npr.org",
"uid": "uuid"
}
},
}
}))
assert loop.run_until_complete(async_setup_component(
hass, 'intent_script', {
@ -113,15 +97,10 @@ def alexa_client(loop, hass, test_client):
def _intent_req(client, data={}):
return client.post(alexa.INTENTS_API_ENDPOINT, data=json.dumps(data),
return client.post(intent.INTENTS_API_ENDPOINT, data=json.dumps(data),
headers={'content-type': 'application/json'})
def _flash_briefing_req(client, briefing_id):
return client.get(
"/api/alexa/flash_briefings/{}".format(briefing_id))
@asyncio.coroutine
def test_intent_launch_request(alexa_client):
"""Test the launch of a request."""
@ -467,44 +446,3 @@ def test_intent_from_built_in_intent_library(alexa_client):
text = data.get("response", {}).get("outputSpeech",
{}).get("text")
assert text == "Playing the shins."
@asyncio.coroutine
def test_flash_briefing_invalid_id(alexa_client):
"""Test an invalid Flash Briefing ID."""
req = yield from _flash_briefing_req(alexa_client, 10000)
assert req.status == 404
text = yield from req.text()
assert text == ''
@asyncio.coroutine
def test_flash_briefing_date_from_str(alexa_client):
"""Test the response has a valid date parsed from string."""
req = yield from _flash_briefing_req(alexa_client, "weather")
assert req.status == 200
data = yield from req.json()
assert isinstance(datetime.datetime.strptime(data[0].get(
alexa.ATTR_UPDATE_DATE), alexa.DATE_FORMAT), datetime.datetime)
@asyncio.coroutine
def test_flash_briefing_valid(alexa_client):
"""Test the response is valid."""
data = [{
"titleText": "NPR",
"redirectionURL": "https://npr.org",
"streamUrl": NPR_NEWS_MP3_URL,
"mainText": "",
"uid": "uuid",
"updateDate": '2016-10-10T19:51:42.0Z'
}]
req = yield from _flash_briefing_req(alexa_client, "news_audio")
assert req.status == 200
json = yield from req.json()
assert isinstance(datetime.datetime.strptime(json[0].get(
alexa.ATTR_UPDATE_DATE), alexa.DATE_FORMAT), datetime.datetime)
json[0].pop(alexa.ATTR_UPDATE_DATE)
data[0].pop(alexa.ATTR_UPDATE_DATE)
assert json == data