实时低通过滤 wav 文件或原始输入音频并在 Python 中同时播放

Question

我想在 Python 中对音频数据进行低通滤波并同时播放。我正在寻找有关改进我的代码的建议，我将分享我当前但非常不完整的问题解决方案。尽管我请求改进它的建议，但我不会完全重写所有代码。我想借此机会学习信号处理的低级基础知识，同时更好地了解 Python 3。总的来说，我对 Python 的语法很满意，但我可能忽略了很多更有效地做事的方法。

我将在下面提供程序代码中最重要的部分。我的代码至少松散地基于我在此处阅读的答案以及我自己的一些想法（例如环形缓冲区）。虽然我的主要目标是寻求帮助，但我写这篇文章还有另一个原因。感谢这个社区为我提供的信息，我将回馈我现在所知道的，希望将所有信息集中在一个地方将帮助其他想要实现相同或类似目标的人。脚本详述如下。

首先，加载必要的内置模块。

import sys, wave, math, subprocess

一些全局变量被声明和初始化。重要的是它们是全局的，因为数据需要在过滤器函数的调用之间保持不变。钳位函数非常重要，因为没有它，从signed int 转换回s16le 将失败并出现溢出错误。我还需要将 aplay 作为子进程加载，以将原始 s16le 样本发送到处理后。我选择了aplay作为输出声音的方法，因为它本身可以方便地缓冲数据，并且使用起来超级简单，因为你只需将数据传递给它。

aplay=subprocess.Popen(('aplay','-f','cd'),stdin=subprocess.PIPE)
source=wave.open(sys.argv[1],"rb")
frameRate=source.getframerate()
frequencyRatio=(int(sys.argv[2])/frameRate)
global windowSize
windowSize=int(math.sqrt(0.196196+frequencyRatio**2)/frequencyRatio)
global bufferIndex0; bufferIndex0=0
global bufferL0; bufferL0=[]
global bufferR0; bufferR0=[]
for _ in range(windowSize+1):
    bufferL0.append(0); bufferR0.append(0
clamp = lambda n, minn, maxn: max(min(maxn, n), minn)

接下来，波形文件作为“数据”加载到内存中，“数据”被撕开并拆分为存储为 frame[] 数组的离散帧。 “数据”对象/变量现在没用了，可能占用 1GB 或更多的 RAM，因此它会被“del data”攻击。然后，帧数据循环并转换为有符号的 16 位整数。它被传递给滚动平均函数。正如你所看到的，我有它的三个副本来实现我想要的频率截止量。我还尽可能避免将数据存储在变量中，这大大减少了内存消耗并大大加快了我的代码速度。这让它从口吃变成了流畅播放，但同时消耗了一个内核上几乎所有可用的处理能力。

if __name__=="__main__":
    length=source.getnframes()
    data=source.readframes(length)
    frame=[data[_:_+4] for _ in range(0,len(data),4)]
    del data
    channel=[]
    for _ in range(length):
        channel=rollingAverage_stage2(rollingAverage_stage1(rollingAverage_stage0([int.from_bytes(frame[_][:2], byteorder='little', signed=True),int.from_bytes(frame[_][2:], byteorder='little', signed=True)])))
        aplay.stdin.write(bytearray(channel[0].to_bytes(2, byteorder='little', signed=True)+channel[1].to_bytes(2, byteorder='little', signed=True)))

这里详细介绍了“rollingAverage”函数。我显然只展示了一份副本，因为除了变量名之外它们都是相同的。额外的全局变量 ringIndex1、ringIndex2、bufferL1、bufferL2、bufferR1 和 bufferR2 分别在脚本的开头和各自的函数中声明。或许基于多个 pass 输入参数动态创建变量和“rollingAverage”类的一些实例会更好，而不是三个固定副本。

def rollingAverage_stage0(channel):
    global bufferL0; global bufferR0; global ringIndex0
    bufferL0[ringIndex0],bufferR0[ringIndex0]=channel[0],channel[1]
    channel=[clamp(int(sum(bufferL0)/windowSize),-32768,32767),clamp(int(sum(bufferR0)/windowSize),-32768,32767)]
    if ringIndex0==windowSize:
        ringIndex0=0
    else:
        ringIndex0+=1
    return channel

总结了代码。它对于低截止频率（例如：500）的效果出奇的好，但在与超过 5000Hz 的截止频率一起使用时会剪辑音频。对我的使用来说不是问题，因为我打算切断 3000Hz 及更低的语音频带的频率，我可能会选择 2000Hz 及更低的频率。我的最终脚本将使用子进程管道从 rtl_sdr 读取原始帧。该程序将像这样使用：

lowfilter.py <parameters>

rtl_sdr 是一个命令，用于控制和从基于 USB Realtek RTL2832 的软件控制无线电中获取数据。我可能还会在脚本中抛出一个 sox 子进程来执行降噪。

就像我之前提到的，我很可能不会完全重写程序，但接近完全重写的重大修改肯定在计划中。到目前为止，我已经花了几天的时间，并在此过程中学到了很多东西。

我打算处理的音频将是 NOAA 气象广播和类似的低带宽 FM 传输。如果可能的话，我还将实现某种自动增益控制和动态范围压缩，但就目前而言，过滤已经足够好，这就是我要保持目标的地方。这正是我想要实现的目标，因为它是了解更多有关 Python 的原因。

也许有些东西可以适应协程和生成器以减少资源使用。我不太了解他们，如果有的话，但我愿意学习。似乎它们可能非常适用于此应用程序。除了 numpy 和用于多线程的东西之外，我所做的任何修改都将在没有额外库的情况下更可取。我也不懂 numpy，所以这对我来说是全新的。指向初学者最佳资源的链接会有所帮助，因为即使是我发现的一些低通滤波等示例也超出了我的理解范围。这就是我编写自己的低通滤波的原因。 Numpy 可能比内置数学库快很多，所以这是我的第二个关注点。

我目前主要关注的是优化和负载平衡。通过将我的脚本分成不同的文件，并使用 subprocess.Popen 启动它们，我可以让它们作为单独的进程运行。这将不可避免地导致它们被内核放在不同的内核上。这意味着我可以将大块数据发送到标准输入，并且在获取最后一个数据块后，子进程立即转储到标准输出并将其传递给下一个子进程。这甚至可以在 bash 脚本而不是主要的 python 脚本中完成。这种优化的结果将意味着主脚本将大部分时间花在对数据进行混洗，并会导致性能大幅提升。

无论你有什么建议，我都想听听。

（请注意建议编辑的人：感谢您帮助改进我的问题中的语法和拼写。请注意名称 rollingAverage 应保持原样，因为它是函数的名称，而 subprocess.Popen 不是句子之间缺少空格。）

Answer 1

我在这个主题上得到了某人的帮助，过去几天我也做了一些自己的研究。我的问题的答案（目前）如下：

多线程对于这个应用程序来说是绝对必要的，我找到了一种非常合理的方式来实现它，而不必处理全局解释器锁。我将脚本的功能拆分为几个文件，它们通过 localhost 上的套接字相互传输数据。由于只需要一种方式的通信，因此无需实现特殊的 IPC，原始数据只需在脚本之间发送。

此处列出了 input.py 的代码，它加载 WAV 文件（并将很快从从 USB SDR 获取原始音频的命令中获取数据）。为了清楚起见，代码中添加了描述每个步骤的注释。

输入.py：

#!/bin/env python3
# Low pass filter written in python3
# (C) TheNH813 2018
# License: WTFPLv2

# Import sys for argv[], wave for io, socket for threads
import sys, wave, socket

# Start listener server on localhost:37420
listener=socket.socket()
listener.bind(("localhost",37420))
listener.listen()
# Load the file specified in argv[1] into data
source=wave.open(sys.argv[1],"rb")
data=source.readframes(source.getnframes())
# Split the frames into an array from frames
frame=[data[_:_+4] for _ in range(0,len(data),4)]
# Delete data because we have everything we need in frame[]
del data

while True:
    # Accept incoming connections, execution stops until a connection is made
    network,address=listener.accept()
    # Let user know it's working and sending data
    print("inputServer: Sending data")
    try:
        # Attempt to send data until it's all sent
        for _ in range(source.getnframes()):
            network.send(frame[_])
    # Trap all errors and restart if something goes wrong
    except BaseException as ded:
        # Let user know server encountered error or connection died
        print("inputServer: "+str(ded))

这里列出了执行过滤的代码，不再依赖于一堆全局变量。已经进行了一些优化，例如改变 ringBuffer 数组而不是来回传递它们。在我开始之前，我不知道这是可能的。数据通过多个阶段的简单滚动平均滤波器发送，以获得非常尖锐的滤波器截止。对于我的预期应用程序，这就是我所需要的，简单地删除或添加一两个通道将减少或加强效果。完成处理后，数据以 65535 帧的块形式发送到播放它的最终脚本。我无缘无故选择了 65535，除了需要超过 1 秒的缓冲区，我喜欢 2 的幂。代码在这里。

lowpass.py：

#!/bin/env python3
# Low pass filter written in python3
# (C) TheNH813 2018
# License: WTFPLv2

# Import sys for argv[], socket for multithreading, math for sqrt and sum
import sys, socket, math

# Calculating the frequency ratio is necessary for calculating window length
# Change 44100 if your sample rate is different, or get as aparameter with sys.argv[]
frequencyRatio=int(sys.argv[1])/44100
windowSize=int(math.sqrt(0.196196+frequencyRatio**2)/frequencyRatio)
# This function is necessary to clamp the output value to the 16 bit signed limits
clamp = lambda n, minn, maxn: max(min(maxn, n), minn)

# This function perform the actual filtering
def lowPassFilter(channel,ringBuffer,ringIndex):
    # Set the current index of the ring buffer to the input
    ringBuffer[ringIndex]=channel
    # Average the contents of the ring buffer
    channel=int(sum(ringBuffer)/windowSize)
    # Check if ring buffer has reached end
    if ringIndex==windowSize:
        # If it has, reset index to 0
        ringIndex=0
    else:
        # Otherwise, increment by 1
        ringIndex+=1
    # Return the processed sample and current buffer position
    return channel,ringIndex

# Define the ring buffers
ringBuffer0L=[]
for _ in range(windowSize+1):
    ringBuffer0L.append(0)
# Copying is faster the iterating again
ringBuffer0R=ringBuffer0L.copy()
ringBuffer1L=ringBuffer0L.copy()
ringBuffer1R=ringBuffer0L.copy()
ringBuffer2L=ringBuffer0L.copy()
ringBuffer2R=ringBuffer0L.copy()
# Define the ring buffer indexes
ringIndex0L=0
ringIndex0R=0
ringIndex1L=0
ringIndex1R=0
ringIndex2L=0
ringIndex2R=0

# Start the server on localhost:37421, and recieve from localhost:37420
listener=socket.socket()
listener.bind(("localhost",37421))
listener.listen()
stream=socket.socket()
stream.connect(("localhost",37420))

while True:
    # Accept incoming connections
    network,address=listener.accept()
    # Let the user know it's started and working
    print("filterServer: Began sending data to "+str(address))
    try:
        # Keep looping forever unless error occurs
        while True:
            # Clear array
            frame=[]
            # Load data into array as buffer
            for _ in range(65535):
                frame.append(stream.recv(4))
            # Perform digital signal processing
            for _ in range(65535):
                # Split each frame into the left and right channels as integers
                channel=[int.from_bytes(frame[_][:2], byteorder='little',signed=True),int.from_bytes(frame[_][2:],byteorder='little',signed=True)]
                # Perform low pass filter on left channel
                channel[0],ringIndex0L=lowPassFilter(channel[0],ringBuffer0L,ringIndex0L)
                # Perform low pass filter on left channel (again)
                channel[0],ringIndex1L=lowPassFilter(channel[0],ringBuffer1L,ringIndex1L)
                # Perform low pass filter on left channel (again)
                channel[0],ringIndex2L=lowPassFilter(channel[0],ringBuffer2L,ringIndex2L)
                # Clamp left channel so it dosen't go out of bounds and cause a over/underflow error
                channel[0]=clamp(channel[0],-32768,32767)
                # Perform low pass filter on left channel
                channel[1],ringIndex0R=lowPassFilter(channel[1],ringBuffer0R,ringIndex0R)
                # Perform low pass filter on right channel (again)
                channel[1],ringIndex1R=lowPassFilter(channel[1],ringBuffer1R,ringIndex1R)
                # Perform low pass filter on right channel (again)
                channel[1],ringIndex2R=lowPassFilter(channel[1],ringBuffer2R,ringIndex2R)
                # Clamp rihgt channel so it dosen't go out of bounds and cause a over/underflow error
                channel[1]=clamp(channel[1],-32768,32767)
                # Join the integers back into a single binary frame
                frame[_]=bytearray(channel[0].to_bytes(2, byteorder='little', signed=True)+channel[1].to_bytes(2, byteorder='little', signed=True))
            # Iterate through processed data
            for data in frame:
                # Send it off to the playback script
                network.send(data)
    except BaseException as ded:
        # Let the user know if the server dies or encountered a error
        print("filterServer: "+str(ded))

最后，我们获取原始数据，然后将其扔给 aplay，aplay 会对其进行缓冲并将其写入声卡。这仅适用于安装了 ALSA 的 Linux 或其他系统。

player.py：

#!/bin/env python3
# Low pass filter written in python3
# (C) TheNH813 2018
# License: WTFPLv2

# Import socket for multithreading, subprocess for audio output
import socket, subprocess

# Connect to the filter server and recieve the packets
stream=socket.socket()
stream.connect(("localhost",37421))
# Launch aplay subprocess and have it ready to play audio
aplay=subprocess.Popen('aplay -f cd'.split(),stdin=subprocess.PIPE)

# Infinite loop
while True:
    # Clear data buffer
    data=[]
    # Fill buffer with data from server
    for _ in range(65535):
        # Request a frame of audio
        data.append(stream.recv(4))
    # Iterate through data
    for frame in data:
        # Play the frames of audio
        aplay.stdin.write(frame)

我确信我可以做更多的事情来提高代码的性能，但我现在对此很满意。我将继续应用我学到的东西来改进这个脚本，但是，我在这里发布的这个版本（当前版本）远远足以回答我原来的问题。我提供这些信息是希望它能够帮助其他人。

您可以使用该代码做任何事情，我特此在 WTFPLv2 下发布它。