数年前からやっている Donkeycarを使った自律走行カーですが、もうすこし機能を拡張しようかなと、音声で相手の接近を検知できないかを試してみました。
音声を考えたのは現在使っている Donkeycar のカメラが前にしか向いていないので、後方からの追い抜きをブロックしたらウケるおもしろいのではないか..というのが最初の動機です。
とりあえずはいきなりDonkeycarに実装しないで、単体のRaspberry Piにマイクをつけて試してみました。
なお、今回紹介する内容の Python コードは以下のGitHubリポジトリにおいてあります。
環境構築
Raspberry Pi は4B/8GB を使用してます。
近年すっかり手に入りにくくなったRaspberry Pi、一瞬去年作った スパコンもどきで使用していたRaspberry Piを使おいかと思ったのですが、たまたまスイッチサイエンスで1箱買うことができたので、壊さずにすみました..
Raspberry Pi OSは当時最新(今でもかな) Bullseye 64bitを使っています。
今回使用したのは Amazonでみつけてきたこの安いUSBマイク。選択の理由は、自分の車体はちょうど後方にRaspberry PiのUSBコネクタが向いているので、この小ささならアクセサリのじゃまにならず欲しい方向に設置できるからです。
必要なライブラリのインストール
DonkeycarはそもそもPythonベースで、Donkeycarのフレームワーク上に組み込むにはPythonでPartクラスを作成することになります。
sudo apt-get install -y build-essential python3 python3-dev python3-pip python3-virtualenv python3-numpy python3-pandas python3-pillow
Donkeycar アプリケーションはvirtualenv環境下で作成するので、実際には環境を作ってその中でライブラリをセットアップしました。
また音声異常検知にはRaspberry Pi用のTensorFlow 2.8.0を使っています。whlファイルは こちら からダウンロードしました。
:
sudo apt-get install -y libhdf5-dev libc-ares-dev libeigen3-dev gcc gfortran libgfortran5 libatlas3-base libatlas-base-dev libopenblas-dev libopenblas-base libblas-dev liblapack-dev git cython3 openmpi-bin libopenmpi-dev
:
pip install keras_applications==1.0.8 --no-deps
pip install keras_preprocessing==1.1.0 --no-deps
pip install numpy==1.22.1 -U
pip install h5py==3.6.0
pip install pybind11
pip install python_speech_features
pip install six wheel mock -U
pip install sklearn
:
wget "https://raw.githubusercontent.com/PINTO0309/Tensorflow-bin/main/previous_versions/download_tensorflow-2.8.0-cp39-none-linux_aarch64_numpy1221.sh"
chmod +x ./download_tensorflow-2.8.0-cp39-none-linux_aarch64_numpy1221.sh
./download_tensorflow-2.8.0-cp39-none-linux_aarch64_numpy1221.sh
pip install tensorflow-2.8.0-cp39-none-linux_aarch64.whl
:
Pythonからこのマイクを操作できないと意味がありません。マイクから音声データを取得するPythonライブラリは PyAudio を使いました。
PyAudioは内部でPortAudioを使用しているので、こちらもインストールします。
sudo apt-get install -y libportaudio2 libportaudiocpp0 portaudio19-dev
USBマイクの検出
PyAudioからUSBマイクを操作するには、まず搭載したUSBマウスのインデックス番号を知らないといけません。
USBマウスを挿した状態で、以下のプログラムを実行して表示内容からUSBマイクの番号をメモします。
import pyaudio
audio = pyaudio.PyAudio()
for i in range(audio.get_device_count()):
print(audio.get_device_info_by_index(i))
先のUSBマイクではありませんが、Sound Blasterを指した状態だと次のような表示がでてきます。
aultHighOutputLatency': 0.034829931972789115, 'defaultSampleRate': 44100.0}
{'index': 1, 'structVersion': 2, 'name': 'Sound Blaster Play! 3: USB Audio (hw:1,0)', 'hostApi': 0, 'maxInputChannels': 2, 'maxOutputChannels': 2, 'defaultLowInputLatency': 0.008684807256235827, 'defaultLowOutputLatency': 0.008684807256235827, 'defaultHighInputLatency': 0.034829931972789115, 'defaultHighOutputLatency': 0.034829931972789115, 'defaultSampleRate': 44100.0}
:
上記の表示ならインデックス番号は `1` であることがわかります。
## 録音
まず、対象音源を録音する必要がありますが、今回はwav形式で音声データを取得しました。
定期的にwavファイルを所定のディレクトリに保管するプログラムを別プロセスで実行しておき、音声異常検知は別のプロセスが保管先の最新音声ファイルを使って判断するしくみでつくりました。
import wave
import pyaudio
:
# 配列frames データをwavファイルにして保存
wavefile = wave.open('test_10.wav','wb')
wavefile.setnchannels(1) # チャネル:1
wavefile.setsampwidth(audio.get_sample_size(pyaudio.paInt16)) # ビットレート:int16ビット
wavefile.setframerate(44100) # サンプリングレート:44100kHz
チャネルの`1`はモノラルです。ステレオにするには`2`とします。
サンプリングレートはCDと同等の音質とものの本にはかかれている 44100 kHz にしています。
ビットレートは、wavファイルに格納される要素１つのサイズですが今回はint16ビットを使用しています（参考にしたコードもこの値だったので..）。
# PyAudio インスタンス化
audio = pyaudio.PyAudio()
# 引数情報に従って、PyAudio ストリーム生成
stream = audio.open(
format=pyaudio.paInt16, # ビットレート:int16ビット,
rate = 44100, # サンプリングレート：44100kHz
channels = 1, # チャネル:1
input_device_index = 1, # デバイスインデックス番号:1 ← 検出した数字
input = True,
frames_per_buffer=4096) # チャンク:4096
# 録音秒数
record_secs = 10
# 指定秒数の音声をchunkサイズごとに取得し、配列framesへ追加
max_count = int((44100 / 4096) * record_secs)
for i in range(0, max_count):
frames = []
# IOError対策 exception_on_overflow=False
frames.append(stream.read(chunk, exception_on_overflow=False))
wavefile.writeframes(b''.join(frames))
if i != 0 and i % 100 == 0 and debug:
print(f'wrote {i}/{max_count} frame(s).')
# ストリームの停止およびクロース
stream.stop_stream()
stream.close()
# PyAudioインスタンスの停止
audio.terminate()
上記コードでは10秒の音声を録音しています。
長時間録音するとIOErrorがたまに出てとまるので無視して続けるように`exception_on_overflow=False`を指定しています。
音声異常検知
音声異常検知は、古くから研究されている領域です。なので方法もある程度先人の方が考えてくれています。
特徴抽出
まず、音声データを音響特徴量というデータに加工します。
これにより巨大な音声データを検出したい特徴をなるべくそこなわないようにかつある程度操作しやすいサイズにしています。
以下の表は日本音響学会の学会誌より一部引用したものです。
検知したい対象に合わせて音響特徴量のアルゴリズムを変えています。今回はログフィルタバンクを使用しました。
フィルタバンクというのは音響学ではよく使われる用語で、元の音声にフィルタバンク行列のドット積をとり特徴を損なわないように次元を下げる方法です。
上表のようにさまざまな特徴量があるります。人の感性に近い音響特徴量を得たいという方はメルフィルタバンクをよくつかうそうですが、今回は`python_speech_features`の`logfbank`を使用しました。
from python_speech_features import logfbank
:
(rate,sig) = wav.read(eval_path)
input_data = logfbank(sig,44100,winlen=0.01,nfilt=20)
分類器
音響特徴量をもとに正常か異常かを判定する機能です。
むかしはOCSVNなどの数理最適化で使用されたアルゴリズムを使用していましたが、最近は機械学習モデルを使うようになってきました。
なので、音響特徴量でごりごりにサイズを落とすことはしなくて良くなっているそうです。
今回は全結合4層で真ん中で半分の次元に落としている簡単なエンコーダデコーダ型のモデルを使いました。
このモデルは入力データと出力データが同じ形式で、入力データをそのまま復元するように学習させます。
なのでモデルの真ん中で絞らないとすぐに収束してしまいます。
ですが個のモデルでは真ん中で半分のサイズになるので、情報損失が必ず発生して完全にもとには戻せません。
なので他のモデル同様トレーニング時間は長くはないですが短くないです(私のPCだと10秒データで5～10分くらい)。
このモデルのメリットは、トレーニング用の学習データとして異常音声データをわざわざ収集しなくても良いところです。
工場とかの実際にシビアに使用される現場だと異常音を録音する機会はなかなかめぐってこないとおもいます。
先人の方々はよく考えておられます。
ちなみに 評価、テストのためには異常音声データは必要です。 むしろ正常動作ではないことを教えていない以上、すべての異常ケースの音声が必要になってきます。
IT企業のソリューションとして音声異常検知を掲げているところで、正常音声だけでできますとうたっている会社も一部ありますが、個人的にはそういう会社は信用しないほうがいいと思います。
モデルの実装はTensorFlow.. というかほぼ keras ですね。
from keras.layers import Dense, BatchNormalization, Activation
from keras.models import Sequential, load_model
:
input_size = 20
:
model = Sequential()
model.add(Dense(input_size,input_shape=(input_size,)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Dense(int(input_size/2)))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Dense(input_size))
model.add(BatchNormalization())
model.add(Activation('relu'))
model.add(Dense(input_size))
model.compile(optimizer='adam', loss='mean_squared_error')
model.summary()
損失関数
損失関数はどうするかというと、入力データとモデル出力データの距離（平均二乗誤差）をとります。
距離が0に近い、つまり入力データとほぼ同じ音響特徴量である、ということです。
算出された距離にしきい値を設けて、その値より小さい場合は「正常」、そうでない場合は「異常」と判断させます。
今回は`sklearn.metrics`の`mean_squared_error`を使用しています。
from sklearn.metrics import mean_squared_error
:
# 異常判定スコア計算
score = mean_squared_error(input_data, output_data)
学習データ
今回学習データとして10秒間の音声データ(wav形式)を使用しています。
wavファイルならどこで取ったものでもトレーニング用のデータとして使えますが、本番と同じ環境で収集したものが望ましいのはいうまでもありません。
評価データ
学習データは10秒だったのですが、評価は2秒データにしています。
Donkeycarはデフォルトで1秒間に20回ループが回るようになっています。なので音声データを2秒間取り続けるわけには行きません。なので別プロセスで音声を収集していて、Donkeycarアプリ側は最新の2秒データを評価しています。
なのでループの40回くらいは同じ音声データの結果で判断してしまいます。このあたりはしきい値を緩めて早めに検知するようにして対応することになりますね。
テスト
Donkeycarに実装せずにテストしたかったので、タミヤの高速・低速ギア変更可能な2輪駆動バギーをつかってテストしました。
定期的に録音した音声の異常検知スコアをFlaskでグラフ化するWebアプリを別途動かしておき、バギーを近づけたり遠ざけたりしてみました。
単独バギーの接近検知
バギーのギア変速検知（高速→低速）
チューニングの違うバギーを聞き分け
二輪駆動四輪駆動の聞き分け
上記いずれの動画では、どれも判別できている様子がわかるとおもいます。
..が、実は上記のモデルのうち「二輪駆動四輪駆動の聞き分け」で使用した分類器以外の３つのユースケースの学習データは、何も動いていない状態で録音した無音声データを使っています。
だからわざわざ機械学習モデルなんぞつかわなくてもwavファイルをそのまま評価するだけで、無音→音ありはすぐに判別できるのです。
音声異常検知モデルは、デモ検証ではいい成績を残すことが多いですが、実際の現場のデータの場合はそうはいきません。
Donkeycarレースであっても、参加者が徐々に増えるなど現場の環境音が常時変化する場合はそのたび学習し直すことになります。
現場に合わせて音響特徴量を磨く作業が必要になります。
結論を言うと、音声異常検知はDonkeycarレースのような常に環境音に変化のある場合は難しいのです。
超音波の送受信で距離をはかるセンサなども安く手に入るので、後方に向けて設置して於けばいいだけの話なのです。
（きちんと計測してませんが）そのほうがRaspberry Piのコンピュータリソースを食わないとおもいます。
