« ^ »

今日やった事 - 20240823

所要時間: 約 8分

学校で学んだ事は、案外役に立たないわけじゃない。

所詮メモの大原則

ブログを読んでいたら、ちょっと好きな文章に出会った。ブログの本文ではなく、脚注として添えられていた文章だけれど、自分も同じだったし確かにそうだとも思った。

挫折したアドベントカレンダーもそうだけど、やはり連載宣言はするものじゃない.自分のような興味関心がブレまくる人間が連載宣言をしても、宣言した時をピークにモチベーションは下がり続ける.インプットのためのアウトプット、所詮メモの大原則を忘れている.

「世にも恐ろしいSIGPIPE、ソケットプログラミングの落とし穴」(百日半狂乱 - Shut the fuck up and write some code!!)から引用1

macOS上のffmpegで使用可能なデバイスの一覧を確認する

ffmpeg -f avfoundation -list_devices true -i ""
[AVFoundation indev @ 0x157e05340] AVFoundation video devices:
[AVFoundation indev @ 0x157e05340] [0] FaceTime HDカメラ
[AVFoundation indev @ 0x157e05340] [1] Capture screen 0
[AVFoundation indev @ 0x157e05340] AVFoundation audio devices:
[AVFoundation indev @ 0x157e05340] [0] MacBook Airのマイク
[AVFoundation indev @ 0x157e05340] [1] XXXXXX
[AVFoundation indev @ 0x157e05340] [2] 外部マイク

ffmpegを使ってAVFoundationデバイスから音声をキャプチャし標準出力に書き出す

`ffmpeg`を使ってAVFoundationデバイスから音声をキャプチャし、WAVフォーマットで標準出力に出力するには、以下のようなコマンドを使用します。ここでは、外部マイクデバイスを選択し、WAVフォーマットで標準出力に出力します。

```sh ffmpeg -f avfoundation -i ":2" -f wav - ```

このコマンドについて簡単に説明します:

  1. `ffmpeg`: コマンドの呼び出し。
  2. `-f avfoundation`: AVFoundationフォーマットを指定。
  3. `-i ":2"`: 入力デバイスとして外部マイク(リストでインデックス2)を指定。AVFoundationでは、音声デバイスを指定するために `":index"` の形式を使用します。
  4. `-f wav`: 出力フォーマットをWAVに指定。
  5. `-`: 出力を標準出力に指定。

このコマンドを実行すると、外部マイクからキャプチャされた音声がWAVフォーマットで標準出力に出力されます。例えば、これを他のプログラムやパイプで処理することもできます。

ffmpegで指定可能な形式を確認する

ffmpeg -formats

ffmpegには入力と出力があり、それぞれサポートする形式が異なる。入力をデマクシング(Demuxing)と言い、サポートしている場合は D と表示される。出力はマクシング(Muxing)と言い、サポートしている場合は E と表示される。

発話を検出する

マイクへの入力を、ffmpegを用いてAVFoundation経由で取得し、それを元に文字に変換しようとしたが、「ピ!」や「あ」といったような意味不明な文字に変換され続け、上手く動かなかった。

speech-to-speech-japaneseのソースコードを読んでみるとVADHandlerというものが実装されていた。僕はVADというものを知らなかったのだが、どうやらVoice Activity Detectionの頭文字のようだ。つまり発話検出といった所だろうか。内部ではPyTorchとsnakers4/silero-vadを使って実現されているようだった。ある程度の実装を確認できたので、自分でも実装してみる事にした。

"""
ffmpeg -f avfoundation -i ":2" -f wav - | python3 -u voice_activity_detection.py
"""
import time
import sys

import numpy as np
import torch

current_sample = 0
temp_end = 0
threshold = 0.5
triggered = False

speech_list = []
buffer = []

model, _ = torch.hub.load("snakers4/silero-vad", "silero_vad")
model.reset_states()

fd = sys.stdin.fileno()

def frame_generator(signal, frame_size):
    for i in range(0, len(signal), frame_size):
        yield signal[i:i + frame_size]

with open(fd, "rb", buffering=0) as stdin:
    while True:
        buf = stdin.read(10240)
        if not buf:
            time.sleep(0.2)
            continue
        # sound data
        audio_int16 = np.frombuffer(buf, dtype=np.int16)

        if np.abs(audio_int16).max() > 0:
            mask_value = (1 / 32768)
        else:
            mask_value = 1

        audio_float32 = (audio_int16.astype('float32') * mask_value).squeeze()

        # Ensure the correct shape for the model, i.e., (N, 512) for 16000Hz
        sampling_rate = 16000  # Ensure this is constant for the used model
        frame_size = 512 if sampling_rate == 16000 else 256

        # Generate frames with the appropriate size
        frames = list(frame_generator(audio_float32, frame_size))

        for frame in frames:
            if len(frame) != frame_size:
                continue

            x = torch.from_numpy(frame).float().unsqueeze(0)  # Ensure single batch

            with torch.no_grad():
                speech_prob = model(x, sampling_rate).item()

            current_sample += frame_size
            min_silence_samples = sampling_rate * 10 / 1000

            if (speech_prob >= threshold) and temp_end:
                temp_end = 0

            if (speech_prob >= threshold) and not triggered:
                triggered = True
                continue

            if (speech_prob < threshold - 0.15) and triggered:
                if not temp_end:
                    temp_end = current_sample
                if current_sample - temp_end < min_silence_samples:
                    continue
                else:
                    # end of speak
                    temp_end = 0
                    triggered = False
                    spoken_utterance = buffer
                    buffer = []
                    speech_list.append(spoken_utterance)
                    print(f"SPEECH!!: {type(spoken_utterance)}")
                    import sys
                    sys.exit(0)
                    continue

            if triggered:
                buffer.append(x)

以下のように実行する。

ffmpeg -f avfoundation -i ":2" -f wav - | python3 -u voice_activity_detection.py

ただしここでは、発話検出を行うだけであって、文字起こしまではしない。

力を使うために訓練を通して覚悟を育む

覚悟がなければ、力を手に入れたとしても力を使う事はできない。覚悟は、日々の訓練でしか育む事はできない。

日本語の会話を文字起こしする

日本語の会話を文字起こしする ReazonSpeech というツールがある。このツールはオープンソースであり、商用利用も可能だ。

GitHubからソースコードを取得する。

git clone https://github.com/reazon-research/ReazonSpeech
ReazonSpeechのリポジトリをクローンする

取得したソースコードからパッケージをインストールする。k2-asr、espnet-asr、espnet-onesegなどを指定して使う事もできる。

pip install ReazonSpeech/pkg/nemo-asr
ReazonSpeechをインストールする

インストールが完了すると reazonspeech-nemo-asr というコマンドが使用できるようになる。まずはこのコマンドから試す。

引数にはwavファイルを指定する。まずはVOICEVOXで出力したWAVファイルを指定した。

reazonspeech-nemo-asr zun.wav
[00:00:00.000 --> 00:00:02.140] ぼくたちはまいんボンボンなのだ。

1字1句違わず、テキスト化できた。すごい。では次に文章を読み、それを拾ってWAVファイルを作る。

マイクテスト、右、左、上、下、はい、いいえ、終了

マイクからの入力をffmpegでAVFoundation経由で取得し、WAVファイルを作った。

ffmpeg -f avfoundation -i ":2" -f wav - > foo.wav

作成したファイルをreazonspeech-nemo-asrに読み込ませる。

reazonspeech-nemo-asr ./foo.wav
[00:00:01.980 --> 00:00:02.940] ライト右。
[00:00:06.780 --> 00:00:08.460] ナイクジャスト右。
[00:00:09.419 --> 00:00:09.980] 上。
[00:00:10.700 --> 00:00:11.179] 下。
[00:00:12.540 --> 00:00:13.099] 下り。
[00:00:13.900 --> 00:00:14.460] はい。
[00:00:15.259 --> 00:00:15.820] いいえ。
[00:00:16.859 --> 00:00:17.500] 終了。

結果は微妙だ。しかし、このWAVファイルにはかなりノイズが入っており、また録音の環境やマイクもあまり良い状態ではない。ノイズを除去できれば、テキスト化の精度も上がりそうだ。まあ、あたりまえか。

音声データを伝達する方法について考える

マイクに入ってくるデータは、データを拾うロジックが動き続けていれば、ずっとデータが入ってくる。しかし、そのデータにはノイズが入っているし、そもそも必要なのはテキストであって音声ではないため、何段階かの処理を行う必要がある。それらの処理の多くは時間がかかるものが多いため、それぞれ独立して動作する事が望ましそうだ。

つまりマイクのデータを適度に区切ってデータストアに保存する。そのデータストアに入ったデータを加工し、次のデータストアに保存する。そういった一連のデータの流れを作れる事が望ましい。=speech-to-speech-japanese= は、実際threadingモジュールとqueueを使って、プロセス内でそのように実装していた。

Pythonでスレッドを使う

スレッドを使うのであれば、threading.Thread()のtargetに呼び出し可能オブジェクトを指定する方法が一番シンプルだろう。Pythonの公式ドキュメントのthreadingモジュールのはじめの方にも同じような説明が記載されている。

WIP 音声ファイルを読み込み文字起こしするプログラムを実装する

# ffmpeg -f avfoundation -i ":2" -f wav - | python3 -u rez.py
import sys
import numpy as np
from reazonspeech.nemo.asr import load_model, transcribe, audio_from_numpy

model = load_model()

fd = sys.stdin.fileno()

with open(fd, "rb", buffering=0) as stdin:
    while True:
        buf = stdin.read(512)
        if not buf:
            continue
        data = np.frombuffer(buf, dtype=np.int16)
        audio = audio_from_numpy(data, 16000)
        transcribe_result = transcribe(model, audio)
        print(transcribe_result.text, flush=True)

torch.no_gradを指定し勾配の計算を抑止する

PyTorchは、オープンソースの機械学習用ライブラリであり広く使われている。動的計算グラフを用いて、モデルの学習中にグラフを動的に構築する。PyTorchには、no_gradを指定できる。これを指定すると勾配の計算を継続せず、モデルの重みなどが更新されなくなる。

with torch.no_grad():
    speech_prob = model(x, sampling_rate).item()

参考

旧PCのデータを退避しながら旧PCの思い出を振りかえる

作業用PCを乗り換えたので、旧PCに滞留したデータを外付HDDに退避させた。こういったデータは、はっきりいって二度と使う事はないのだけれど、それでも何かあった時のためにとっておいてある。今回はそのデータの退避を行った。

その旧作業PCは、少し思い出のあるPCだ。インテルチップを搭載した型落ちMac Book Proだ。以前、作業マシンとして使用していたPCを、ひょんな事から壊してしまった。USB Type-Cの電源ケーブルのコネクタを挟み込んだまま、ノートパソコンを閉じてしまったのだ。画面が映らなくなり、修理に出す事になるのだが、数週間は帰ってこない。そのためその間に使う作業PCが必要になる。

僕は貧乏性だから、家にはこのPCしか開発で使えそうなものがなかった。そこで渋谷のじゃんぱらに行き、ぎりぎり開発できそうなマシンを7万円で購入した。それがこの旧PCだ。その後のこのPCは使い続けた。使い勝手が良かった、というか特に問題なかった。

病める時も、健やかなる時も、忙しくて瀕死の時も、チックとパニックに苦しんでいた時も、このマシンで作業をした。お気に入りのEmacsやHHKBと同様に、一緒に戦った戦友だ。本当にありがとう。君という中古macが働いてくれたから、ここまでやってこれたよ。

でも、このPCとはお別れする。バッテリーが「もうむりぽ」と言っているからだ。これからは、次の世代の玩具として、また働いてもらおうと考えている。

今まで、本当にありがとう。