Talking Piano（リアルタイムFFT）

朗読の音声をFFTしてMIDIとしてDAWからピアノで鳴らした。

元の朗読の音源と重ねたもの。

声をリアルタイムでFFTしているバージョン。

構成

Python
pyFFTW
mido → IAC Driver → Logic

音はアプリ側で合成せず、MIDIとしてIAC Driver経由でLogicに送り、ピアノ音源に鳴らしてもらっている。

周波数からMIDIノートへ

平均律は半音ごとに周波数が2の12乗根倍になる。この比を使って、C0を起点に128音ぶんの「この範囲の周波数ならこのノート」という表を先に作っておく。

# 12th root of 2
multiplier = numpy.float_power(2.0, 1.0 / 12)

notes = {0: base}
for i in range(1, 128):
    mid  = multiplier * notes[i - 1][1]
    low  = (mid + notes[i - 1][1]) / 2.0
    high = (mid + (multiplier * mid)) / 2.0
    notes.update({i: [low, mid, high]})

境界を隣の音との中点に置いているので、検出した周波数がどのノートに属するかを引くだけで決まる。毎フレーム対数計算をする必要がない。

スペクトルからピークを拾う

窓長2048サンプル、ホップ512サンプル、サンプルレート44.1kHz。ハニング窓をかけてから実数FFTにかけている。

ピークは、しきい値を超えていて左右の隣より大きいビンを拾い、近すぎるものは間引く。

peaks = []
for i in range(1, len(spectrum)-1):
    if spectrum[i] > threshold and spectrum[i] > spectrum[i-1] and spectrum[i] > spectrum[i+1]:
        if not peaks or i - peaks[-1][0] >= min_distance:
            peaks.append((i, spectrum[i]))
peaks.sort(key=lambda x: x[1], reverse=True)

基音を決めるときは、単に一番強いピークを採らない。倍音の方が基音より強く出ることがあるので、最大ピークの半分以上の強さを持つものの中でいちばん低い周波数を基音として扱っている。

fundamental_idx = 0
for i in range(1, len(freqs)):
    if mags[i] > mags[fundamental_idx] * 0.5:
        if freqs[i] < freqs[fundamental_idx]:
            fundamental_idx = i

numpyからpyFFTWへ

リアルタイム化にあたって、FFT部分をnumpyからpyFFTWに書き換えた。

numpy.fft は呼ぶたびに内部で作業領域を確保する。窓をずらしながら秒間何十回もFFTをかける用途では、この確保と解放が積み重なってくる。

pyFFTWは、変換の計画（plan）と入出力バッファを最初に一度だけ作り、あとは同じ領域を使い回す書き方ができる。

self.input_buffer  = pyfftw.empty_aligned(self.window_size, dtype='float64')
self.output_buffer = pyfftw.empty_aligned(self.window_size//2 + 1, dtype='complex128')
self.fft_object = pyfftw.FFTW(
    self.input_buffer,
    self.output_buffer,
    direction='FFTW_FORWARD',
    flags=('FFTW_MEASURE',),
    threads=1
)

empty_aligned はSIMD命令が使えるようにメモリ境界を揃えた確保、FFTW_MEASURE は初期化時に実際にいくつかの手順を試して速い方を選ぶ指定。起動時に少し時間を使う代わりに、以降の1回あたりが軽くなる。

実行時はバッファに書き込んで呼ぶだけになる。

def _compute_fft(self, buffer: np.ndarray) -> np.ndarray:
    self.input_buffer[:] = buffer
    self.fft_object()
    return np.abs(self.output_buffer)

音声の受け渡し

マイクからの入力は一定の塊で届くが、FFTは窓長ぶんまとまってから、ホップ幅ずつずらして処理したい。この粒度の違いをリングバッファで吸収している。

書き込みと読み出しを別スレッドから触るため、ロックを持たせた。バッファ端をまたぐ場合は2回に分けて書く。

if self.write_pos + n_samples <= self.capacity:
    self.buffer[self.write_pos:self.write_pos + n_samples] = data
else:
    first_part = self.capacity - self.write_pos
    self.buffer[self.write_pos:] = data[:first_part]
    self.buffer[:n_samples - first_part] = data[first_part:]

いっぱいのときは読み取り位置の方を進めて、古い音を捨てて最新を残す。リアルタイム用途では、遅れて全部鳴らすより最新に追従する方が都合がよい。

参考

元ネタのPeter Ablingerはちょうど先月亡くなってしまった。