Add Alexa Flash Briefing Skill API support (#3745)

* Add Alexa Flash Briefing Skill API support

* Set default value for text to empty string as per API docs

* Clean up existing Alexa tests

* Update configuration parsing and validation

* Add tests for the Flash Briefing API

* Update test_alexa.py
This commit is contained in:
Robbie Trencheny 2016-10-13 09:14:22 -07:00 committed by Paulus Schoutsen
parent 8c13d3ed4c
commit c663d85129
2 changed files with 349 additions and 179 deletions

View File

@ -7,16 +7,20 @@ https://home-assistant.io/components/alexa/
import copy import copy
import enum import enum
import logging import logging
import uuid
from datetime import datetime
import voluptuous as vol import voluptuous as vol
from homeassistant.const import HTTP_BAD_REQUEST from homeassistant.const import HTTP_BAD_REQUEST
from homeassistant.helpers import template, script, config_validation as cv from homeassistant.helpers import template, script, config_validation as cv
from homeassistant.components.http import HomeAssistantView from homeassistant.components.http import HomeAssistantView
import homeassistant.util.dt as dt_util
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
API_ENDPOINT = '/api/alexa' INTENTS_API_ENDPOINT = '/api/alexa'
FLASH_BRIEFINGS_API_ENDPOINT = '/api/alexa/flash_briefings/<briefing_id>'
CONF_ACTION = 'action' CONF_ACTION = 'action'
CONF_CARD = 'card' CONF_CARD = 'card'
@ -28,6 +32,23 @@ CONF_TITLE = 'title'
CONF_CONTENT = 'content' CONF_CONTENT = 'content'
CONF_TEXT = 'text' CONF_TEXT = 'text'
CONF_FLASH_BRIEFINGS = 'flash_briefings'
CONF_UID = 'uid'
CONF_DATE = 'date'
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' DOMAIN = 'alexa'
DEPENDENCIES = ['http'] DEPENDENCIES = ['http']
@ -61,6 +82,16 @@ CONFIG_SCHEMA = vol.Schema({
vol.Required(CONF_TEXT): cv.template, vol.Required(CONF_TEXT): cv.template,
} }
} }
},
CONF_FLASH_BRIEFINGS: {
cv.string: vol.All(cv.ensure_list, [{
vol.Required(CONF_UID, default=str(uuid.uuid4())): cv.string,
vol.Optional(CONF_DATE, default=datetime.utcnow()): 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) }, extra=vol.ALLOW_EXTRA)
@ -68,16 +99,19 @@ CONFIG_SCHEMA = vol.Schema({
def setup(hass, config): def setup(hass, config):
"""Activate Alexa component.""" """Activate Alexa component."""
hass.wsgi.register_view(AlexaView(hass, intents = config[DOMAIN].get(CONF_INTENTS, {})
config[DOMAIN].get(CONF_INTENTS, {}))) flash_briefings = config[DOMAIN].get(CONF_FLASH_BRIEFINGS, {})
hass.wsgi.register_view(AlexaIntentsView(hass, intents))
hass.wsgi.register_view(AlexaFlashBriefingView(hass, flash_briefings))
return True return True
class AlexaView(HomeAssistantView): class AlexaIntentsView(HomeAssistantView):
"""Handle Alexa requests.""" """Handle Alexa requests."""
url = API_ENDPOINT url = INTENTS_API_ENDPOINT
name = 'api:alexa' name = 'api:alexa'
def __init__(self, hass, intents): def __init__(self, hass, intents):
@ -235,3 +269,69 @@ class AlexaResponse(object):
'sessionAttributes': self.session_attributes, 'sessionAttributes': self.session_attributes,
'response': response, 'response': response,
} }
class AlexaFlashBriefingView(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__(hass)
self.flash_briefings = copy.deepcopy(flash_briefings)
template.attach(hass, self.flash_briefings)
# pylint: disable=too-many-branches
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 self.Response(status=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].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].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].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].render()
else:
output[ATTR_REDIRECTION_URL] = item.get(CONF_DISPLAY_URL)
if isinstance(item[CONF_DATE], str):
item[CONF_DATE] = dt_util.parse_datetime(item[CONF_DATE])
output[ATTR_UPDATE_DATE] = item[CONF_DATE].strftime(DATE_FORMAT)
briefing.append(output)
return self.json(briefing)

View File

@ -2,6 +2,7 @@
# pylint: disable=protected-access,too-many-public-methods # pylint: disable=protected-access,too-many-public-methods
import json import json
import time import time
import datetime
import unittest import unittest
import requests import requests
@ -13,19 +14,27 @@ from tests.common import get_test_instance_port, get_test_home_assistant
API_PASSWORD = "test1234" API_PASSWORD = "test1234"
SERVER_PORT = get_test_instance_port() SERVER_PORT = get_test_instance_port()
API_URL = "http://127.0.0.1:{}{}".format(SERVER_PORT, alexa.API_ENDPOINT) BASE_API_URL = "http://127.0.0.1:{}".format(SERVER_PORT)
INTENTS_API_URL = "{}{}".format(BASE_API_URL, alexa.INTENTS_API_ENDPOINT)
HA_HEADERS = { HA_HEADERS = {
const.HTTP_HEADER_HA_AUTH: API_PASSWORD, const.HTTP_HEADER_HA_AUTH: API_PASSWORD,
const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON, const.HTTP_HEADER_CONTENT_TYPE: const.CONTENT_TYPE_JSON,
} }
SESSION_ID = 'amzn1.echo-api.session.0000000-0000-0000-0000-00000000000' SESSION_ID = "amzn1.echo-api.session.0000000-0000-0000-0000-00000000000"
APPLICATION_ID = 'amzn1.echo-sdk-ams.app.000000-d0ed-0000-ad00-000000d00ebe' APPLICATION_ID = "amzn1.echo-sdk-ams.app.000000-d0ed-0000-ad00-000000d00ebe"
REQUEST_ID = 'amzn1.echo-api.request.0000000-0000-0000-0000-00000000000' REQUEST_ID = "amzn1.echo-api.request.0000000-0000-0000-0000-00000000000"
# pylint: disable=invalid-name
hass = None hass = None
calls = [] calls = []
NPR_NEWS_MP3_URL = "https://pd.npr.org/anon.npr-mp3/npr/news/newscast.mp3"
# 2016-10-10T19:51:42+00:00
STATIC_TIME = datetime.datetime.utcfromtimestamp(1476129102)
def setUpModule(): # pylint: disable=invalid-name def setUpModule(): # pylint: disable=invalid-name
"""Initialize a Home Assistant server for testing this module.""" """Initialize a Home Assistant server for testing this module."""
@ -36,23 +45,40 @@ def setUpModule(): # pylint: disable=invalid-name
bootstrap.setup_component( bootstrap.setup_component(
hass, http.DOMAIN, hass, http.DOMAIN,
{http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD, {http.DOMAIN: {http.CONF_API_PASSWORD: API_PASSWORD,
http.CONF_SERVER_PORT: SERVER_PORT}}) http.CONF_SERVER_PORT: SERVER_PORT}})
hass.services.register('test', 'alexa', lambda call: calls.append(call)) hass.services.register("test", "alexa", lambda call: calls.append(call))
bootstrap.setup_component(hass, alexa.DOMAIN, { bootstrap.setup_component(hass, alexa.DOMAIN, {
# Key is here to verify we allow other keys in config too # Key is here to verify we allow other keys in config too
'homeassistant': {}, "homeassistant": {},
'alexa': { "alexa": {
'intents': { "flash_briefings": {
'WhereAreWeIntent': { "weather": [
'speech': { {"title": "Weekly forecast",
'type': 'plaintext', "text": "This week it will be sunny.",
'text': "date": "2016-10-09T19:51:42.0Z"},
{"title": "Current conditions",
"text": "Currently it is 80 degrees fahrenheit.",
"date": STATIC_TIME}
],
"news_audio": {
"title": "NPR",
"audio": NPR_NEWS_MP3_URL,
"display_url": "https://npr.org",
"date": STATIC_TIME,
"uid": "uuid"
}
},
"intents": {
"WhereAreWeIntent": {
"speech": {
"type": "plaintext",
"text":
""" """
{%- if is_state('device_tracker.paulus', 'home') {%- if is_state("device_tracker.paulus", "home")
and is_state('device_tracker.anne_therese', and is_state("device_tracker.anne_therese",
'home') -%} "home") -%}
You are both home, you silly You are both home, you silly
{%- else -%} {%- else -%}
Anne Therese is at {{ Anne Therese is at {{
@ -64,23 +90,23 @@ def setUpModule(): # pylint: disable=invalid-name
""", """,
} }
}, },
'GetZodiacHoroscopeIntent': { "GetZodiacHoroscopeIntent": {
'speech': { "speech": {
'type': 'plaintext', "type": "plaintext",
'text': 'You told us your sign is {{ ZodiacSign }}.', "text": "You told us your sign is {{ ZodiacSign }}.",
} }
}, },
'CallServiceIntent': { "CallServiceIntent": {
'speech': { "speech": {
'type': 'plaintext', "type": "plaintext",
'text': 'Service called', "text": "Service called",
}, },
'action': { "action": {
'service': 'test.alexa', "service": "test.alexa",
'data_template': { "data_template": {
'hello': '{{ ZodiacSign }}' "hello": "{{ ZodiacSign }}"
}, },
'entity_id': 'switch.test', "entity_id": "switch.test",
} }
} }
} }
@ -96,11 +122,19 @@ def tearDownModule(): # pylint: disable=invalid-name
hass.stop() hass.stop()
def _req(data={}): def _intent_req(data={}):
return requests.post(API_URL, data=json.dumps(data), timeout=5, return requests.post(INTENTS_API_URL, data=json.dumps(data), timeout=5,
headers=HA_HEADERS) headers=HA_HEADERS)
def _flash_briefing_req(briefing_id=None):
url_format = "{}/api/alexa/flash_briefings/{}"
FLASH_BRIEFING_API_URL = url_format.format(BASE_API_URL,
briefing_id)
return requests.get(FLASH_BRIEFING_API_URL, timeout=5,
headers=HA_HEADERS)
class TestAlexa(unittest.TestCase): class TestAlexa(unittest.TestCase):
"""Test Alexa.""" """Test Alexa."""
@ -108,231 +142,267 @@ class TestAlexa(unittest.TestCase):
"""Stop everything that was started.""" """Stop everything that was started."""
hass.block_till_done() hass.block_till_done()
def test_launch_request(self): def test_intent_launch_request(self):
"""Test the launch of a request.""" """Test the launch of a request."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': True, "new": True,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': {}, "attributes": {},
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'LaunchRequest', "type": "LaunchRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z' "timestamp": "2015-05-13T12:34:56Z"
} }
} }
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
resp = req.json() resp = req.json()
self.assertIn('outputSpeech', resp['response']) self.assertIn("outputSpeech", resp["response"])
def test_intent_request_with_slots(self): def test_intent_request_with_slots(self):
"""Test a request with slots.""" """Test a request with slots."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': False, "new": False,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': { "attributes": {
'supportedHoroscopePeriods': { "supportedHoroscopePeriods": {
'daily': True, "daily": True,
'weekly': False, "weekly": False,
'monthly': False "monthly": False
} }
}, },
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'IntentRequest', "type": "IntentRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z', "timestamp": "2015-05-13T12:34:56Z",
'intent': { "intent": {
'name': 'GetZodiacHoroscopeIntent', "name": "GetZodiacHoroscopeIntent",
'slots': { "slots": {
'ZodiacSign': { "ZodiacSign": {
'name': 'ZodiacSign', "name": "ZodiacSign",
'value': 'virgo' "value": "virgo"
} }
} }
} }
} }
} }
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
text = req.json().get('response', {}).get('outputSpeech', text = req.json().get("response", {}).get("outputSpeech",
{}).get('text') {}).get("text")
self.assertEqual('You told us your sign is virgo.', text) self.assertEqual("You told us your sign is virgo.", text)
def test_intent_request_with_slots_but_no_value(self): def test_intent_request_with_slots_but_no_value(self):
"""Test a request with slots but no value.""" """Test a request with slots but no value."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': False, "new": False,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': { "attributes": {
'supportedHoroscopePeriods': { "supportedHoroscopePeriods": {
'daily': True, "daily": True,
'weekly': False, "weekly": False,
'monthly': False "monthly": False
} }
}, },
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'IntentRequest', "type": "IntentRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z', "timestamp": "2015-05-13T12:34:56Z",
'intent': { "intent": {
'name': 'GetZodiacHoroscopeIntent', "name": "GetZodiacHoroscopeIntent",
'slots': { "slots": {
'ZodiacSign': { "ZodiacSign": {
'name': 'ZodiacSign', "name": "ZodiacSign",
} }
} }
} }
} }
} }
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
text = req.json().get('response', {}).get('outputSpeech', text = req.json().get("response", {}).get("outputSpeech",
{}).get('text') {}).get("text")
self.assertEqual('You told us your sign is .', text) self.assertEqual("You told us your sign is .", text)
def test_intent_request_without_slots(self): def test_intent_request_without_slots(self):
"""Test a request without slots.""" """Test a request without slots."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': False, "new": False,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': { "attributes": {
'supportedHoroscopePeriods': { "supportedHoroscopePeriods": {
'daily': True, "daily": True,
'weekly': False, "weekly": False,
'monthly': False "monthly": False
} }
}, },
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'IntentRequest', "type": "IntentRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z', "timestamp": "2015-05-13T12:34:56Z",
'intent': { "intent": {
'name': 'WhereAreWeIntent', "name": "WhereAreWeIntent",
} }
} }
} }
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
text = req.json().get('response', {}).get('outputSpeech', text = req.json().get("response", {}).get("outputSpeech",
{}).get('text') {}).get("text")
self.assertEqual('Anne Therese is at unknown and Paulus is at unknown', self.assertEqual("Anne Therese is at unknown and Paulus is at unknown",
text) text)
hass.states.set('device_tracker.paulus', 'home') hass.states.set("device_tracker.paulus", "home")
hass.states.set('device_tracker.anne_therese', 'home') hass.states.set("device_tracker.anne_therese", "home")
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
text = req.json().get('response', {}).get('outputSpeech', text = req.json().get("response", {}).get("outputSpeech",
{}).get('text') {}).get("text")
self.assertEqual('You are both home, you silly', text) self.assertEqual("You are both home, you silly", text)
def test_intent_request_calling_service(self): def test_intent_request_calling_service(self):
"""Test a request for calling a service.""" """Test a request for calling a service."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': False, "new": False,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': {}, "attributes": {},
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'IntentRequest', "type": "IntentRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z', "timestamp": "2015-05-13T12:34:56Z",
'intent': { "intent": {
'name': 'CallServiceIntent', "name": "CallServiceIntent",
'slots': { "slots": {
'ZodiacSign': { "ZodiacSign": {
'name': 'ZodiacSign', "name": "ZodiacSign",
'value': 'virgo', "value": "virgo",
} }
} }
} }
} }
} }
call_count = len(calls) call_count = len(calls)
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
self.assertEqual(call_count + 1, len(calls)) self.assertEqual(call_count + 1, len(calls))
call = calls[-1] call = calls[-1]
self.assertEqual('test', call.domain) self.assertEqual("test", call.domain)
self.assertEqual('alexa', call.service) self.assertEqual("alexa", call.service)
self.assertEqual(['switch.test'], call.data.get('entity_id')) self.assertEqual(["switch.test"], call.data.get("entity_id"))
self.assertEqual('virgo', call.data.get('hello')) self.assertEqual("virgo", call.data.get("hello"))
def test_session_ended_request(self): def test_intent_session_ended_request(self):
"""Test the request for ending the session.""" """Test the request for ending the session."""
data = { data = {
'version': '1.0', "version": "1.0",
'session': { "session": {
'new': False, "new": False,
'sessionId': SESSION_ID, "sessionId": SESSION_ID,
'application': { "application": {
'applicationId': APPLICATION_ID "applicationId": APPLICATION_ID
}, },
'attributes': { "attributes": {
'supportedHoroscopePeriods': { "supportedHoroscopePeriods": {
'daily': True, "daily": True,
'weekly': False, "weekly": False,
'monthly': False "monthly": False
} }
}, },
'user': { "user": {
'userId': 'amzn1.account.AM3B00000000000000000000000' "userId": "amzn1.account.AM3B00000000000000000000000"
} }
}, },
'request': { "request": {
'type': 'SessionEndedRequest', "type": "SessionEndedRequest",
'requestId': REQUEST_ID, "requestId": REQUEST_ID,
'timestamp': '2015-05-13T12:34:56Z', "timestamp": "2015-05-13T12:34:56Z",
'reason': 'USER_INITIATED' "reason": "USER_INITIATED"
} }
} }
req = _req(data) req = _intent_req(data)
self.assertEqual(200, req.status_code) self.assertEqual(200, req.status_code)
self.assertEqual('', req.text) self.assertEqual("", req.text)
def test_flash_briefing_invalid_id(self):
"""Test an invalid Flash Briefing ID."""
req = _flash_briefing_req()
self.assertEqual(404, req.status_code)
self.assertEqual("", req.text)
def test_flash_briefing_date_from_str(self):
"""Test the response has a valid date parsed from string."""
req = _flash_briefing_req("weather")
self.assertEqual(200, req.status_code)
self.assertEqual(req.json()[0].get(alexa.ATTR_UPDATE_DATE),
"2016-10-09T19:51:42.0Z")
def test_flash_briefing_date_from_datetime(self):
"""Test the response has a valid date from a datetime object."""
req = _flash_briefing_req("weather")
self.assertEqual(200, req.status_code)
self.assertEqual(req.json()[1].get(alexa.ATTR_UPDATE_DATE),
'2016-10-10T19:51:42.0Z')
def test_flash_briefing_valid(self):
"""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 = _flash_briefing_req("news_audio")
self.assertEqual(200, req.status_code)
response = req.json()
self.assertEqual(response, data)