asterisk + python watchdog + whisperで言った言わないを撲滅する
言った、言わないってありますよね？こちらをどうしてもなくしたかったので設定しました。
やったことはasteriskで自動録音→Whisperで自動文字おこしです。
実際に行う場合は受電時に自動音声による録音同意の案内などが必要と思いますので、コアな部分だけこちらに掲載します。
かなり必要な設定が多岐にわたりますので、間違っていたり、落としてしまっている設定などあるかもしれません。もしも気づいた方はぜひぜひこちらにコメントください。
asteriskのインストール
こちらでソースからインストールします。
extensions.conf
該当部分のみ。MixMonitorコマンドで会話をwavファイルに書き出すことができます。
exten=> _X.,1,Set(CALLERID(num)=${CALLERID(num)})
same=>n,Set(CALLFILENAME=${STRFTIME(${EPOCH},,%Y%m%d-%H%M)}-${EXTEN}-${CALLERID(num)})
same=>n,MixMonitor(${CALLFILENAME}.wav)
same=>n,Dial(PJSIP/${EXTEN})
same=>n,Hangup()
whisperのインストール
cudaのインストール
下図のように自分の環境に応じて選んでいけば必要なコマンドが表示されています。
pytorchのインストール
こちらを参照してください。
cuda 12 ⇔ cuda 11.7 は下位互換があるようで、新しいのをインストールしても問題ありませんでした。
whisperのインストール
pip install -U openai-whisper
pyannote.audioのインストール
pyannote.audio自体のインストール
pip install pyannote.audio
huggingfaceのspeaker-diarizationを有効化する。
こちらのAPIを利用するため、APIを申請し、有効化します。
huggingfaceの登録のやり方はこちら。
pyannote.audio + whisperはこちらをご覧ください。
素晴らしい記事はこちら。
こちらのコードをコピペして文字おこし部分を作成しています。
class WhisperHandler:
def __init__(self,use_auth_token=None) -> None:
self.use_auth_token = use_auth_token
self.audio = Audio(sample_rate=16000, mono=True)
self.model = whisper.load_model("small")
self.pipeline = Pipeline.from_pretrained("pyannote/speaker-diarization",use_auth_token=self.use_auth_token)
def transcribeToText(self,path:str)-> list[str]:
if(pathlib.Path(path).suffix == ".wav"):
diarization = self.pipeline(pathlib.Path(path))
result = []
for segment, _, speaker in diarization.itertracks(yield_label=True):
waveform, sample_rate = self.audio.crop(pathlib.Path(path), segment)
text = self.model.transcribe(waveform.squeeze().numpy())["text"]
result.append(f"[{segment.start:03.1f}s - {segment.end:03.1f}s] {speaker}: {text}\n")
return result
else:
raise ValueError("file is now wav or not exist")
watchdogのインストール
pip install watchdog
コード
こちらのコードを拝借して改変しています。
on_closedメソッドをオーバーライドすることでasteriskのファイルストリームが終わった際に起動するようにしています。
また、今回はシンプルにテキストファイルに書き出していますが、こちらをデータベースに書き出すことで検索などが可能となります。
例えばfirestoreに書き出すとalgoliaによる全文検索が簡単にできるようになりますので、私はそうしています。
class MyWatchHandler(FileSystemEventHandler):
"""監視ハンドラ"""
def __init__(self,whisperer):
"""コンストラクタ"""
super().__init__()
self.whisperer = whisperer
def on_closed(self,event):
print(f"[on_closed] {event}")
if(pathlib.Path(event.src_path).suffix == ".wav"):
with open("/path/to/monitor/directory","a") as f:
now = datetime.datetime.now()
f.write(f"[{now}] wave file is closed : {event}\n")
results = self.whisperer.transcribeToText(event.src_path)
fileName = event.src_path.replace(".wav",".txt")
with open(fileName ,mode='w') as file:
for result in results:
file.write(result)
def monitor(path):
"""監視実行関数
Args:
path: 監視対象パス
"""
handler = WhisperHandler(use_auth_token="")
event_handler = MyWatchHandler(handler)
observer = Observer()
observer.schedule(event_handler, "/path/to/monitor/directory", recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
def main():
"""メイン関数"""
print('monitor started')
# ===== ArgumentParserの設定
parser = ArgumentParser(description="Monitoring Tool")
# 引数の処理
parser.add_argument("-p", "--path", action="store", dest="path", help="監視対象パス")
# コマンドライン引数のパース
args = parser.parse_args()
# 引数の取得
path = args.path
# pathの指定がない場合は実行ディレクトリに設定
if path is None:
path = "."
# モニター実行
monitor(path)
if __name__ == "__main__":
main()
systemdへの登録
/etc/systemd/systemディレクトリにwhisper_service.serviceファイルを作ります。
[Unit]
Description=SFTP Sync Daemon
[Service]
User=yohei
ExecStart=/path/to/python3 /path/to/watcherdirectory/watcher.py
Type=simple
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable whisper_service.service
sudo systemctl start hoge.service
以上です。お疲れさまでした。
TODO
- たまにプロセスが動かないときがあった。→Transcribeごとにモデルを読み込んだ方が良いか？
Discussion