はじめに
リアルタイム音声処理などに使えそうなので備忘録として残しておきます。
使用機器と環境
Raspberry Pi 4 モデルB 4GB（Raspberry Pi OS、32-bit）
マイク（USB接続）
スピーカー（オーディオジャック接続）
Python 3.92
PyAudio 0.2.13
マイクとスピーカーはRaspberry Pi側で音声入出力のデフォルト設定にしておきます。
Raspberry PiへのPyAudioのインストール
公式ドキュメントのGNU/Linuxのインストール方法と同じです。
sudo apt-get install python3-pyaudio
pip install pyaudio
リアルタイム音声処理のためのソースコード
import pyaudio
import numpy as np
from typing import Callable, Tuple, Union
class AudioStream:
def __init__(self, n_channels: int = 1, sampling_rate: int = 16000,
audio_preprocessing: Union[Callable[[np.ndarray], np.ndarray]] = None,
verbose: bool = False) -> None:
self.audio = pyaudio.PyAudio()
self.n_channels = n_channels
self.sampling_rate = sampling_rate
self.audio_preprocessing = audio_preprocessing
self.verbose = verbose
self.format = pyaudio.paInt16
self.stream = None
def open(self) -> None:
self.stream = self.audio.open(format=self.format,
channels=self.n_channels,
rate=self.sampling_rate,
start=True,
input=True,
output=True,
stream_callback=self.callback)
if self.verbose:
print("Open & Start streaming")
def close(self) -> None:
self.stream.stop_stream()
self.stream.close()
self.audio.terminate()
self.stream = None
if self.verbose:
print("Stop & Close streaming")
def __enter__(self) -> "AudioStream":
self.open()
return self
def __exit__(self, exc_type, exc_value, traceback) -> None:
self.close()
def callback(self, in_data: bytes, frame_count: int,
time_info: dict, status: int) -> Tuple[bytes, int]:
if self.audio_preprocessing is not None:
in_data_array = np.frombuffer(in_data, np.int16)
out_data_array = self.audio_preprocessing(in_data_array)
out_data = out_data_array.astype(np.int16).tobytes()
else:
out_data = in_data
return (out_data, pyaudio.paContinue)
def is_active(self) -> bool:
if self.stream is None:
return False
else:
return self.stream.is_active()
使い方
以下のように使います。実行すると"Open & Start streaming"と標準出力されるので、デフォルト設定のマイクに声を入れるとスピーカーから音が出ます。
import time
n_channels = 1 # モノラル
sampling_rate = 48000 # サンプリングレート
audio_preprocessing = lambda x: x # 入力音声への前処理（ここでは何もしない設定）
with AudioStream(n_channels=n_channels, sampling_rate=sampling_rate,
audio_preprocessing=audio_preprocessing,
verbose=True) as audiostream:
while audiostream.is_active():
time.sleep(1)
サンプリングレートを48000から96000にすると、体感では遅延がなかったです。
ちなみに48000だと音ゲーには使えないかなというレベルの小さな遅延がありました。
audio_preprocessingは入力音声への前処理を定義でき、関数の入力裏と返り値はモノラルの場合、1次元のNumpy配列を想定しています。
このaudio_preprocessingをお好きに変更することでリアルタイム声質変換とかを実装できます。
おわりに
前処理による遅延さえ何とかできれば、コナン君の蝶ネクタイごっこができるかもしれませんね。
参考文献
PyAudio Documentation — PyAudio 0.2.13 documentation (Installation)
PyAudio Documentation — PyAudio 0.2.13 documentation (PyAudio.Stream)
PyAudioでマイクから入ってきた音をそのまま聞く話 - EnsekiTT Blog