PyAudio/threading/Matplotlibでリアルタイム音声処理

　Pythonを使えば、リアルタイムの音声録音と解析も簡単に行えます。まずPyAudioで音声を録音し、次にScipyでFFT（高速フーリエ変換）を使って解析を行います。しかし、これをスムーズに行うには並列処理が欠かせません。ここでは、Pythonのthreadingモジュールを駆使して、リアルタイムで音声を録音しながらFFT結果を表示する方法を紹介します。

こんにちは。wat(@watlablog)です。今回はリアルタイムに信号処理をする方法について紹介します！

事前準備

PyAudioによる録音

　WATLABブログでは信号処理カテゴリでPCのマイクを使った録音コードを紹介しています。録音にはPyAudioというサードパーティ製の外部ライブラリを使います。PyAudioのインストールや基本的な使い方は以下の記事を参考にしてください。

Check Point： PyAudioによる録音コード

PythonのPyAudioで音声録音する簡単な方法
PyAudioのpip installやマイクの検出、波形を録音する方法を紹介！

SciPyによるFFT

　FFT（高速フーリエ変換）によって音の時間波形を周波数波形に変換できます。この記事ではリアルタイムに音を周波数分析するコードを紹介しますが、SciPyを使ったFFTについての詳細はWATLABブログの過去記事を参照してください。記事の中ではMatplotlibによるグラフ描画の方法も紹介しています。

Check Point：SciPyを使ったFFT計算コード

PythonでFFT実装!SciPyのフーリエ変換まとめ
フーリエ変換の意味や応用例、関連の処理をひとまとめに紹介！

動作環境

Windows	OS	Windows10 64bit
CPU	Intel 11th Core i7-11800H:2.3[GHz]
メモリ	16[GB]

Python	Python 3.12.3
PyAudio	0.2.14
Numpy	1.26.4
Scipy	1.13.1
Matplotlib	3.9.0

並列化の基本

リアルタイム処理に並列化が必要な理由

　今回はリアルタイムで次の2つの処理を行います。

録音
FFT→グラフ描画

　この処理を直列処理（Sequential processing）で書くと、下図のように録音が終わってからFFT計算し、その後にグラフ描画を行います。
　しかし、グラフ描画が終わるまで次の録音は始まりません。直列処理でも処理能力の高いPCを使えばリアルタイム感を演出できますが、録音と録音の間が空いてしまい信号の連続性が損なわれてしまいます。これはリアルタイムではありません。

　並列処理（Parallel processing）は並列化させたい処理をスレッド（Thread）と呼ばれるまとまりに分割します。スレッド間でデータのやり取りを行う場合は、データをキュー（Queue）と呼ばれる入れ物に格納して使います。データを格納するエンキュー（Enqueue）とデータを取りだすデキュー（Dequeue）を各スレッドで独立して行うことが特徴です。

　「録音」と「FFT→グラフ描画」の2つの処理を並列化した場合のイメージ図を下図に示します。 Thread 2ではキューにデータがない場合はFFT計算やグラフ描画を行わないような配慮も必要です。

キューは色々な場面で活用されている！

　キューを使った並列処理の仕組みは、私たちが使うコンピューターでは様々なところで活用されています。プリンターは複数の人のジョブをキューに格納し、順番に印刷（デキュー）しますし、Windows等のOSもタスクをキューに格納し順番に処理しています。

並列処理をthreadingで行うPythonコード

　 threadingを使う練習をしてみましょう。次のコードは録音とグラフ描画（プロット）の関数を用意し、キューへのデータ格納と取りだしのみを行うPythonコードです。ただし、実際に録音やグラフ描画をしない模擬コードです。このコードを眺めることで threadingの使い方を学んでみてください。

Pythonコード1

import queue
import threading
import time

# キュー
data_queue = queue.Queue()

def record_thread():
    """リアルタイムに音声を録音するスレッド"""

    while True:
        time.sleep(1)
        data = 'waveform'
        data_queue.put(data)
        print(data_queue.queue)

    return


def plot_thread():
    """波形をプロットする関数"""

    while True:
        if not data_queue.empty():
            data = data_queue.get()
            time.sleep(0.5)

    return


if __name__ == '__main__':
    """メイン文"""

    # 測定スレッド
    threading.Thread(target=record_thread, args=(), daemon=True).start()

    # プロットスレッド
    plot_thread()

import queue

import threading

import time

# キュー

data_queue = queue.Queue()

def record_thread():

"""リアルタイムに音声を録音するスレッド"""

while True:

time.sleep(1)

data = 'waveform'

data_queue.put(data)

print(data_queue.queue)

return

def plot_thread():

"""波形をプロットする関数"""

while True:

if not data_queue.empty():

data = data_queue.get()

time.sleep(0.5)

return

if __name__ == '__main__':

"""メイン文"""

# 測定スレッド

threading.Thread(target=record_thread, args=(), daemon=True).start()

# プロットスレッド

plot_thread()

1, 2行目：threadingのインポート

　Python標準ライブラリである queueと threadingをインポートしてキューを使った並列処理ができるようにしましょう。

6行目：キューの準備

　 data_queueがデータを格納するためのキューです。 queue.Queue()でキューオブジェクトを生成します。

11, 14行目：録音スレッド

　録音スレッドで .put()によるデータエンキュー（格納）を実行します。プログラムが動いている間中ずっと録音し続ける必要があるので、 while True:で無限ループをさせています。

23～25行目：プロットスレッド

　波形をプロットする関数のスレッドも while True:で無限ループをさせています。 if not data_queue.empty():はキューが空ではない時の条件分岐です。録音スレッドよりもプロットスレッドの方が高速な処理をしている場合は空の時に何も処理を行いません。
　データのデキュー（取りだし）は data_queue.get()で行います。

35, 38行目：スレッド処理

　 threading.Thread()でスレッドを設定し、 .start()でスレッド処理を開始します。 threading.Thread()の引数である targetでスレッド化したい関数を指定し、関数の引数は args=()に設定します。 daemonは Trueにしておくことでスレッドをデーモンスレッド（バックグラウンド実行）にできます。メインのプログラムが終了すると、デーモンスレッドは自動的に終了するようになります。

実行結果1

　こちらが実行結果です。録音にかかる時間を1[s]、プロットにかかる時間を0.5[s]と設定したので、録音時のエンキュー操作後は常に 'waveform'が1つだけキューに入っている状態です。適切に処理がされている状態と言えるでしょう。

deque(['waveform'])
deque(['waveform'])
deque(['waveform'])
deque(['waveform'])
.
.
.

deque(['waveform'])

Pythonコード2

　適切に処理がされない場合も模擬してみましょう。プロット部分の time.sleep()を2[s]に変更します。

import queue
import threading
import time

# キュー
data_queue = queue.Queue()

def record_thread():
    """リアルタイムに音声を録音するスレッド"""

    while True:
        time.sleep(1)
        data = 'waveform'
        data_queue.put(data)
        print(data_queue.queue)

    return


def plot_thread():
    """波形をプロットする関数"""

    while True:
        if not data_queue.empty():
            data = data_queue.get()
            time.sleep(2)

    return


if __name__ == '__main__':
    """メイン文"""

    # 測定スレッド
    threading.Thread(target=record_thread, args=(), daemon=True).start()

    # プロットスレッド
    plot_thread()

def plot_thread():

"""波形をプロットする関数"""

while True:

if not data_queue.empty():

data = data_queue.get()

time.sleep(2)

return

実行結果2

　録音スレッドでエンキューした結果、 'waveform'がどんどん増えていく結果となりました。これはデキューの処理が間に合わないことを意味しています。今回は意図的にそういう状況にしましたが、実際にアプリケーションを作った時もPCによって同じ状況になる可能性があるので注意しましょう。

deque(['waveform'])
deque(['waveform'])
deque(['waveform'])
deque(['waveform', 'waveform'])
deque(['waveform', 'waveform'])
deque(['waveform', 'waveform', 'waveform'])
deque(['waveform', 'waveform', 'waveform'])
.
.
.

deque(['waveform'])

deque(['waveform', 'waveform'])

deque(['waveform', 'waveform', 'waveform'])

リアルタイムに音声をFFTするPythonコード

録音/FFT→グラフ描画を並列化

　 threadingの使い方や注意点を学んだので、早速リアルタイムに音声をFFTするPythonコードを書いていきましょう。

Pythonコード

　キューや並列処理に関連する部分のみハイライトさせてみました。今回は先ほどのコードに「録音」や「FFT」、「グラフ描画」の機能が追加されています。

import numpy as np
import queue
import threading
import pyaudio
import matplotlib.pyplot as plt
from scipy import fftpack

# キュー
data_queue = queue.Queue()

def record_thread(index, samplerate, frames_per_buffer):
    """リアルタイムに音声を録音するスレッド"""

    # PyAudioインスタンスの生成とストリーム開始
    pa = pyaudio.PyAudio()
    stream = pa.open(format=pyaudio.paInt16, channels=1, rate=samplerate,
                     input=True, input_device_index=index, frames_per_buffer=frames_per_buffer)
    # リアルタイム録音
    try:
        while True:
            data = stream.read(frames_per_buffer, exception_on_overflow=False)
            data = np.frombuffer(data, dtype="int16") / float((np.power(2, 16) / 2) - 1)
            data_queue.put(data)
            print(len(data_queue.queue))
    finally:
        stream.stop_stream()
        stream.close()
        pa.terminate()

def calc_fft(data, samplerate):
    """FFTを計算する関数"""

    spectrum = fftpack.fft(data)
    amp = np.sqrt((spectrum.real ** 2) + (spectrum.imag ** 2))
    amp = amp / (len(data) / 2)
    phase = np.arctan2(spectrum.imag, spectrum.real)
    phase = np.degrees(phase)
    freq = np.linspace(0, samplerate, len(data))

    return spectrum, amp, phase, freq

def plot_waveform(samplerate):
    """波形をプロットする関数"""

    # プロットの設定
    plt.rcParams['font.size'] = 14
    plt.rcParams['font.family'] = 'Times New Roman'
    plt.rcParams['xtick.direction'] = 'in'
    plt.rcParams['ytick.direction'] = 'in'

    fig, (ax1, ax2) = plt.subplots(2, 1)
    ax1.yaxis.set_ticks_position('both')
    ax1.xaxis.set_ticks_position('both')
    ax2.yaxis.set_ticks_position('both')
    ax2.xaxis.set_ticks_position('both')

    ax1.set_xlabel('Time [s]')
    ax1.set_ylabel('Amplitude')
    ax2.set_xlabel('Frequency [Hz]')
    ax2.set_ylabel('Amplitude')

    ax1.set_ylim(-1, 1)
    ax2.set_yscale('log')
    ax2.set_xlim(0, 10000)
    ax2.set_ylim(0.00001, 1)

    line1, = ax1.plot([], [], label='Time waveform', lw=1, color='red')
    line2, = ax2.plot([], [], label='Amplitude', lw=1, color='blue')

    while plt.fignum_exists(fig.number):
        if not data_queue.empty():
            data = data_queue.get()
            time_axis = np.linspace(0, len(data) / samplerate, num=len(data))
            spectrum, amp, phase, freq = calc_fft(data, samplerate)

            line1.set_data(time_axis, data)
            line2.set_data(freq[:len(freq) // 2], amp[:len(amp) // 2])

            ax1.relim()
            ax1.autoscale_view()
            ax2.relim()
            ax2.autoscale_view()

            fig.tight_layout()
            try:
                plt.pause(0.01)
            except Exception as e:
                print('Error')

if __name__ == '__main__':
    """メイン文"""

    # 録音設定：サンプリングレート/フレームサイズ/マイクチャンネル
    samplerate = 44100
    frames_per_buffer = 4096
    index = 0

    # 測定スレッド
    threading.Thread(target=record_thread, args=(index, samplerate, frames_per_buffer), daemon=True).start()

    # プロットスレッド
    plot_waveform(samplerate)

100

101

102

import numpy as np

import queue

import threading

import pyaudio

import matplotlib.pyplot as plt

from scipy import fftpack

# キュー

data_queue = queue.Queue()

def record_thread(index, samplerate, frames_per_buffer):

"""リアルタイムに音声を録音するスレッド"""

# PyAudioインスタンスの生成とストリーム開始

pa = pyaudio.PyAudio()

stream = pa.open(format=pyaudio.paInt16, channels=1, rate=samplerate,

input=True, input_device_index=index, frames_per_buffer=frames_per_buffer)

# リアルタイム録音

try:

while True:

data = stream.read(frames_per_buffer, exception_on_overflow=False)

data = np.frombuffer(data, dtype="int16") / float((np.power(2, 16) / 2) - 1)

data_queue.put(data)

print(len(data_queue.queue))

finally:

stream.stop_stream()

stream.close()

pa.terminate()

def calc_fft(data, samplerate):

"""FFTを計算する関数"""

spectrum = fftpack.fft(data)

amp = np.sqrt((spectrum.real ** 2) + (spectrum.imag ** 2))

amp = amp / (len(data) / 2)

phase = np.arctan2(spectrum.imag, spectrum.real)

phase = np.degrees(phase)

freq = np.linspace(0, samplerate, len(data))

return spectrum, amp, phase, freq

def plot_waveform(samplerate):

"""波形をプロットする関数"""

# プロットの設定

plt.rcParams['font.size'] = 14

plt.rcParams['font.family'] = 'Times New Roman'

plt.rcParams['xtick.direction'] = 'in'

plt.rcParams['ytick.direction'] = 'in'

fig, (ax1, ax2) = plt.subplots(2, 1)

ax1.yaxis.set_ticks_position('both')

ax1.xaxis.set_ticks_position('both')

ax2.yaxis.set_ticks_position('both')

ax2.xaxis.set_ticks_position('both')

ax1.set_xlabel('Time [s]')

ax1.set_ylabel('Amplitude')

ax2.set_xlabel('Frequency [Hz]')

ax2.set_ylabel('Amplitude')

ax1.set_ylim(-1, 1)

ax2.set_yscale('log')

ax2.set_xlim(0, 10000)

ax2.set_ylim(0.00001, 1)

line1, = ax1.plot([], [], label='Time waveform', lw=1, color='red')

line2, = ax2.plot([], [], label='Amplitude', lw=1, color='blue')

while plt.fignum_exists(fig.number):

if not data_queue.empty():

data = data_queue.get()

time_axis = np.linspace(0, len(data) / samplerate, num=len(data))

spectrum, amp, phase, freq = calc_fft(data, samplerate)

line1.set_data(time_axis, data)

line2.set_data(freq[:len(freq) // 2], amp[:len(amp) // 2])

ax1.relim()

ax1.autoscale_view()

ax2.relim()

ax2.autoscale_view()

fig.tight_layout()

try:

plt.pause(0.01)

except Exception as e:

print('Error')

if __name__ == '__main__':

"""メイン文"""

# 録音設定：サンプリングレート/フレームサイズ/マイクチャンネル

samplerate = 44100

frames_per_buffer = 4096

index = 0

# 測定スレッド

threading.Thread(target=record_thread, args=(index, samplerate, frames_per_buffer), daemon=True).start()

# プロットスレッド

plot_waveform(samplerate)

24行目：キューのデータ数確認

　先ほどは .queueでキューの中身そのものを print()関数で出力していましたが、 len()関数を使って長さを確認するようにしました。こうすることで数値がどんどん増えていくかどうかを簡単に確認できます。

　85～88行目：例外処理

　 plt.pause()を tryブロックで囲み例外処理をしていますが、これはMatplotlibの内部エラーでウィンドウがクラッシュすることを回避しています。具体的にはリアルタイム録音中にウィンドウをドラッグしたりすると ValueError: PyCapsule_New called with null pointerが発生することがあります。

実行結果

　こちらが実行結果です。リアルタイムに録音した音声をリアルタイムに周波数波形に変換することができました。口笛を吹いたり、「あいうえお」を話したりしています（恥ずかしいので動画に音声はありません）。

　 print(len(data_queue.queue))の結果はこちらです。時々１以外の値をとっていますが、増え続けることはなく、正常にデキュー処理できていることがわかりました。

まとめ

　この記事では直列処理と並列処理の違いを説明し、リアルタイム音声処理プログラムには並列処理が必要であることを紹介しました。また、 threadingを用いた並列処理について簡単な例題を紹介し、適切な処理とならない例とその調べ方を説明しました。

　 threadingを使ってPyAudioの録音、SciPyのFFT分析、Matplotlibのグラフ表示をリアルタイム化させることに成功し、まずは一安心。今後はリアルタイムスペクトログラムコードの作成に進みたいと思います。

　リアルタイム処理のやり方がわかるとプログラミングの幅が広がるね！
　Xでも関連情報をつぶやいているので、wat(@watlablog)のフォローお待ちしています！

PyAudio/threading/Matplotlibでリアルタイム音声処理

事前準備

PyAudioによる録音

Check Point： PyAudioによる録音コード

SciPyによるFFT

Check Point：SciPyを使ったFFT計算コード

動作環境

並列化の基本

リアルタイム処理に並列化が必要な理由

キューは色々な場面で活用されている！

並列処理をthreadingで行うPythonコード

1, 2行目：threadingのインポート

6行目：キューの準備

11, 14行目：録音スレッド

23～25行目：プロットスレッド

35, 38行目：スレッド処理

リアルタイムに音声をFFTするPythonコード

録音/FFT→グラフ描画を並列化

24行目：キューのデータ数確認

85～88行目：例外処理

まとめ

コメントを残すコメントをキャンセル

最近の投稿

アーカイブ

カテゴリー

事前準備

PyAudioによる録音

Check Point： PyAudioによる録音コード

SciPyによるFFT

Check Point：SciPyを使ったFFT計算コード

動作環境

並列化の基本

リアルタイム処理に並列化が必要な理由

キューは色々な場面で活用されている！

並列処理をthreadingで行うPythonコード

1, 2行目：threadingのインポート

6行目：キューの準備

11, 14行目：録音スレッド

23～25行目：プロットスレッド

35, 38行目：スレッド処理

リアルタイムに音声をFFTするPythonコード

録音/FFT→グラフ描画を並列化

24行目：キューのデータ数確認

85～88行目：例外処理

まとめ

SNSでもご購読できます。

コメントを残す コメントをキャンセル

最近の投稿

アーカイブ

カテゴリー

　85～88行目：例外処理

コメントを残すコメントをキャンセル