- 休憩
- 実装調査
作業場所を変える
二子玉川にあるカフェのテラスは、僕の好きな作業場所の一つだ。僕はメガネ型のHMDを使って作業しているので、見た目が少し怪しい。そのためか、たまに顔をじっと見てくる人もいる。でも、それはそれで悪くない。誰かに見られていると、すぐサボりがちな僕には良い監視の目になっている。ここは雨に濡れる心配もないし、屋外で作業できるのも気持ちがいい。屋内ではどうしても息が詰まり、すぐに悲しい気持ちになってしまうからだ。場所を転々と変えながら、気分転換しつつ作業を続けることが僕なりの努力の一つだ。
このカフェでは、小さな机を挟んで向こう側に人が行き交う。その人たちは様々な年代や属性で、いろんな生活を送っているように見える。一方で、こっちの僕は常に何かしらの野戦をしているような気分だ。上京してからもう10年以上経つが、ずっとそう感じている。向こう側の世界はとても幸せそうに見えるのに、こっち側はまるで別の世界のように感じる。
僕が今抱えている問題は、他の人から見たら大したことないかもしれない。でも、僕にとってはそれが人生の根幹に関わる重要な問題だった。その根幹を踏みにじられたのだから、そいつらを必ず始末する。それが僕の決意だ。そのためなら、10年くらいの時間を費やしても構わない。これまでもそのくらいの時間をかけてやってきた。目的は違うかもしれないが、再度同じことをやるだけだ。僕は特に長期戦が得意だから目的を達成できるだろう。
だから、このカフェのテラスで、気分を変えながら作業を続ける。雨に濡れず、監視の目があり、屋外という開放感もあるこの場所で、僕は自分が今できる事に取り組み続ける。
昨日やりたかった事の延長上
昨日やりたかった事はいくつかあるけれど、あまり達成できた事はなかった。こういう上手くいかない一日というのは結構ある。いろんな大切な事を無視して、こういった一日をこれまれにかなり積み上げてきた。他の人から見れば、無意味な事をやっているバカな奴、もしくは頭の悪い無能に見えるだろう。それらは事実だ。悲しいけれど仕方がない。それでもなぜか、こういう時間をしっかり取らないといけないような気もする。正解かどうか分からないけれど、今日もそういう日にしようと考えている。明日死ぬならと考える事もあるけれど、これしか自分にできるやりたい事がないから。
数日前から音声対話型のAIの実装がどのようになっているのかを調べている。使っている言語、ツール、ライブラリ、モデル等技術スタックを箇条書きにしてみる。
- Python
- ffmpeg
- sounddevice
- PyTorch
- nemo
- snakers4/silero-vad
- reazon-research/reazonspeech-nemo-v2
これだけの技術要素を書けば、感の良い人ならどんな事をどんな実装でやろうとしているのか想像付くと思うし、実装もできる気がする。僕はそれほどスジの良い人間ではないので、一歩ずつ足元を確認しながらここまで来た。今は、sounddeviceとreazon-research/reazonspeech-nemo-v2の間にsnakers4/silero-vadを差し込みたいのだけれど上手いかず、昨日はそこで力尽きた。
幸いにも、今日目が覚めて、しかも未だ事を起こす必要はなかった。その時が来たら全て始末するけれど、まだ少しだけ猶予があるようだった。そのため、昨日の続きに取り組む事にした。まずはファイルをコピーしながら、やった事を確認していく。
まずはマイクからのデータを取得しプログラムに渡す手段が必要だ。当初はffmpegを使用して、それを標準出力に書き出していた。標準出力に書き出す事で、他のプログラムが標準入力からデータを受け取れるようになる。いわゆるUNIX哲学にのっとった実装だ。macOS上でffmpegを実行するのであれば、AVFoundationを経由してデータを取得できる。
ffmpeg -f avfoundation -i ":2" -f wav -このデータをPythonで受け取る場合 sys.stdin から受け取るのではなく、ファイルディスクリプタから標準入力を自前で開く必要があった。
import sys, time
fd = sys.stdin.fileno()
with open(fd, "rb", buffering=0) as stdin:
while True:
buf = stdin.read(10240)
if not buf:
time.sleep(0.2)
continueこの処理方法でも、やりたい事はできるのだろうけれどWAVデータの処理のどこかが正しくないらしく、後の工程で上手くいかず行き詰まってしまった。
speech2speech-japaneseの実装済みコードを確認してみると、sounddeviceというライブラリを使ってデータを取得している。そこで、今度はこの方法を試してみる事にした。sounddeviceを使ってデータを取得するのは非常に簡単だ。
import time
import sounddevice
def callback_func(indata, outdata, frames, time, status):
print(type(indata))
with sounddevice.Stream(samplerate=16000, dtype="int16",
channels=1, blocksize=512,
callback=callback_func):
while True:
time.sleep(0.01)このデータを reazon-research/reazonspeech-nemo-v2 に処理させたが、それでは意味のある言葉を文字起こしする事ができなかった。speech2speech-japaneseでは snakers4/silero-vad によって発話検出させている。発話部分を切り取り、それを reazon-research/reazonspeech-nemo-v2 に渡す事で、文字起こしの精度を向上できるはずだ。そこで音声ファイルの発話検出を実装してみる。
import torch
torch.set_num_threads(1)
SAMPLING_RATE = 16000
model, utils = torch.hub.load("snakers4/silero-vad", "silero_vad")
(get_speech_timestamps,
save_audio,
read_audio,
VADIterator,
collect_chunks) = utils
wav = read_audio("shit.wav", sampling_rate=SAMPLING_RATE)
speech_probs = []
window_size_samples = 512 if SAMPLING_RATE == 16000 else 256
model.reset_states()
for i in range(0, len(wav), window_size_samples):
chunk = wav[i: i+window_size_samples]
if len(chunk) < window_size_samples:
break
speech_prob = model(chunk, SAMPLING_RATE).item()
speech_probs.append(speech_prob)speech_probs は浮動小数点数のリストとなる。昨日は、ここで力尽きた。=speech_probs= をどのように扱えば発話検出になるのか、そしてsounddeviceからどうやってデータを渡せばいいかがよく分かっていない。
昨日は、ここで力尽きた。さて、どちらから取り組もうか。とりあえずこの speech_probs に保持している浮動小数点数が何なのかを調べる事にした。
Speech Probability (`speech_probs`) の意味
`speech_probs` は各時間窓（ここではサンプリングレートが16000Hzの場合512サンプル単位）ごとの音声の確率を表します。それぞれの浮動小数点数は、指定されたウィンドウ内の音声がスピーチである確率を示しています。確率は0から1の範囲内であり、1に近いほど該当ウィンドウがスピーチである可能性が高いことを示します。
なるほど。ちなみに chunk は torch.Tensor のインスタンスだった。これを部分音声として分割しようとすると、次のようになる。
import torch
torch.set_num_threads(1)
SAMPLING_RATE = 16000
model, utils = torch.hub.load("snakers4/silero-vad", "silero_vad")
(get_speech_timestamps,
save_audio,
read_audio,
VADIterator,
collect_chunks) = utils
wav = read_audio("shit.wav", sampling_rate=SAMPLING_RATE)
speech_probs = []
window_size_samples = 512 if SAMPLING_RATE == 16000 else 256
model.reset_states()
speech_intervals = []
current_start = None
for i in range(0, len(wav), window_size_samples):
chunk = wav[i: i+window_size_samples]
if len(chunk) < window_size_samples:
break
speech_prob = model(chunk, SAMPLING_RATE).item()
if speech_prob > 0.5: # 閾値
if current_start is None: # 閾値越えで発話検出されていない => 開始地点を保存
current_start = i
elif current_start is not None: # 閾値越えておらず発話検出中 => 開始地点と終了地点を会話として保存
speech = wav[current_start:i]
# speech_intervals.append((current_start, i))
current_start = None
if current_start is not None: # 音声終了で発話検出中 => 開始地点と終了地点を会話として保存
speech = wav[current_start:len(wav)]
# speech_intervals.append((current_start, len(probs)))このデータを次のモデルであるreazon-research/reazonspeech-nemo-v2に渡して文字起こしを行う。
音声ファイルを文字起こしするコードは、昨日次のように実装して確認した。
import sys
from reazonspeech.nemo.asr.audio import audio_from_path
from reazonspeech.nemo.asr.transcribe import transcribe, load_model
from reazonspeech.nemo.asr.interface import AudioData
from nemo.collections.asr.models import EncDecRNNTBPEModel
audio: AudioData = audio_from_path(sys.argv[1])
model = EncDecRNNTBPEModel.from_pretrained(
'reazon-research/reazonspeech-nemo-v2',
map_location="cpu")
ret = transcribe(model, audio)
for ts in ret.segments:
print(ts, end="")
print("\n")このデータの橋渡しをする必要がある。AudioDataは以下の実装となっている。
@dataclass
class AudioData:
"""Container for audio waveform"""
waveform: np.float32
samplerate: intaudio_from_path は指定したパスからファイルを読み出して AudioData を返す。この関数は reazonspeech/nemo/asr/audio.py で実装されており、audio_to_xxxという名前の似たような関数が定義されている。それらは結局全て AudioData を返すが、何からデータを受け取るかという所が異なる。今回の場合、torch.TensorからAudioDataを作りたい。関数を確認していると、それっぽい audio_from_tensor という関数があったため、これを使う事にした。そして、それを reazon-research/reazonspeech-nemo-v2 に処理させるように実装を修正した。
import torch
from reazonspeech.nemo.asr.audio import audio_from_tensor
from nemo.collections.asr.models import EncDecRNNTBPEModel
from reazonspeech.nemo.asr.transcribe import transcribe
torch.set_num_threads(1)
SAMPLING_RATE = 16000
model, utils = torch.hub.load("snakers4/silero-vad", "silero_vad")
rez_model = EncDecRNNTBPEModel.from_pretrained(
'reazon-research/reazonspeech-nemo-v2',
map_location="cpu")
(get_speech_timestamps,
save_audio,
read_audio,
VADIterator,
collect_chunks) = utils
wav = read_audio("zun.wav", sampling_rate=SAMPLING_RATE)
speech_probs = []
window_size_samples = 512 if SAMPLING_RATE == 16000 else 256
model.reset_states()
speech_intervals = []
current_start = None
for i in range(0, len(wav), window_size_samples):
chunk = wav[i: i+window_size_samples]
if len(chunk) < window_size_samples:
break
speech_prob = model(chunk, SAMPLING_RATE).item()
if speech_prob > 0.5: # 閾値
if current_start is None: # 閾値越えで発話検出されていない => 開始地点を保存
current_start = i
elif current_start is not None: # 閾値越えておらず発話検出中 => 開始地点と終了地点を会話として保存
speech = wav[current_start:i]
audio = audio_from_tensor(speech, SAMPLING_RATE)
ret = transcribe(rez_model, audio)
for ts in ret.segments:
print(ts, end="")
print("\n")
current_start = None
if current_start is not None: # 音声終了で発話検出中 => 開始地点と終了地点を会話として保存
speech = wav[current_start:len(wav)]
audio = audio_from_tensor(speech, SAMPLING_RATE)
ret = transcribe(rez_model, audio)
for ts in ret.segments:
print(ts, end="")
print("\n")餃子
息子と餃子を食べに行った。誰かと食べる食事はよいものだ。良い思い出作りとなった。帰ってくると、悲しい手紙が届いていた。誰かのために一生懸命書いた文章が、このような形で使われる事が悲しい。