在 Python 中实现实时信号处理 - 如何连续捕获音频？

Question

我打算用 Python 实现一个“类似 DSP”的信号处理器。它应该通过 ALSA 捕获音频的小片段，处理它们，然后通过 ALSA 播放它们。

为了开始，我编写了以下（非常简单的）代码。

import alsaaudio

inp = alsaaudio.PCM(alsaaudio.PCM_CAPTURE, alsaaudio.PCM_NORMAL)
inp.setchannels(1)
inp.setrate(96000)
inp.setformat(alsaaudio.PCM_FORMAT_U32_LE)
inp.setperiodsize(1920)

outp = alsaaudio.PCM(alsaaudio.PCM_PLAYBACK, alsaaudio.PCM_NORMAL)
outp.setchannels(1)
outp.setrate(96000)
outp.setformat(alsaaudio.PCM_FORMAT_U32_LE)
outp.setperiodsize(1920)

while True:
    l, data = inp.read()
    # TODO: Perform some processing.
    outp.write(data)

问题是，音频“断断续续”并且不是无缝的。我尝试使用 PCM 模式，将其设置为 PCM_ASYNC 或 PCM_NONBLOCK，但问题仍然存在。我认为问题在于“在”两个后续调用“inp.read()”之间的样本丢失了。

有没有办法在 Python 中“连续”捕获音频（最好不需要太“特定”/“非标准”的库）？我希望信号总是“在后台”被捕获到某个缓冲区中，我可以从中读取一些“瞬时状态”，而即使在我执行读取操作时，音频也会被进一步捕获到缓冲区中.我怎样才能做到这一点？

即使我使用专用进程/线程来捕获音频，该进程/线程总是至少必须 (1) 从源中读取音频，(2) 然后将其放入某个缓冲区（“信号处理”进程/线程然后读取）。因此，这两个操作在时间上仍然是连续的，因此样本会丢失。我该如何避免这种情况？

非常感谢您的建议！

编辑 2： 现在我可以运行它了。

import alsaaudio
from multiprocessing import Process, Queue
import numpy as np
import struct

"""
A class implementing buffered audio I/O.
"""
class Audio:

    """
    Initialize the audio buffer.
    """
    def __init__(self):
        #self.__rate = 96000
        self.__rate = 8000
        self.__stride = 4
        self.__pre_post = 4
        self.__read_queue = Queue()
        self.__write_queue = Queue()

    """
    Reads audio from an ALSA audio device into the read queue.
    Supposed to run in its own process.
    """
    def __read(self):
        inp = alsaaudio.PCM(alsaaudio.PCM_CAPTURE, alsaaudio.PCM_NORMAL)
        inp.setchannels(1)
        inp.setrate(self.__rate)
        inp.setformat(alsaaudio.PCM_FORMAT_U32_BE)
        inp.setperiodsize(self.__rate / 50)

        while True:
            _, data = inp.read()
            self.__read_queue.put(data)

    """
    Writes audio to an ALSA audio device from the write queue.
    Supposed to run in its own process.
    """
    def __write(self):
        outp = alsaaudio.PCM(alsaaudio.PCM_PLAYBACK, alsaaudio.PCM_NORMAL)
        outp.setchannels(1)
        outp.setrate(self.__rate)
        outp.setformat(alsaaudio.PCM_FORMAT_U32_BE)
        outp.setperiodsize(self.__rate / 50)

        while True:
            data = self.__write_queue.get()
            outp.write(data)

    """
    Pre-post data into the output buffer to avoid buffer underrun.
    """
    def __pre_post_data(self):
        zeros = np.zeros(self.__rate / 50, dtype = np.uint32)

        for i in range(0, self.__pre_post):
            self.__write_queue.put(zeros)

    """
    Runs the read and write processes.
    """
    def run(self):
        self.__pre_post_data()
        read_process = Process(target = self.__read)
        write_process = Process(target = self.__write)
        read_process.start()
        write_process.start()

    """
    Reads audio samples from the queue captured from the reading thread.
    """
    def read(self):
        return self.__read_queue.get()

    """
    Writes audio samples to the queue to be played by the writing thread.
    """
    def write(self, data):
        self.__write_queue.put(data)

    """
    Pseudonymize the audio samples from a binary string into an array of integers.
    """
    def pseudonymize(self, s):
        return struct.unpack(">" + ("I" * (len(s) / self.__stride)), s)

    """
    Depseudonymize the audio samples from an array of integers into a binary string.
    """
    def depseudonymize(self, a):
        s = ""

        for elem in a:
            s += struct.pack(">I", elem)

        return s

    """
    Normalize the audio samples from an array of integers into an array of floats with unity level.
    """
    def normalize(self, data, max_val):
        data = np.array(data)
        bias = int(0.5 * max_val)
        fac = 1.0 / (0.5 * max_val)
        data = fac * (data - bias)
        return data

    """
    Denormalize the data from an array of floats with unity level into an array of integers.
    """
    def denormalize(self, data, max_val):
        bias = int(0.5 * max_val)
        fac = 0.5 * max_val
        data = np.array(data)
        data = (fac * data).astype(np.int64) + bias
        return data

debug = True
audio = Audio()
audio.run()

while True:
    data = audio.read()
    pdata = audio.pseudonymize(data)

    if debug:
        print "[PRE-PSEUDONYMIZED] Min: " + str(np.min(pdata)) + ", Max: " + str(np.max(pdata))

    ndata = audio.normalize(pdata, 0xffffffff)

    if debug:
        print "[PRE-NORMALIZED] Min: " + str(np.min(ndata)) + ", Max: " + str(np.max(ndata))
        print "[PRE-NORMALIZED] Level: " + str(int(10.0 * np.log10(np.max(np.absolute(ndata)))))

    #ndata += 0.01 # When I comment in this line, it wreaks complete havoc!

    if debug:
        print "[POST-NORMALIZED] Level: " + str(int(10.0 * np.log10(np.max(np.absolute(ndata)))))
        print "[POST-NORMALIZED] Min: " + str(np.min(ndata)) + ", Max: " + str(np.max(ndata))

    pdata = audio.denormalize(ndata, 0xffffffff)

    if debug:
        print "[POST-PSEUDONYMIZED] Min: " + str(np.min(pdata)) + ", Max: " + str(np.max(pdata))
        print ""

    data = audio.depseudonymize(pdata)
    audio.write(data)

但是，即使我对音频数据进行最轻微的修改（例如，在其中注释该行），我也会在输出端收到很多噪音和极度失真。看来我没有正确处理 PCM 数据。奇怪的是，“电平表”等的输出似乎都说得通。但是，当我稍微偏移它时，输出完全失真（但连续）。

编辑 3：我刚刚发现，当我将算法（未包括在此处）应用于波形文件时，它们可以正常工作。所以问题似乎实际上归结为 ALSA API。

EDIT 4：我终于找到了问题所在。他们如下。

1st - ALSA 在请求 PCM_FORMAT_U32_LE 时悄悄地“退回”到 PCM_FORMAT_U8_LE，因此我假设每个样本都是 4 字节宽，从而错误地解释了数据。当我请求 PCM_FORMAT_S32_LE 时它可以工作。

2nd - ALSA 输出似乎期望以 bytes 为单位的周期大小，尽管它们在规范中明确声明它期望以 frames 为单位。因此，如果使用 32 位采样深度，则必须将周期大小设置为输出的四倍。

3rd - 即使在 Python 中（有一个“全局解释器锁”），与线程相比，进程也很慢。通过更改为线程可以大大降低延迟，因为 I/O 线程基本上不做任何计算密集型的事情。

Answer 1

当你

1. 读取一大块数据，
2. 写入一大块数据，
3. 然后等待读取第二块数据，

如果第二个块不比第一个块短，则输出设备的缓冲区将变为空。

在开始实际处理之前，您应该用静音填充输出设备的缓冲区。那么输入或输出处理中的小延迟将无关紧要。

Answer 2

您可以手动完成所有操作，正如@CL 在他/她的answer 中推荐的那样，但我建议只使用 GNU Radio 改为：

它是一个框架，负责完成所有“将小块样本输入和输出算法”；它可以很好地扩展，您可以使用 Python 或 C++ 编写信号处理。

事实上，它带有一个音频源和一个音频接收器，它们可以直接与 ALSA 对话，并且只是提供/获取连续样本。我建议阅读 GNU Radio 的 Guided Tutorials；他们准确地解释了为音频应用程序进行信号处理所必需的。

一个真正最小的流程图如下所示：

您可以将高通滤波器替换为您自己的信号处理模块，或使用现有模块的任意组合。

有一些有用的东西，比如文件和 wav 文件接收器和源、过滤器、重采样器、放大器（好的、乘法器）……

Answer 3

我终于找到了问题。他们如下。

1st - ALSA 在请求 PCM_FORMAT_U32_LE 时悄悄地“退回”到 PCM_FORMAT_U8_LE，因此我假设每个样本都是 4 字节宽，从而错误地解释了数据。当我请求 PCM_FORMAT_S32_LE 时它可以工作。

2nd - ALSA 输出似乎期望以字节为单位的周期大小，即使它们明确声明它在规范中以帧为单位。因此，如果使用 32 位采样深度，则必须将周期大小设置为输出的四倍。

3rd - 即使在 Python 中（有一个“全局解释器锁”），与线程相比，进程也很慢。通过更改为线程可以大大降低延迟，因为 I/O 线程基本上不做任何计算密集型的事情。

音频现在是无缝且不失真的，但延迟太高了。