Alexaに消防車がどこにいるか教えてもらおう

Alexaに消防車がどこにいるか教えてもらおう:

この記事はスマートスピーカー Advent Calendar 2018 16日目の記事です。

消防車の出動状況を教えてくれるスキルを作った

消防車のサイレンが聞こえたとき、近くで火事でもあったのかな？って思いますよね。

あまり知られていないようですが、多くの自治体では、消防車の出動状況がWebで公開されています。

私の住んでいる千葉市だと
http://business4.plala.or.jp/chiba119/web/chiba/annai.html

で公開されています。

消防車のサイレンが聞こえたとき、よくこのページを見ていましたが、スマートスピーカーで内容わかったら便利！てことでスキルを作って公開してみました。

↓↓↓

https://www.amazon.co.jp/%E3%81%B5%E3%81%AA-%E6%B6%88%E9%98%B2%E3%83%8A%E3%83%93/dp/B07KQ5CCGT/

今の所千葉市消防局のみ対応しています。

HowTo

バックエンド側はPythonでコードを組んでAWS Lambdaで動かしています。

消防車の出動状況を取得する。

消防局のページはインラインフレームになっており、取得したい情報は
http://business4.plala.or.jp/chiba119/web/chiba/annai_list.html

から取得できることがわかりました。これをスクレイピングしてリストにします。

出動状況の文字列が一行ごとにstrongで囲まれていたのでこれを利用します。いつも出動しているとは限らないのでデバッグ用にローカルにhtmlをおいています。

消防車の出動状況をリストにして返す関数

# 消防車の出動状況を調べる 
# 出動状況一覧をリストで返す 
def get_fire_info(url): 
    content = urllib.request.urlopen(url).read().decode('shift_jisx0213') 
 
    # 本番用 
    soup = BeautifulSoup(content, "html.parser") 
 
    # デバッグ用 
    # soup = BeautifulSoup(open('test_list.html',encoding="utf-8_sig"), "html.parser") 
 
    content_list = soup.find_all("strong") 
    text_list = [] 
 
    for c in content_list: 
        text_list.append(c.text) 
 
    return text_list 

出動状況のリストをAlexa用に処理する。

出動状況を読み上げるのですが、全部読み上げてると何件も出動していると読み上げが終わらないので、

一件ずつ読み上げられるようにしました。

ユーザの現在の読み上げ位置を記録するために。session_attributesを用いました。
https://dev.classmethod.jp/cloud/alexa-sdk-v2-fourth/#toc-3-session-attributes

AlexaにJSONをかえすときにsession attributesを含めておくと、次の会話のときにAlexa側からのJSONに前の会話のsession attributesが含まれるようになります。

            info = get_fire_info(FIRE_INFO_URL) 
            title = '消防車の出動状況' 
            reprompt = 'ほかの情報も聞きますか？' 
            message = '' 
            should_end_session = True 
            session_attributes = event['session']['attributes'] 
            if 'pos' in session_attributes: 
                message += info[session_attributes['pos']] 
                session_attributes['pos'] += 1 
                if len(info) > session_attributes['pos']: 
                    message += 'ほかの情報も聞きますか？' 
                    should_end_session = False 
                else: 
                    message += '出動情報は以上です。' 

全体の処理

その他、ヘルプ処理（ヘルプと言ったら使い方を説明する）、消防車が出動していないときの返答などを書くと

import logging 
import urllib.request 
from bs4 import BeautifulSoup 
from distutils.util import strtobool 
import sys 
 
# 消防署のURL 
FIRE_INFO_URL = 'http://business4.plala.or.jp/chiba119/web/chiba/annai_list.html' 
 
# 何もないときに出ているメッセージ 
NO_FIRE = 'お出かけ前、お休み前、必ず火の元の点検をしましょう。' 
 
 
# 消防車の出動状況を調べる 
# 出動状況一覧をリストで返す 
def get_fire_info(url): 
    content = urllib.request.urlopen(url).read().decode('shift_jisx0213') 
 
    # 本番用 
    soup = BeautifulSoup(content, "html.parser") 
 
    # デバッグ用 
    # soup = BeautifulSoup(open('test_list.html',encoding="utf-8_sig"), "html.parser") 
 
    content_list = soup.find_all("strong") 
    text_list = [] 
 
    for c in content_list: 
        text_list.append(c.text) 
 
    return text_list 
 
 
def lambda_handler(event, context): 
    logging.info(event) 
    is_new = str(event['session']['new']) 
 
 
    if strtobool(is_new) or event['request']['type'] == 'LaunchRequest': 
        return onLaunchRequest(event) 
    elif event['request']['type'] == 'IntentRequest': 
        return onIntentRequest(event) 
    else: 
        return onSessionEndedRequest(event) 
 
def onLaunchRequest(event): 
    try: 
        info = get_fire_info(FIRE_INFO_URL) 
        title = '消防車の出動状況' 
        reprompt = 'ほかの情報も聞きますか？' 
        message = '' 
        should_end_session = True 
        session_attributes = {"pos" : 1} 
        if info[0] == NO_FIRE: 
            message = '現在、市内で消防車は出動していません。' 
        else: 
            message += str(len(info)) + "件の出動情報があります。" 
            message += info[0] 
        if len(info) > 1: 
            message += 'ほかの情報も聞きますか？' 
            should_end_session = False 
    except Exception as e: 
        tb = sys.exc_info()[2] 
        message = 'エラー {}'.format(e.with_traceback(tb)) 
 
    logging.info(message) 
 
    speechlet_response = build_speechlet_response(title, message, reprompt, should_end_session) 
    return build_response(session_attributes, speechlet_response) 
 
def onIntentRequest(event): 
    intent_name = event['request']['intent']['name'] 
    if intent_name == 'GetFireInfoIntent': 
        return onLaunchRequest(event) 
    elif intent_name == 'YesIntent': 
        try: 
            info = get_fire_info(FIRE_INFO_URL) 
            title = '消防車の出動状況' 
            reprompt = 'ほかの情報も聞きますか？' 
            message = '' 
            should_end_session = True 
            session_attributes = event['session']['attributes'] 
            if 'pos' in session_attributes: 
                message += info[session_attributes['pos']] 
                session_attributes['pos'] += 1 
                if len(info) > session_attributes['pos']: 
                    message += 'ほかの情報も聞きますか？' 
                    should_end_session = False 
                else: 
                    message += '出動情報は以上です。' 
            else: 
                return onLaunchRequest(event) 
        except Exception as e: 
            tb = sys.exc_info()[2] 
            message = 'エラー {}'.format(e.with_traceback(tb)) 
        logging.info(message) 
        speechlet_response = build_speechlet_response(title, message, reprompt, should_end_session) 
        return build_response(session_attributes, speechlet_response) 
 
    elif intent_name == 'AMAZON.HelpIntent': 
        title = '消防車の出動状況' 
        message = '千葉市内の消防車の出動状況をお知らせします。「消防ナビを開いて」と話しかけてみてください。' 
        reprompt = '「消防ナビを開いて」と話しかけてみてください。' 
        speechlet_response = build_speechlet_response(title, message, reprompt, False) 
        return build_response('', speechlet_response) 
 
    else: 
        return onSessionEndedRequest(event) 
 
 
def onSessionEndedRequest(event): 
    message = 'ご利用ありがとうございました。' 
    speechlet_response = build_speechlet_response('', message, '', True) 
    return build_response('', speechlet_response) 
 
def build_speechlet_response(title, output, reprompt_text, should_end_session): 
    """Build general speech response. 
    Args: 
        title: The card title. 
        output: The text Alexa will read and loud. 
        reprompt_text: The text for reprompt. 
        should_end_session: The bool value indicates the session will end or 
                            not. 
    Returns: 
        The speech response dictionary will be converted into JSON. 
    """ 
    return { 
        'outputSpeech': { 
            'type': 'PlainText', 
            'text': output 
        }, 
        'card': { 
            'type': 'Simple', 
            'title': title, 
            'content': output 
        }, 
        'reprompt': { 
            'outputSpeech': { 
                'type': 'PlainText', 
                'text': reprompt_text 
            } 
        }, 
        'shouldEndSession': should_end_session 
    } 
 
 
def build_response(session_attributes, speechlet_response): 
    """Build response will be sent to Alexa. 
    Args: 
        session_attributes: The session attributes. 
        speechlet_response: The speech response. 
    Returns: 
        The response dictionary will be converted into JSON. 
    """ 
    return { 
        'version': '1.0', 
        'sessionAttributes': session_attributes, 
        'response': speechlet_response 
    } 
 
 

これをLambda側に設定して、Alexa Skill Kitとつなげることでスキルが出来上がります。

まとめ

スマートスピーカーのスキルを作ったのは初めてでしたが、JSONのやりとりで会話が作れるので思いの外簡単でした。

一応スキルを公開するときに審査があるのですが、落ちたのは一回だけで、千葉市非公認であることを明記するように指摘されただけでした。

TシャツやAmazon Echo Spotがもらえるキャンペーンも定期的に開催されているようなのでみなさんも開発してみてはいかがでしょうか。

今後

千葉市消防局にしか対応していないので、他の自治体にも対応したいですね。東京消防庁はWeb上で公開していないようですが・・・・　ユーザの住所から近い現場順に読み上げるとか実装すると便利そうです。