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 /sf/ask/291212281/

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。