Google AsisstantやAlexaに頼らないオリジナルスマートスピーカーの作り方

オリジナルスマートスピーカーのすゝめ

皆さん、世界に一つしかないオリジナルのスマートスピーカーを作ってみませんか？
今回は、ラズベリーパイを使ってGoogle AsisstantやAlexaに頼らないスマートスピーカーを作っていきたいと思います。

準備

####準備物

• Raspberry pi 4(Buster)
• ReSpeaker 4-Mic Array for Raspberry Pi
• GROVE - スピーカー(アンプ内蔵)

####使用するサービス（アカウントを作ってAPIKeyを控えておく。）

• Amivoice
• COTOBA Agent (後ほど設定します。スキップしてください。)
• VoiceTextWebAPI

構成

上の図のような構成となります。最初にWakewordを認識したら、対話状態を開始します。Respeakerが音声を拾って、音声データをAmivoiceに渡します。Amivoiceは音声データをテキストに変換するサービスを提供しています。変換されたテキストはCOTOBA Agentで処理され、テキスト（対話の返答）を返します。COTOBA Agentは対話シナリオの作成、運用を提供します。最後に、ユーザーに音声で返答するためにテキストを音声に変換します。今回はVoiceTextWebAPIを使用します。

Respeakerのセットアップ

参考：https://github.com/SeeedDocument/ReSpeaker-4-Mic-Array-for-Raspberry-Pi

スピーカーから音を出す

/boot/config.txtを開いて

dtoverlay=audremap

を追加します。

sudo reboot

再起動をしたら、マイクをrespeakerの右側のGPIOポートに装着してください。可変抵抗を調整しスピーカーから音が出ることを確認します。（Youtubeで動画を流すなど確認はなんでもok）

マイクから音を拾う(WakeWordのサンプルを実行する)

続いてSPIを有効にします。

sudo raspi-config

1. advance>audio>audiojackを選択
2. SPIを有効にする

mkdir /home/pi/git
cd ~/git
git clone https://github.com/respeaker/seeed-voicecard.git
cd seeed-voicecard
sudo ./install.sh 4mic
sudo reboot

次にライブラリのインストールを行います。

cd ~/git
sudo apt install swig python-dev libpulse-dev
sudo apt-get install libatlas-base-dev
sudo apt-get install libasound2-dev
sudo pip install pocketsphinx webrtcvad
sudo pip install pyaudio pyusb
sudo pip install respeaker

wget http://www.portaudio.com/archives/pa_stable_v190600_20161030.tgz
tar -xvzf ./pa_stable_v190600_20161030.tgz
cd portaudio
./configure
make
sudo make install

サンプルコードを試してみます。

cd ~/git
git clone https://github.com/respeaker/respeaker_python_library.git
cd respeaker_python_library/examples
sudo apt-get install python-pyaudio
sudo python ./offline_voice_assistant.py

sudo python ./offline_voice_assistant.pyを実行して何もエラーが発生しなければ、(流暢な英語で)マイクに向かって"respeaker"と話しかけてみてください。うまくいけば、Wake Upと表示されるはずです。

Amivoiceセットアップ

参考：https://acp.amivoice.com/main/manual/websocket%E9%9F%B3%E5%A3%B0%E8%AA%8D%E8%AD%98api%E3%81%AE%E5%88%A9%E7%94%A8%E4%BD%93%E9%A8%93/
今回は、AmivoiceのWebsocket機能を使用します。

サンプルプログラムのダウンロード

まずAmivoice公式サンプルプログラムをダウンロードします。

cd ~/Download
mkdir sample
cd sample
unzip ../sample.zip
cd ..
mv sample ~/git
cd ~/git/sample/Wrp/python
mv src/com/ .

パスと認証ファイルを通す

set PYTHONPATH=src
set SSL_CERT_FILE=../../curl-ca-bundle.crt
export PYTHONPATH=./src
export SSL_CERT_FILE=../../curl-ca-bundle.crt

export ~のコマンドは毎回起動ごとに実行しなかればなりませんが、面倒くさい場合は~/.profileに追加してください。

サンプルコードの実行

cd ~/git/sample/Wrp/python
export PYTHONPATH=./src
export SSL_CERT_FILE=../../curl-ca-bundle.crt
python WrpSimpleTester.py wss://acp-api.amivoice.com/v1/ ../../audio/test.wav 16K -a-general [APIKEY]

何もエラーなくレスポンスが返ってきたら成功です。

VoiceTextWebAPIセットアップ

参考： https://cloud.voicetext.jp/webapi
###インストール

cd git
git clone https://github.com/youtalk/python-voicetext.git
cd python-voicetext
sudo python setup.py install

サンプルプログラムを実行

test_voicetext.pyの１２行目YOUR_API_KEY部分にVoiceTextWebAPIで
登録して送られてきたキーを入力してください。ファイルのなかのAPI=[APIKEY]に自分のAPIを入れる

cd python-voicetext/test/
python test_voicetext.py

音声が流れたら成功です。

COTOBA Agentセットアップ

参考：https://docs.cotoba.net/documentation/cli/

CLIのインストール

pip3 install cotoba-cli

cotobaと入力してusasgeが表示されたら次に進みます。パスが通っていないなどのエラーがあった場合は export PATH=$PATH:/home/pi/.local/binを実行してください。(自分の場合はこれでいけた)

CLIの設定、ボットの作成

チュートリアルを進めて、CLIの設定、ボットを作成してください。

チュートリアルで作成したボットの活用

先ほど作成したボットを利用するには、以下のアドレスにアクセスしてください。ENDOPOINT_URLはcotoba configureで設定したものを、BOT_IDはチュートリアル時に取得したものを使ってください。

アクセス先: https://ENDPOINT_URL/bots/BOT_ID/ask

いくつかリクエスト例を載せて置きます。(これを応用すればスマートスピーカー以外にも活用できます。)

curl

curl -d  '{"utterance": "こんにちは"}' 'https://ENDPOINT_URL/bots/BOT_ID/ask'

python

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import requests
import json

url = 'https://ENDPOINT_URL/bots/BOT_ID/ask'
json_data = json.dumps({"utterance": "こんにちは"})
result = requests.post(url, json_data, headers={'Content-Type': 'application/json'}).json()
result=result["response"]
print(result)

スマートスピーカー作成例

ここまで、色んな技術を準備してきました。では、これまでの技術を統合してスマートスピーカーとして活用できるプログラムを書きましょう。注意点があります。先ほどまでのプログラムはすべてpython2で実行しましたが、以下のプログラムはpython3で実行します。

全てのライブラリを一つのファイルにまとめる

cd /home/pi
mkdir workshop
cd workshop
cp -r ~/git/python-voicetext/voicetext .
cp -r ~/git/sample/Wrp/python/src .
cp -r ~/git/respeaker_python_library/respeaker .
cp -r ~/git/sample/Wrp/python/com .
cp ~/git/sample/curl-ca-bundle.crt .
set SSL_CERT_FILE=./curl-ca-bundle.crt
export SSL_CERT_FILE=./curl-ca-bundle.crt
mkdir audio
sudo apt-get- install install python3-dev
sudo pip3 install pyaudio pyusb
sudo pip3 install respeaker
sudo pip3 install wave
sudo pip3 install pocketsphinx webrtcvad

VoiceTextWebAPIをPython3に対応させる

~/workshop/voicetext/__init__.pyの184行目を以下のように修正してください。

修正前

data = temp.readframes(min(data, self.CHUNK))

修正後

data = temp.readframes(min(int.from_bytes(data,'big'), self.CHUNK))

サンプルコード

nano /home/pi/workshop/smart_speaker.py

以下のコードは「respeaker」と話しかけて起動します。そのあとの音声をAmivoiceで文章化し、COTOBA Agentで処理、その結果をVoiceTextWebAPIで読み上げるものです。このプログラムは、既製品のスマートスピーカーのように話し終わったら対話状態を終了するような実装はしていません。(つまり、一度ウェイクアップすると対話状態のままになります。)
また、各個人で取得したAPIを [COTOBA_URL]、[AMIVOICE_API]、[VOICETEXT_APIKEY] に入れてください。

smart_speaker.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
import json
import pyaudio
import numpy as np
import wave
import codecs
import com.amivoice.wrp.WrpListener
import com.amivoice.wrp.Wrp
import sys
import time
from io import BufferedReader
import unittest
from voicetext import VoiceText, VoiceTextException
import requests
import logging
import time
from threading import Thread, Event
from respeaker import Microphone

sys.dont_write_bytecode = True

CHUNK = 1024
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 48000  # 48kHz amivoiceの指定レートは16kHzだか、一旦48kHzにして割る3する
RECORD_SECONDS = 5 #起動したあとに喋り続けれる最大秒数
WAVE_OUTPUT_FILENAME = "./audio/output.wav" #話した音声を一度保管する場所
SPEAK_OUTPUT_FILENAME="./audio/speak.wav" #応答音声を一度保管する場所
THRESHOLD = 1200 #「話し中」とする音声の大きさの基準点
MAX_COUNT = 40 #沈黙後に「話し終了」とするカウント


#WakeUp機能
def task(quit_event):
    mic = Microphone(quit_event=quit_event)

    while not quit_event.is_set():
        if mic.wakeup('respeaker'): #「リスピーカー」で起動
            print('Wake up')
            mic.stop()
            mic.close()
            return True


#入力音声を処理する
def input_audio():
    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK)
    #print("* recording")
    frames = []

    while True:
        data = stream.read(CHUNK)
        buf = np.frombuffer(data, dtype="int16")
        count = 0
        if buf.max() > THRESHOLD: #声量が基準より高かったら録音開始
            frames.append(b''.join(buf[::3]))

            for i in range(0, int(RATE / CHUNK * int(RECORD_SECONDS))):
                data = stream.read(CHUNK)
                buf = np.frombuffer(data, dtype="int16")
                frames.append(b''.join(buf[::3]))

                if buf.max() <= THRESHOLD:
                    count = count+1
                if count > MAX_COUNT:
                    break

        if count > 0:
            break
    #print("* done recording")

    stream.stop_stream()
    stream.close()
    p.terminate()
    wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
    wf.setnchannels(CHANNELS)
    wf.setsampwidth(p.get_sample_size(FORMAT))
    wf.setframerate(RATE / 3)
    wf.writeframes(b''.join(frames))
    wf.close()
    
    return b''.join(frames)


#オーディオ再生
def playWav(file):
    try:
        wf = wave.open(file, "r")
        
    except FileNotFoundError:
        print("[Error 404] No such file or directory: " + file)
        return 0
    
    p = pyaudio.PyAudio()
    stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
                    channels=wf.getnchannels(),
                    rate=wf.getframerate(),
                    output=True)

    data = wf.readframes(CHUNK)
    
    while len(data) > 0:
        stream.write(data)
        data = wf.readframes(CHUNK)
        
    stream.stop_stream()
    stream.close()
    p.terminate()
    
    
#amivoice処理
class WrpSimpleTester(com.amivoice.wrp.WrpListener):
    @staticmethod
    def main(audio):
        serverURL = "wss://acp-api.amivoice.com/v1/"
        audioFileName = WAVE_OUTPUT_FILENAME
        codec = "16K"
        grammarFileNames = "-a-general"
        authorization = "[AMIVOICE_API]"

        listener = WrpSimpleTester()

        wrp = com.amivoice.wrp.Wrp.construct()
        wrp.setListener(listener)
        wrp.setServerURL(serverURL)
        wrp.setCodec(codec)
        wrp.setGrammarFileNames(grammarFileNames)
        wrp.setAuthorization(authorization)

        if not wrp.connect():
            #            print(wrp.getLastMessage())
            print(u"WebSocket 音声認識サーバ %s への接続に失敗しました。" % serverURL)
            return

        try:
            if not wrp.feedDataResume():
                # print(wrp.getLastMessage())
                print(u"WebSocket 音声認識サーバへの音声データの送信の開始に失敗しました。")
                return

            try:
            #オーディオをamivoiceに2048づつ送信
                with open(audioFileName, "rb") as audioStream:
                    audioData = audioStream.read(2048)

                    while len(audioData) > 0:
                        maxSleepTime = 50000

                        while wrp.getWaitingResults() > 1 and maxSleepTime > 0:
                            wrp.sleep(100)
                            maxSleepTime -= 100

                        if not wrp.feedData(audioData, 0, len(audioData)):
                            # print(wrp.getLastMessage())
                            print(u"WebSocket 音声認識サーバへの音声データの送信に失敗しました。")
                            break

                        audioData = audioStream.read(2048)
            except:
                print(u"失敗しました。")

            if not wrp.feedDataPause():
                # print(wrp.getLastMessage())
                print(u"WebSocket 音声認識サーバへの音声データの送信の完了に失敗しました。")
                return

        finally:
            # WebSocket 音声認識サーバからの切断
            wrp.disconnect()

    def __init__(self):
        pass

    def utteranceStarted(self, startTime):
        # print("S %d" % startTime)
        pass

    def utteranceEnded(self, endTime):
        # print("E %d" % endTime)
        pass

    def resultCreated(self):
        # print("C")
        pass

    def resultUpdated(self, result):
    #変換された文章が随時送られてくる
        d = json.loads(result)
        print('\r%s' % d["text"], end='')

    def resultFinalized(self, result):
    #変換終了したら最終結果が送られる
        d = json.loads(result)
        print('\r%s' % d["text"])
        if len(d["text"])>=2:
            text2voice(cotoba(d["text"]))

    def TRACE(self, message):
        pass


#文章を音声化
def text2voice(text):
    vt = VoiceText('[VOICETEXT_API]')
    file = SPEAK_OUTPUT_FILENAME
    with open(file, 'wb') as f:
        f.write(vt.to_wave(text))
    playWav(file)


#cotobaエージェント処理
def cotoba(text):
    url = '[COTOBA_URL]'
    json_data = json.dumps({"utterance": text})
    result = requests.post(url, json_data, headers={'Content-Type': 'application/json'}).json()
    result=result["response"]
    return result

def main():
    quit_event = Event()
    if task(quit_event) is True:
        while True:
            mic_input = input_audio()
            WrpSimpleTester.main(mic_input)

if __name__ == "__main__":
    main()

python3 smart_speaker.py

「Respeaker」と話してwakeupさせた後、なにか話してみてください。device didn’t detacted などのwarninigが出るが、無視して大丈夫です。表示させたくない場合、

python3 smart_speaker.py 2>/dev/null

で無視できます。しかし、エラーが出た時に出力されないので注意してください。

最終的なディレクトリ構成は以下のようになると思います。

├── audio
│   ├── output.wav
│   └── speak.wav
├── com
│   ├── __init__.py
│   ├── __pycache__
│   │   └── __init__.cpython-37.pyc
│   └── amivoice
│       ├── __init__.py
│       ├── __pycache__
│       │   └── __init__.cpython-37.pyc
│       └── wrp
│           ├── Wrp.py
│           ├── WrpListener.py
│           ├── Wrp_.py
│           ├── __init__.py
│           └── __pycache__
│               ├── Wrp.cpython-37.pyc
│               ├── WrpListener.cpython-37.pyc
│               └── __init__.cpython-37.pyc
├── curl-ca-bundle.crt
├── respeaker
│   ├── __init__.py
│   ├── __pycache__
│   │   ├── __init__.cpython-37.pyc
│   │   ├── fft.cpython-37.pyc
│   │   ├── microphone.cpython-37.pyc
│   │   ├── pixel_ring.cpython-37.pyc
│   │   ├── player.cpython-37.pyc
│   │   ├── spectrum_analyzer.cpython-37.pyc
│   │   ├── spi.cpython-37.pyc
│   │   └── vad.cpython-37.pyc
│   ├── bing_speech_api.py
│   ├── fft.py
│   ├── gpio.py
│   ├── microphone.py
│   ├── pixel_ring.py
│   ├── player.py
│   ├── pocketsphinx-data
│   │   ├── dictionary.txt
│   │   ├── hmm
│   │   │   ├── feat.params
│   │   │   ├── mdef
│   │   │   ├── means
│   │   │   ├── noisedict
│   │   │   ├── sendump
│   │   │   ├── transition_matrices
│   │   │   └── variances
│   │   └── keywords.txt
│   ├── spectrum_analyzer.py
│   ├── spi.py
│   ├── usb_hid
│   │   ├── __init__.py
│   │   ├── __pycache__
│   │   │   ├── __init__.cpython-37.pyc
│   │   │   ├── hidapi_backend.cpython-37.pyc
│   │   │   ├── interface.cpython-37.pyc
│   │   │   ├── pyusb_backend.cpython-37.pyc
│   │   │   └── pywinusb_backend.cpython-37.pyc
│   │   ├── hidapi_backend.py
│   │   ├── interface.py
│   │   ├── pyusb_backend.py
│   │   └── pywinusb_backend.py
│   └── vad.py
├── smart_speaker.py
├── src
├── tree.txt
└── voicetext
    ├── __init__.py
    └── __pycache__
        └── __init__.cpython-37.pyc

作ってみて

これで完成です。最後にまとめです。

よかったところ

• 既製品スマートスピーカーとは比べ物にならないほどの自由度。
• この開発を通してスマートスピーカーの仕組みを理解することが出来る。

大変だったところ

• 完全な自作は大変なので、いくつかのサービスに頼りがち。
• ネット上でスマートスピーカーを自作をするための情報が少ない。

注意点

このスマートスピーカーは複数の外部サービスを使用しています。特にAPIについては、リクエスト数によって料金が発生することがあるので注意してください。

参考文献

RaspberryPi + Python3でPyaudioとdocomo音声認識APIを使ってみる
条件を１秒間満たし続けたらforから抜けたい