Alexa のスマートホームスキルで呼ばれる lambda で取得したイベントを SQS を使ってデバイスに送る方法を紹介します。目的は AWS の勉強ですので正直プロダクトじゃない場合はこの辺は自作せず買った方が良い気がします。
これは以前の記事にも書きましたが AWS greengrass でもっときれいに実現できる（と思います）。一応 python でスマートホームスキル書いたのでそれの軽い解説もしておきます。
前提
- python 3.6
- Alexa での smart home skill の作り方は知っている
- 私は初めてのホームスキル作成
参考サイト
Alexa のスキルに対応するコード
Alexa のメッセージが送られて来る Lambda の部分です。namespaceとnameに合わせて処理内容を書きます。今回は基本的に Discovery のイベント以外はラズパイに投げてしまい、ラズパイで色々な作業をします。
def my_handler(event, context):
logging("DEBUG", "Request", event)
header = event['directive']['header']
namespace = header['namespace']
name = header['name']
# Discovery request だったら Lambda で処理
if namespace == ALEXA_DISCOVERY and name == DISCOVER:
return handleDiscovery(event, context)
# 今回は Lambda 側ではあまり作業はせずにイベントをデバイスに渡してしまう
# configファイルの読み込み
ini = configparser.ConfigParser()
ini.read("./config.ini")
# 現在(2018/01/14)日本語の Alexa スキルはオレゴンでしか実行できないので
# 東京リージョンのSQSを使うためにリージョンを指定
sqs = boto3.client('sqs', region_name='ap-northeast-1')
requestQueUrl = ini['sqs']['requestQueUrl']
responseQueUrl = ini['sqs']['responseQueUrl']
# イベントを送信
result = sendQueBody(sqs, requestQueUrl, event)
if result is False:
return createErrorResponse(event, 'INTERNAL_ERROR', "Failed to send request to the device")
# 結果を受信
response = receiveQueBody(sqs, responseQueUrl)
if response is None:
return createErrorResponse(event, 'INTERNAL_ERROR', "Failed to receive a device's response")
logging("DEBUG", "Response", response)
return response
requestQueUrlとresponseQueUrlには SQS の URL を設定ファイルから読み取っています。https://sqs.ap-northeast-1.amazonaws.com/XXXXXXXXXX/YYYYYYYYYってかんじのやつです。
SQS クライアントは作った場所のリージョンを指定する必要があるので気をつけてください。
スキルの作成時にどこで実行するのかを選べるのでおそらくエッジロケーションで lambda は実行されていると思います。Far-east と指定ができるのでおそらく日本付近で実行されているのではないかと。なので SQS も日本に作っておくのが良いと思います。
まずは本題の SQS の部分の説明、次におまけでhandleDiscovery(event, context)を紹介します。
SQS の送受信
sendQueBody()とreceiveQueBody()の実装です。
# -*- coding: utf-8 -*-
import json
import hashlib
# url で指定した SQS に body を json で送信
def sendQueBody(sqs, url, body):
jsonBody = json.dumps(body)
response = sqs.send_message(
QueueUrl=url,
DelaySeconds=0,
MessageBody=(
jsonBody
)
)
# 送信に成功したか確認
if response['MD5OfMessageBody'] != hashlib.md5(jsonBody.encode('utf-8')).hexdigest():
logging("ERROR", "sendQueBody", "Failed to send sqs")
return False
return True
# SQS からメッセージを受け取る
# returnType: dict
def receiveQueBody(sqs, url):
response = sqs.receive_message(
QueueUrl=url,
AttributeNames=[
'SentTimestamp'
],
MaxNumberOfMessages=1,
VisibilityTimeout=0,
WaitTimeSeconds=20
)
# キューになにもない場合
if 'Messages' not in response:
return None
message = response['Messages'][0]
body = json.loads(message['Body'])
# メッセージを削除するための情報を取得
receipt_handle = message['ReceiptHandle']
# 場合によっては処理が完了してからメッセージを削除するが
# 今回は受信した時点で削除する
sqs.delete_message(
QueueUrl=url,
ReceiptHandle=receipt_handle
)
return body
SQS からデータを削除するタイミングは場合によると思いますが、今回は Alexa に命令したものが一回失敗したら次も失敗する想定だったのでリトライとか考えずに受信したら削除してしまいます。sqs.send_message()はMD5OfMessageBodyによって送信ができているかが確認できます。
handleDiscovery(event, context)
Alexa にスマートホーム家電を見つけてなどと指示をしたときの処理をする部分です。これで「電気をつけて」などに対応できることを Alexa に伝えられます。対応する部分は別に実装する必要があります。
# Alexa の Discovery request に対応するためのコード
# ここでこのスキルで使えるデバイスの内容を全て返してあげる。
def handleDiscovery(event):
payload = {
"endpoints": [
{
# 電気
"endpointId": RIGHT_00["endpointId"],
"manufacturerName": COMPANY_NAME,
"friendlyName": RIGHT_00["friendlyName"],
"description": "This is smart device",
"displayCategories": [RIGHT_00["displayCategories"]],
"capabilities": [{
"type": "AlexaInterface",
"interface": "Alexa",
"version": "3"
},
{
"interface": "Alexa.PowerController",
"version": "3",
"type": "AlexaInterface",
"properties": {
"supported": [{
"name": "powerState"
}],
"retrievable": True
}
}
]
},
{
# TV
"endpointId": TV_00["endpointId"],
"manufacturerName": COMPANY_NAME,
"friendlyName": TV_00["friendlyName"],
"description": "This is smart device",
"displayCategories": [TV_00["displayCategories"]],
"capabilities": [{
"type": "AlexaInterface",
"interface": "Alexa",
"version": "3"
},
{
"interface": "Alexa.PowerController",
"version": "3",
"type": "AlexaInterface",
"properties": {
"supported": [{
"name": "powerState"
}],
"retrievable": True
}
}
]
},
{
# エアコン
"endpointId": THERMOSTAT_00["endpointId"],
"manufacturerName": COMPANY_NAME,
"friendlyName": THERMOSTAT_00["friendlyName"],
"description": "This is smart device",
"displayCategories": [THERMOSTAT_00["displayCategories"]],
"capabilities": [{
"type": "AlexaInterface",
"interface": "Alexa",
"version": "3"
},
{
"interface": "Alexa.PowerController",
"version": "3",
"type": "AlexaInterface",
"properties": {
"supported": [{
"name": "powerState"
}],
"retrievable": True
}
}
]
}
]
}
header = event['directive']['header']
header['name'] = "Discover.Response"
logging("DEBUG", 'Discovery Response: ', ({'header': header, 'payload': payload}))
return {
'event': {'header': header, 'payload': payload}
}
所感
Alexa スキルが Lambda で実装されていることをしり、面白そうだなと Alexa 予約してスキルを作っています。JSONでの送受信フォーマットが違うと Alexa は毎回「XXの応答がありません」としか返してくれないのでデバックがやりにくいです。またいざ自分で作ってみると日本語に対応していない API が多いようでがっかりしています。早く対応してほしい。これからさかんになる IoT の実装を体験できたのは良い勉強だったのではないでしょうか。
まとめ
今回は Lambda から SQS の送受信を説明しました。また Alexa に Discovery 部分も軽く紹介しました。今度はラズパイでの家電を実際に動かす実装を紹介します。