おかたづけアシスタントをつくったはなし

これは何?

日々発生する「片付け」という作業.整理整頓を心掛けてもどうしても時間が経つと片づけた場所が思い出せないことがある.これをスマートスピーカーを使って解決していこうという取り組み.


モノを片づけたときにスマートスピーカーに片づけた場所を言う.後々スマートスピーカーに片づけた場所を聞くと,片づけた場所を返答してくれる.

背景:片付けの呪縛

モノを片づけるのは面倒だ.最初に,片付け場所を見つけなければならない.片づけた,と思えばそれを記憶しなければならない.結果忘れてしまってモノが見つからないということもある.

これはカイゼンしていきたいと思うではないか.

在庫管理ではRFIDを活用し,モノに付与されたタグをスキャンすることで場所がわかるようになっている.
ja.wikipedia.org
でもこれを家庭まで持ち込むのはナンセンスだ.いちいちモノを収納するときにタグを付与する以上工程が増える.時間的/経済的コストもかかる.

では,画像認識はどうだろうか.モノを認識,収納先を記録していくようなシステムになることは想像できる.が,これも認識する分手間がかかる.モノの認識精度も現状課題がある.現在の画像認識は特定のモノの認識にのみ特化したものが多い.なのでこれも使えない.

モノの記録にはどうしてもひと手間かかる.これをいかにして減らすか.ラクにするか.

そこで,私は我々が普段生活を送る中で頻繁には使わないチャンネルである,音声に着目した.このチャンネルを使うことで,入力の手間が省けるのではないか,という試みだ.入力の際は,現在普及しているスマートスピーカーが使える.そこで,これを使ったシステムの開発に取り組み始めた.

仕組み

ベースとなるのは過去に作った仕組み.私は音声入力のパターンマッチングによるコンピュータの操作(プログラミング)を可能とした.
www.rect29.com
このシステムでは正規表現により音声入力されたテキストから変数名を抽出する方法を実現している.今回も似たような方法で片付けの記録を行う.「[モノの名前]を[片付けた場所]に片づけたよ~」というような発話は正規表現そのものだ.
片付け場所はDBとして記録する.このDBはプライバシー性が高いため,Raspberry piに記録していく.
まとめるとこんな感じになる.

f:id:newbie29979:20200504131808p:plain
システムの概要

フローは以下の通り.
片付け場所の記録:

  1. ユーザがスマートスピーカーに入力
  2. IFTTT,beebotteを介してRaspberry piに音声入力内容を通知,DBに書き込み

片付け場所の確認:

  1. ユーザがスマートスピーカーに入力
  2. IFTTT,beebotteを介してRaspberry piに音声入力内容を通知,DBから場所を読み出し
  3. 読み出した場所のテキストをCloud Text-to-Speechにより音声に変換
  4. 生成された音声をGoogle Cloud Strageに保存しそのURLをRaspberry pi を介してGoogle Homeにキャスト

実装

スマートスピーカーへの入力音声の認識については前の記事に書いたので,ここでは省略する.また,要素技術の詳細の課題と解決についてはそれぞれ別記事にまとめていく.
今回において,前回のプログラミングにおける変数にあたるのがモノや場所の名前である.変数とは異なり,これらは日本語であることが想定される.従って,パターンマッチングの際にうまくモノや場所とコマンドを分離する必要がある.今回は,モノや場所を 漢字+カタカナ,カタカナのみ または 漢字のみで構成すると仮定し,以下のようにコマンドを構成する.

OkatadukeCommands.json

{
  ".*([一-龥]*[ァ-ー]+|[一-龥]+)(を|は)([一-龥]*[ァ-ー]+|[一-龥]+).*": "1,${w1},${w2}",
  ".*([一-龥]*[ァ-ー]+|[一-龥]+)はどこ.*": "2,${w1}"
}

漢字やカタカナの正規表現については以下のサイトが詳しい.
www.asahi-net.or.jp

正直なところ,漢字やカタカナの正規表現が可能であるのは意外だったが,このようにコマンドの定義づけをすることで書き込みの際には「スマートフォンを机に」といった命令を,読み出しの際には「スマートフォンはどこ」というような発話に対応できる.

Raspberry piで扱うDBはSqlite3を使用する.使用バージョンは現在の最新バージョンである3.31.1.認識結果に従って,以下のように書き込みと読み出しを行う.

Main.py

   try:
        converted_result_ = convert_recog_result_to_command(recog_result_)
        op_ = converted_result_.split(',')
        command_ = Command(op_[0], op_[1:len(op_)])

        conn_ = sqlite3.connect(os.path.join(TARGET_DIR_PATH, 'StrageSpaceDB.sqlite3'))
        print(os.path.join(TARGET_DIR_PATH, 'StrageSpaceDB.sqlite3'))
        c_ = conn_.cursor()

        try:
            if command_.type == CommandType.OKATADUKE:
                c_.execute("INSERT INTO storage_space VALUES (?, ?) ON CONFLICT(thing) DO UPDATE SET place = ?;",
                          (command_.operands[0], command_.operands[1], command_.operands[1]))
                conn_.commit()

            elif command_.type == CommandType.WHERE:
                c_.execute("SELECT place FROM storage_space WHERE thing = ?;", (command_.operands[0],))

                place_ = c_.fetchall()[0][0]
                speech_generator.generate_speech(place_+'です!')

        except sqlite3.Error as e:
            print('sqlite3 error: ', e.args[0])
            speech_generator.generator_speech('データベースのエラーです')

        conn_.close()

    except KeyError:
        print('Key error')

クラスCommandは音声で入力されるコマンドを扱うためのクラス.あるモノを片づける際(DBへの書き込み)にはそのモノの名前と場所がオペランドに,その場所を聞く(DBからの読み出し)のであればそのモノの名前が唯一オペランドになる.

片付け場所の記録

音声認識の結果入力パターンが片付け場所の記録であった場合,DBへの書き込みを行う.今回の場合,モノの名前をUNIQUEキーにした{モノの名前,場所}を1レコードとするDB.モノを移動することも考え,UPSERT構文を使用する.UPSERT構文とはレコードの記録の際,新しいレコードの記録もしくは既存のレコードの上書きが可能であるようなSQL構文を指す.
www.sqlite.org
該当箇所

c_.execute("INSERT INTO storage_space VALUES (?, ?) ON CONFLICT(thing) DO UPDATE SET place = ?;",
                     (command_.operands[0], command_.operands[1], command_.operands[1])

UPSERT構文はSQLiteのバージョンが3.24.0以降でなければ動作しないため,pythonにインポートされるSQLiteのバージョンは要確認である.プレインストールされているものやapt installなどで用意したsqliteはバージョンが古く対応していないこともあるため,場合によってはソースからインストールしなければならない.実際に私も本番環境のSQLiteのバージョンが古く,対応が必要だった.解決の方法は以下の記事にまとめておく.
www.rect29.com

片付け場所の確認

片づけた場所の発声には,対象の片づけた場所のDBから読み出し及びそのテキストの音声合成が必要である.そして合成した音声のGoogle Homeへのキャストもまた,必要である. クライアントが特定のテキストの音声合成クラウドにリクエスト→クラウド音声合成クラウドが合成音声のURL生成→クラウドがクライアントにURLを返答→クライアントが返答されたURLをGoogle Homeにキャスト の順.

Cloud Text-to-Speech:Cloud TTS

cloud.google.com
今回はGoogleのCloud Text-to-Speechを使用し音声合成を行う.Cloud TTSは音声合成に際して400万[文字/月]まで無料であるため,個人利用であれば実質無料で音声合成ができると考えられる.
Cloud TTSの使い方はココを参考に記述した.まとめると,適切なリクエストの記述さえできれば,API Keyと併用することで入力テキストの音声合成が可能である.参考の参照元となっているStack Overflowのこの記事が詳しい.
stackoverflow.com
Cloud TTSはリクエストを受け取ると,json形式でBase64により暗号化されたmp3を返す.これではGoogle Homeにキャストする際に都合が悪いため,そのURLをクラウドから返してもらうようにしたい.そこで利用するのがGoogle Cloud StrageとCloud Functionsである.

Google Cloud Strage:GCS

cloud.google.com
もはやWebベースのサービス提供に必須ともいえるクラウドサービスだが,中でもGoogleが提供しているクラウドサービスである.今回は合成音声の保存と,そのURLの配信のために使用する.GCSでは,クラウドへの何らかのイベントをトリガーとして実行可能な関数であるCloud Functionsが利用できる.すなわち,GCS上でpythonのコードを走らせることが可能である.そこで今回はGCSに対してリクエストがクライアントから送られたとき,Cloud Functionにより音声合成をリクエスト,合成音声のURLを作成しクライアントであるRaspberry piに返すコードをCloud functionとして作成した.

Chrome Cast

URLが入手できれば,次はそれをGoogle Homeにキャストする.そこで利用可能なライブラリがpychromecastである.
pypi.org
これはかなり簡単にGoogle Homeでのキャストができるライブラリである.ココを参考にするとすぐに実装可能.

以上をまとめたソース

以上クライアントと,GCSのソースをまとめると以下のようになる.

クライアントサイド(SpeechGenerator.py)

class SpeechGenerator:
    generator_params = {}
    googlehome = None

    def __init__(self, _generator_params):
        self.generator_params = _generator_params

        chromecasts_ = pychromecast.get_chromecasts()
        self.googlehome = next(cc_ for cc_ in chromecasts_ if cc_.device.friendly_name == self.generator_params["googlehome_name"])

    def generate_speech(self, _request_text):
        json_data_ = {
            'api_key':      self.generator_params['tts_api_key'],
            'text':         _request_text,
            'bucket_name':  self.generator_params['gcs_bucket_name'],
            'key_filename': self.generator_params['gcs_key_filename']
        }

        url_ = self.generator_params['gcp_cloud_function_url']
        jd_ = json.dumps(json_data_)
        headers_ = {'Content-Type': 'application/json; charset=utf-8'}

        s_ = requests.Session()
        res_ = requests.post(url_, data=jd_, headers=headers_)

        time.sleep(2)

        self.googlehome.wait()
        self.googlehome.media_controller.play_media(res_.text, 'audio.mp3')
        self.googlehome.media_controller.block_until_active()

リクエストを投げる際のapi_keyは音声合成の際のAPI Keyを示しており,textは音声合成の対象のテキストを示す.GCSではプロジェクト内でバケットという単位でストレージを扱う.ストレージ内ファイルを扱うにはこのバケットを指定するbucket_nameが必要である.key_filenameは後述する署名付きURL発行のトリックのための措置.

クラウドサイド(CloudFunctions)

API_ACCESS_ENDPOINT = 'https://storage.googleapis.com'

def main(request):
    request_json = request.get_json()
    api_key = request_json['api_key']
    request_text = request_json['text']
    bucket_name = request_json['bucket_name']
    key_filename = request_json['key_filename']

    # Request for Text to Speech
    url = "https://texttospeech.googleapis.com/v1beta1/text:synthesize?key=" + api_key
    headers = {'Content-Type': 'application/json; charset=utf-8'}
    json_data = {
        'input': {
            'text': request_text
        },
        'voice': {
            'languageCode': 'ja-JP',
            'name': 'ja-JP-Wavenet-A',
            'ssmlGender': 'FEMALE'
        },
        'audioConfig': {
            'audioEncoding': 'MP3'
        }
    }

    jd = json.dumps(json_data)
    s = requests.Session()
    res = requests.post(url, data=jd, headers=headers)

    if res.status_code == 200:
        # save mp3 file
        parsed_data = json.loads(res.text)
        filename = '/tmp/generated_data' + datetime.now().strftime('%Y%m%d_%H%M%S') + '.mp3'
        with open(filename, 'wb') as f:
            f.write(base64.b64decode(parsed_data['audioContent']))
        
        # upload saved mp3 to GCS
        client = storage.Client()
        bucket = client.bucket(bucket_name)
        blob = bucket.blob(filename)
        blob.upload_from_filename(filename)

        # generate signed mp3 url on GCS
        blob = bucket.blob(key_filename)
        json_string = blob.download_as_string()
        key_file_dict = json.loads(json_string)
        credentials = ServiceAccountCredentials.from_json_keyfile_dict(key_file_dict)

        gcs_filename = '/%s/%s' % (bucket_name, filename)
        content_md5, content_type = None, None

        google_access_id = credentials.service_account_email

        expiration = datetime.now() + timedelta(seconds=60)
        expiration = int(time.mktime(expiration.timetuple()))
        
        signature_string = '\n'.join(['GET', content_md5 or '', content_type or '', str(expiration), gcs_filename])
        _, signature_bytes = credentials.sign_blob(signature_string)
        signature = base64.b64encode(signature_bytes)

        query_params = {'GoogleAccessId': google_access_id, 'Expires': str(expiration), 'Signature': signature}

        return '{endpoint}{resource}?{querystring}'.format(endpoint=API_ACCESS_ENDPOINT, resource=gcs_filename, querystring=urllib.parse.urlencode(query_params))
    else:
        return "Error: " + str(res.status_code)

Cloud Functionsでpythonを扱う際にはimportするライブラリを別途Requirements.txtに記述する必要があるため,注意が必要である.始めるにあたっては公式のチュートリアルが最も手っ取り早く簡単に理解可能であるためオススメ.
クラウド上のデータに無期限でアクセス可能なのはセキュリティ上望ましくないため,合成音声のURLとして署名付きURLを発行する.しかしこの署名付きURLの発行については,Cloud Functionsで実装するには少々クセがある.そこで,Cloud FunctionsではGoogleの署名付きURL発行のフォーマットに従って,トリックを用いることで予め用意された署名付きURL発行用のメソッドを使用せずに記述している.この詳細については以下の記事に取り上げている.
www.rect29.com

まとめ

Google Homeを用いたおかたづけアシスタントを構築した.このシステムはGoogle Homeを用いることで簡単に入力でき,その片付け場所のDBはRaspberry piがもつ.私が記述したソースを以下に置いておく.
github.com