Python PyAudio + 麦克风输入 - 特定频率滤波器？

Question

我目前正在从事一个射电天文学项目，我需要随时间监测音频信号的幅度。

我在这里使用了 user1405612 建议的简化 Python 代码 Detect tap with pyaudio from live mic它获取麦克风输入并计算出 RMS 振幅，我添加了一个部分来简单地将值记录到 CSV 文件中。这工作得很好，感谢必须得到 ouser1405612!

但是，有没有一种方法可以对这段代码实现一个简单的频率滤波器。例如，我对频率 19.580khz 的 RMS 振幅感兴趣(实际上我想看看 19.4hkz 到 19.6hkz 的范围)？

有没有办法使用 PyAudio 使用上面链接中的代码通过查看原始流数据或任何其他方式来做到这一点？我不想要任何复杂的东西，比如图表、频谱分析等，只需要一个简单的频率滤波器。不幸的是，麦克风输入前的带通滤波器是不可能的，因此需要在计算机上完成。

提前致谢!

更新 - 2014 年 12 月 31 日 - 她是我当前的代码:

# source https://stackoverflow.com/questions/4160175/detect-tap-with-pyaudio-from-live-mic

import pyaudio
import struct
import math
import datetime


FORMAT = pyaudio.paInt16 
SHORT_NORMALIZE = (1.0/32768.0)
CHANNELS = 1
#RATE = 44100 
RATE = 48000 
INPUT_BLOCK_TIME = 1
INPUT_FRAMES_PER_BLOCK = int(RATE*INPUT_BLOCK_TIME)
filename = 'Data.CSV'

def get_rms(block):

    count = len(block)/2
    format = "%dh"%(count)
    shorts = struct.unpack( format, block )

    # iterate over the block.
    sum_squares = 0.0
    for sample in shorts:
    # sample is a signed short in +/- 32768. 
    # normalize it to 1.0
        n = sample * SHORT_NORMALIZE
        sum_squares += n*n

    return math.sqrt( sum_squares / count )

pa = pyaudio.PyAudio()                                 

stream = pa.open(format = FORMAT,                      
         channels = CHANNELS,                          
         rate = RATE,                                  
         input = True,                                 
         frames_per_buffer = INPUT_FRAMES_PER_BLOCK)   

errorcount = 0                                                  

for i in range(1000):
    try:                                                    
        block = stream.read(INPUT_FRAMES_PER_BLOCK)         
    except IOError, e:                                      
        errorcount += 1                                     
        print( "(%d) Error recording: %s"%(errorcount,e) )  
        noisycount = 1                                      

    amplitude = get_rms(block)
    print amplitude

    #writeCSV
    i = datetime.datetime.now()
    f = open(filename,"a")
    f.write("{},{}\n".format(i,amplitude))
    f.close()

Answer 1

SciPy 具有以数字方式带通信号所需的所有功能。

设计带通滤波器

对于这个例子，我将使用 scipy.signal.butter 设计一个三阶巴特沃斯带通滤波器:

def design_filter(lowcut, highcut, fs, order=3):
    nyq = 0.5*fs
    low = lowcut/nyq
    high = highcut/nyq
    b,a = butter(order, [low,high], btype='band')
    return b,a

运行过滤器

该函数的返回值是一组滤波器系数，可供 spicy.signal.lfilter 函数使用。您会发现的大多数示例都是对数据进行批量操作，因此它们只调用函数一次。由于您正在处理实时流，因此您的流会有点棘手。该函数将先前的过滤器状态作为参数并返回新状态。因此，您需要存储返回的状态，以便下次可以传递它。这大致就是它在现有代码中的工作方式。您需要从 get_rms 函数中重构数据规范化，这无论如何都不是一个坏主意:

def normalize(block):
    count = len(block)/2
    format = "%dh"%(count)
    shorts = struct.unpack( format, block )
    doubles = [x * SHORT_NORMALIZE for x in shorts]
    return doubles


def get_rms(samples):
    sum_squares = 0.0
    for sample in doubles:
        sum_squares += n*n
    return math.sqrt( sum_squares / count )


pa = pyaudio.PyAudio()                                 
stream = pa.open(format = FORMAT,                      
         channels = CHANNELS,                          
         rate = RATE,                                  
         input = True,                                 
         frames_per_buffer = INPUT_FRAMES_PER_BLOCK)   

errorcount = 0                                                  

# design the filter
b,a = design_filter(19400, 19600, 48000, 3)
# compute the initial conditions.
zi = lfilter_zi(b, a)

for i in range(1000):
    try:                                                    
        block = stream.read(INPUT_FRAMES_PER_BLOCK)         
    except IOError as e:                                      
        errorcount += 1                                     
        print( "(%d) Error recording: %s"%(errorcount,e) )  
        noisycount = 1          

    samples = normalize(block)                            

    bandpass_samples,zi = lfilter(b, a, samples, zi)

    amplitude = get_rms(samples)
    bandpass_ampl = get_rms(bandpass_samples)
    print(amplitude)
    print(bandpass_ampl)

抱歉，我无法运行此代码进行测试。 samples 有可能需要转换为 np.array。