Python 音频 FIFO 缓冲区类 - 一种算法

FIFO 缓冲区是许多用例中常用的数据结构。项目被放入缓冲区并按放入的顺序检索。这与队列的功能类似。队列和缓冲区之间的区别在于，在缓冲区中插入的数据都是相同的数据类型。此外，可以插入任意数量的数据，检索量也是任意的。

在哪些用例中，您想要放入和检索任意数量的相同类型的项目？发散到数据处理以外的领域：想想银行账户：存入的钱不同，取出的钱也不同。农场的粮食储存也是如此。但银行和粮仓就是这样做的。IT 留给我们的是音频数据，它到达的速度非常快，但必须以较慢的特定速度离开，这样才能被听到。今天造成这种情况的罪魁祸首是促进人机交互的文本转语音引擎。机器获取文本（可能来自 AI 引擎），然后将其转换为音频字节，以人类可以听到的特定速度发送给人类。正如预期的那样，机器以快速的剪辑生成音频字节，然后必须对其进行缓冲，以便能够以更慢的速度向人类传送。一个类比是您当地的加油站。文本转语音是汽油罐车，它以快速的速度泵送大量汽油，将加油站内部的油箱装满。然后以较慢的速度将它们运送到客户的汽车或其他车辆中。

总之，文本到语音（音频）的转换可以更快地进行。需要一个音频缓冲区，它接收来自文本到语音（tts）的音频以填充缓冲区。然后以人类语音的速率耗尽该缓冲区，并且人类可以理解。

音频数据：数据由一系列数字组成，这些数字表示音频信号以固定间隔（称为采样率）的值。还有通道的概念，即多个音频信号，从而产生多个值的序列。

出于本文的目的，我们将仅考虑输入侧的一个通道和输出侧的一个通道。

Numpy的：这是一种便于存储/检索数字数组的软件，其性能已优化。

设计：

要求是：能够输入任意数量的音频数据帧（一个帧代表一个音频数据点的数字），它是音频信号值的有序数组。在输出端能够检索任意数量的这些帧。当然，我们必须添加方便的功能来处理限制，即有限的缓冲区大小（导致输入时缓冲区已满）、没有可用的音频数据（输出端的缓冲区为空）。其他方便的功能将包括音频数据的零填充，以防请求的音频数据多于缓冲区检索中可用的数据。

执行：

下面描述了这种缓冲区在 Python 中的实现：

传入的音频字节存储在缓冲区中。缓冲区有一个“底部”指针，指向缓冲区的填充程度。它还有一个“起始指针”，即缓冲区的开头，新数据可以推送到该指针。起始指针固定在缓冲区的开头。底部指针是“动态”的，会“上下”移动：提取数据时向上，插入数据时向下。数据始终插入到缓冲区的顶部（起始指针），导致缓冲区中的现有数据被“向下”推送，从而增加底部指针的值。

当底部指针等于起始指针时，发生缓冲区为空的情况。当底部指针等于缓冲区长度时，发生缓冲区为满的情况。

我们还可以包括‘优雅失败’来处理缓冲区已满的情况。

当缓冲区已满且需要插入数据时，会引发异常。当缓冲区为空时（包括请求的数据多于缓冲区中可用的数据的情况），将为缺失数据返回“零”。这相当于没有说话时的“静音”的音频效果。

该缓冲区的图表如下：

编码：（免责声明：没有使用人工智能来生成以下代码。所有责任（赞扬更好）都应归咎于作者。）

遵循面向对象原则，代码以类/对象的形式编写，易于使用。完整代码如下：

 import numpy as np #numpy is the standard package or numerical array processing class AudioBuf: def __init__(self,bufSize:int,name:str='',dtype=np.int16): self.buffer = np.zeros((bufSize), dtype=dtype) self.bufSize=bufSize self.dtype=dtype self.idx=0 self.name=name #give a name to the buffer. def putInBuf(self,inData:np.ndarray[np.dtype[np.int16]]): inData=inData[:, 0] #Get the 1st col = channel 0 - mono remainder = self.bufSize - self.idx #space available for insertion if remainder < len(inData): msg=f'Error: Buffer {self.name} is full' print(msg) self.showBuf() raise ValueError(msg) self.buffer[self.idx:self.idx + len(inData)] = inData self.idx += len(inData) def getFromBuf(self,outDataLen:int = None)->np.ndarray: if not outDataLen: outDataLen=self.idx # return entire data of length is not specified if self.idx >= outDataLen: retVal = self.buffer[:outDataLen] self.buffer[0:self.idx-outDataLen]=self.buffer[outDataLen:self.idx] #move buffer up self.idx -= outDataLen else: retVal=np.zeros((outDataLen), dtype=self.dtype) retVal[0:self.idx] = self.buffer[0:self.idx] self.idx=0 return np.reshape(retVal,(-1,1)) #The -1 value automatically calculates to the number of elements def showBuf(self): print(f'AudioBuf : {self.name} has length= {self.idx} with capacity {self.bufSize}')

结论：

由于音频处理应用程序众多，音频缓冲现在变得必不可少，而且更加重要。上面介绍的音频缓冲算法是一个方便的 Python 类。