適応逆ノッチフィルタの構成

逆ノッチフィルタの周波数応答（振幅特性）は下図のようになります。

ちょうどノッチフィルタの振幅特性を反転したような特性です。

そのため、ノッチフィルタの回路を使って実装することができます。

逆ノッチフィルタの回路は以下のようになります。

赤い破線で囲まれているのがノッチフィルタの回路です。

[px]

この理由ですが、ノッチフィルタ出力信号の周波数特性は以下のようになります、

ωcから離れるほど、振幅・位相特性は入力信号とほぼ同じになる
ωcに近づくと、振幅は急激に小さくなる

従って、入力信号からノッチフィルタ出力信号を引くと、

ωcの遠方では、逆位相の関係で打ち消しあう
ωcの近傍では、ノッチフィルタ出力信号は入力信号の振幅に影響を与えない

というように動くため、結果的に入力信号のωc近傍の成分のみを抽出することができます。

[px]

なお、式を整理すると、

となるため、以下のような回路として扱うこともできます。

直列適応逆ノッチフィルタ

適応ノッチフィルタと同様に、適応逆ノッチフィルタも直列接続することで複数の正弦波を抽出することが可能です。

注意として、直列適応逆ノッチフィルタの構成は以下のように並列回路となります。

適応ノッチフィルタを直列接続することで複数の正弦波を除去した信号を得て、その信号の逆相信号を入力信号に加えることで複数の正弦波のみが残った信号を得ることができます。

実装

結局のところ、前回の適応ノッチフィルタや直列適応ノッチフィルタを利用すればOKです。

AdaptiveNotchFilterクラスは前回記事を参考にしてください。

if __name__ == "__main__":
    filepath = "./orgel.wav"
    savefilepath = "./save.wav"
    sig, sr = sf.read(filepath, always_2d=True)

    M = 8
    bandwidth = 0.99
    anf_list = [[AdaptiveNotchFilter(sr, bandwidth=bandwidth), 
                 AdaptiveNotchFilter(sr, bandwidth=bandwidth)] for m in range(M)]

    current_frame = 0
    blocksize = 512
    tmpdata = np.zeros((blocksize, sig.shape[1]))
    tmpdata2 = np.zeros((blocksize, sig.shape[1]))
    savedata = np.zeros(sig.shape)

    def callback(outdata, frames, time, status):
        global current_frame, tmpdata, tmpdata2, anf, anf_list
        chunksize = min(sig.shape[0] - current_frame, frames)
        tmpdata[:] *= 0.0
        tmpdata[0:chunksize] = sig[current_frame:current_frame + chunksize]

        for m in range(M):
            for k in range(outdata.shape[1]):
                anf_list[m][k].online(tmpdata[:,k], tmpdata2[:,k])
            tmpdata[:] = tmpdata2
            tmpdata2[:] *= 0.0
            
        tmpdata2[0:chunksize] = sig[current_frame:current_frame + chunksize]
        outdata[:] = tmpdata2 - tmpdata
        
        if chunksize < frames:
            raise sd.CallbackStop()
        else:
            savedata[current_frame:current_frame + chunksize] = outdata[:]
        current_frame += chunksize


    event = threading.Event()
    with sd.OutputStream(
        samplerate=sr, 
        blocksize=blocksize,
        channels=sig.shape[1],
        callback=callback,
        finished_callback=event.set
    ):
        event.wait()
        
    sf.write(savefilepath, savedata, sr)