⛳

websocketsでリアルタイム音声処理

2024/09/06に公開

Python

はじめに

アプリケーションから取得した音声をリアルタイムにサーバに送り処理を行う方法としてwebsocketsが使いやすいです。

例えば下記のような音声対話アプリケーションを考えます。

ユーザはスマートフォンなどの端末から音声を入力し、それをサーバ側で処理を行い、AIからの発話音声をまた端末から出力するようなユースケースを考える場合、AI発話の応答速度を早くするために、ユーザの入力音声をリアルタイムでサーバに送る必要があります。
もちろん、一旦ユーザの発話が終わるまで待ってから、まとめてAPIなどでサーバに音声を送る方法もありますが、その場合、ユーザの発話が終わるまで処理を行うことができないため、AI発話のタイミングが遅れるため、ユーザ体験が損なわれます。

したがって、今回は、websocketsを利用して、端末からリアルタイムにサーバに音声を送る部分の実装を行います。
（今回はwebsocketsの使い勝手を確認するため、webアプリなどは作らず、python同士で通信を行います）

準備

開発環境

著者の開発環境は、M2 Mac、RAM16GBです。
今後の記事は、Macを利用していることを前提に記載します。

環境構築

python 3.11を利用します。
pythonが利用できる環境にしてください。

必要パッケージのインストール

pip install pyaudio websockets soundfile numpy

コードの実装

サーバ側

server.py

import websockets
import asyncio
import numpy as np

# オーディオ設定
CHUNK = 100 
RATE = 16000

async def process_audio(websocket, path):
    print("クライアント接続中...")

    try:
        async for message in websocket:
            # 受信した音声データをnumpy配列に変換
            audio_data = np.frombuffer(message, dtype=np.int16)

            # 音量を半分にする
            processed_audio = (audio_data / 2).astype(np.int16)

            # クライアントに処理済みの音声データを送信
            await websocket.send(processed_audio.tobytes())

    except websockets.exceptions.ConnectionClosed:
        print("接続が閉じられました")
    finally:
        print("クライアント接続終了")

# WebSocketサーバーを開始
async def main():
    server = await websockets.serve(process_audio, "localhost", 8000)
    print("WebSocketサーバーが起動しました")
    await server.wait_closed()

if __name__ == "__main__":
    asyncio.run(main())

サーバ側では、クライアント側からリアルタイムに取得した音声の音量を半分にして、クライアント側にそのまま送信します。
（websocketsの使い勝手を確認したいだけなので、非常に単純な処理を実装していますが、下記の部分を必要な処理に変換することで、さまざまな処理を行えるようになります）

# 音量を半分にする
processed_audio = (audio_data / 2).astype(np.int16)

クライアント側

client.py

import pyaudio
import websockets
import asyncio
import threading
import numpy as np

# WebSocketサーバーのURL
WS_SERVER_URL = "ws://localhost:8000"

# オーディオ設定
FORMAT = pyaudio.paInt16
CHANNELS = 1
RATE = 16000
CHUNK = 100 #1024

# PyAudioを初期化
p = pyaudio.PyAudio()

# ストリームを開く（マイク入力用）
input_stream = p.open(format=FORMAT,
                      channels=CHANNELS,
                      rate=RATE,
                      input=True,
                      frames_per_buffer=CHUNK)

# ストリームを開く（再生用）
output_stream = p.open(format=FORMAT,
                       channels=CHANNELS,
                       rate=RATE,
                       output=True,
                       frames_per_buffer=CHUNK)

# グローバルフラグ
running = True

async def send_and_receive_audio(websocket):
    try:
        while running:
            # マイクから音声データを取得
            audio_data = input_stream.read(CHUNK)

            # WebSocketで音声データを送信
            await websocket.send(audio_data)

            # サーバーから処理済みの音声データを受信
            processed_audio = await websocket.recv()

            # 受信したデータを再生
            output_stream.write(processed_audio)

    except Exception as e:
        print(f"エラー: {e}")

async def main():
    async with websockets.connect(WS_SERVER_URL) as websocket:
        await send_and_receive_audio(websocket)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        running = False
    finally:
        # ストリームとPyAudioを閉じる
        input_stream.stop_stream()
        input_stream.close()
        output_stream.stop_stream()
        output_stream.close()
        p.terminate()

クライアント側では、PCのマイクから取得された音声をリアルタイムにサーバ側に送信し、サーバ側から帰ってきた音声をリアルタイムにスピーカから再生しています。

実行

環境構築済みの2つのターミナルを開き、それぞれのターミナルで下記を実行してください。

python server.py

python client.py

server.pyを実行した端末において、下記が表示されたら、マイクに向かって喋ってみてください。
リアルタイムにスピーカから自分の音声が再生されるかと思います。

WebSocketサーバーが起動しました
クライアント接続中...

まとめ

今回はリアルタイムにサーバとクライアント間で音声の転送を行うことのできるwebsocketsを使ってみました。
非常にリアルタイムに音声の転送を行うことができるため、音声対話システムなどの、リアルタイム処理の強力な味方になってくれそうです。

Webアプリなどで利用する場合は、クライアント側のpython実装を、Nextjsなどで同様に実装することで実現可能だと思います。

また、音声対話システムに興味があれば、下記の記事もぜひ

このまで読んでくださってありがとうございました！