[语音识别] 基于Python构建简易的音频录制与语音识别应用

0 音频基础概念

随着深度学习技术的迅猛发展端到端语音识别已广泛应用。然而音频相关的最基础概念如采样频率、采样位数我们仍需有一定了解。声音是由物体振动引起的机械波而音频是声音的电子表示。PCM (Pulse Code Modulation)编码将一种常见将模拟音频信号转换为数字形式的方法。在此过程音频采样是指在一段时间内通过固定间隔采集声音的振幅值以将连续的声音模拟信号转换为离散的数字数据。采样频率表示每秒钟采集的样本数而采样位数则表示每个样本的量化级别即声音的精细度和动态范围。关于音频详细概念介绍见数字音频基础­­­­­­­­­­从PCM说起。

采样频率

音频信号通常是连续的模拟波形为了存储它们需要将其离散化。这通过采样来实现即在固定时间间隔内测量声音信号的幅度。采样的过程就是抽取模拟信号各点的频率值。采样率越高即1秒内抽取的数据点越多音频音质就越好但同时也增加了存储和处理成本。奈奎斯特-香农Nyquist–Shannon采样定理强调采样频率必须高于信号最大频率的两倍以确保从采样值中完全恢复原始模拟信号。在音频信号采样领域常使用两个主要的采样频率16kHz和44.1kHz。
举例来说16kHz表示每秒采样16000次而人类言语声音频率范围在200Hz到8kHz16kHz的采样频率已足够捕捉人类语音频率特征同时减轻了音频数据存储和处理的负担。因此常用语音采样频率为16kHz。人耳可感知20Hz到20kHz的声音为了呈现高质量音频通常选择44.1kHz采样频率以覆盖人耳可听声音的上限。

采样位数和声道

采样后的信号是连续的模拟值为了将其转换为数字形式需要对信号进行量化。量化是将连续的模拟值映射到离散的数字值通常使用固定采样位数例如16位或24位来表示样本的幅度范围。例如使用16位16bit也就是双字节的二进制信号表示音频采样而16位的取值范围为-32768到32767共有65536个可能的取值。因此最终模拟的音频信号在幅度上被分成了65536个数值等级。较高的采样位数能够表示更大的声音幅度范围并保留更多的细节信息。常用的位深度包括8位、16位和24位其中8位是最低要求16位可以满足一般应用的需求而24位则适用于专业音频工作。

声道是指音频信号在播放系统或录音系统中的传输通道。一个声道通常对应于一个单独的音频信号源或信号流并负责传输该信号到扬声器或录音设备。在立体声系统中通常有两个声道分别是左声道和右声道用来分别处理来自音频源的左右声音信号以实现空间立体声效果。声道的概念也可扩展到多声道系统如5.1声道、7.1声道等它们可以支持更多的音频源和更丰富的音效体验比如环绕音效。

常用音频编码格式

PCM编码所获得的音频数据是最为原始的为了进行存储和网络传输需要对其进行二次编码。这些二次音频编码格式都是在PCM编码基础上再次编码和压缩的按照压缩方式又分为无损压缩和有损压缩。无损压缩是指相对于PCM编码完整地保留音频数据的音质。然而无损压缩的音频文件通常比有损压缩的音频文件稍大。有损压缩在编码过程中为了减小文件大小牺牲了部分音频数据的信息和音质。

无损压缩常见的音频编码格式有WAV/WAEVWaveform Audio File FormatFLACFree Lossless Audio CodecAIFFAudio Interchange File Format等。有损压缩常见的音频编码格式有MP3MPEG Audio Layer IIIAACAdvanced Audio CodingWMAWindows Media Audio等。

在获得编码后的音频数据后需要使用合适的文件格式来保存编码数据。一种音频编码可能对应一种文件格式也可能对应多种文件格式一般情况下是一种。例如WAV编码数据对应于.wav文件格式MP3编码数据对应于.mp3文件格式。PCM编码数据对应于.raw或者.pcm文件格式AAC编码数据对应于.acc或者.mp4文件格式等。

音频与视频概念对比

概念	音频	视频
维度	通过声波传递的声音信息是一维的	通过图像序列传递的运动图像信息是二维的
核心特征	包括音调、音量、节奏等由频率和振幅表现	包括画面内容、颜色等由像素和色彩表现
信号频率	以采样率表示,人类对声音的感知更为敏锐因而音频采样率远远大于视频帧率	以帧率表示视频是通过多张静止图像以一定的速度播放来模拟流畅的动画
采样精度	用于表示声音的幅度值常见16bit	用于表示图像的颜色和亮度值常见为8bit256色
处理技术	均衡、压缩、降噪等	剪辑、特效、编解码等
通道	以声道表示如单声道和双声道	以颜色通道表示如GRAY、RGB、RGBA
存储	便于传输和存储占用的空间较小	需要更大的存储空间和带宽来传输和保存

下面的代码展示了读取音频内容为123456789的wav文件并绘制出音频数据的波形图。

# 导入用于绘图的matplotlib库
from matplotlib import pyplot as plt
# 导入用于读取音频文件的soundfile库
# pip install soundfile
import soundfile as sf

# 从demo.wav文件中读取音频数据和采样率,data为numpy数组
data, samplerate = sf.read('asr_example_hotword.wav', dtype='float32')

# 保存音频
sf.write("output.wav", data=data, samplerate=samplerate)

# 打印音频数据的形状和打印采样率
# data为一个numpy数组samplerate为一个整数
print('data shape: {}'.format(data.shape))
print('sample rate: {}'.format(samplerate))

# 绘制音频波形
plt.figure()
plt.plot(data)
plt.show()

代码运行结果如下其中表示音频数据的样本点索引即音频的时间轴。每个样本点都对应音频数据的每一帧从左到右依次递增。纵轴表示音频信号在每个时间点的归一化后的音频强度。所读取的数据以float32表示格式数值采样值范围为-32678~32678soundfile库会除以32678(2^16/2)以归一化到[-1, 1]区间内。

1 PyAudio

1.1 PyAudio介绍与安装

PyAudio是一个用于处理音频输入和输出的Python库其主要变量和接口的实现依赖于C语言版本的PortAudio。PyAudio提供从麦克风或其他输入设备录制音频、保存音频文件、实时处理音频数据以及播放音频文件或实时音频流等功能。此外PyAudio也允许通过设置采样率、位深度、声道数等参数以及支持回调函数和事件驱动机制来满足不同应用需求。PyAudio官方网站见PyAudio。PyAudio的安装需要Python3.7及以上环境。

Windows下PyAudio安装命令如下

python -m pip install pyaudio

Linux下PyAudio按照命令如下

sudo apt-get install python3-pyaudio
python -m pip install pyaudio

本文所用PyAudio版本为0.2.13。

1.2 音频录制与播放

1.2.1 音频播放

以下代码展示了基于PyAudio播放本地音频文件。

# wave为python处理音频标准库
import wave
import pyaudio

# 定义每次从音频文件中读取的音频采样数据点的数量
CHUNK = 1024

filepath = "demo.wav"
# 以音频二进制流形式打开音频文件
with wave.open(filepath, 'rb') as wf:
    # 实例化PyAudio并初始化PortAudio系统资源
    p = pyaudio.PyAudio()

    # 打开音频流
    # format: 指定音频流的采样格式。其中wf.getsampwidth()用于获取音频文件的采样位数sample width。
    # 采样位数指的是每个采样点占用的字节数。通常情况下采样位数可以是1字节8位、2字节16位等。
    # channels指定音频流的声道数。声道数可以是单声道1或立体声2
    # rate指定音频流的采样率。采样率表示每秒钟音频采样的次数常见的采样率有44100Hz或16000Hz
    # output是否播放音频
    stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
                    channels=wf.getnchannels(),
                    rate=wf.getframerate(),
                    output=True)

    # 从音频文件播放样本数据
    while True:
        # data为二进制数据
        data = wf.readframes(CHUNK)
        # len(data)表示读取数据的长度
        # 在此len(data)应该等于采样点占用的字节数wf.getsampwidth()乘以CHUNK
        if len(data):
            stream.write(data)
        else:
            break

    # 或者使用python3.8引入的海象运算符
    # while len(data := wf.readframes(CHUNK)):
    #     stream.write(data)

    # 关闭音频流
    stream.close()

    # 释放PortAudio系统资源
    p.terminate()

1.2.2 音频录制

以下代码展示了基于PyAudio调用麦克风录音并将录音结果保存为本地文件。

import wave
import pyaudio

# 设置音频流的数据块大小
CHUNK = 1024
# 设置音频流的格式为16位整型也就是2字节
FORMAT = pyaudio.paInt16
# 设置音频流的通道数为1
CHANNELS = 1
# 设置音频流的采样率为16KHz
RATE = 16000
# 设置录制时长为5秒
RECORD_SECONDS = 5

outfilepath = 'output.wav'
with wave.open(outfilepath, 'wb') as wf:
    p = pyaudio.PyAudio()
    # 设置wave文件的通道数
    wf.setnchannels(CHANNELS)
    # 设置wave文件的采样位数 
    wf.setsampwidth(p.get_sample_size(FORMAT))
    # 设置wave文件的采样率
    wf.setframerate(RATE)

    # 打开音频流,input表示录音
    stream = p.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True)

    print('Recording...')
    # 循环写入音频数据
    for _ in range(0, RATE // CHUNK * RECORD_SECONDS):
        wf.writeframes(stream.read(CHUNK))
    print('Done')  

    stream.close()
    p.terminate()

1.2.3 全双工音频录制与播放

全双工系统full-duplex可以同时进行双向数据传输而半双工系统half-duplex只能在同一时间内进行单向数据传输。在半双工系统中一台设备传输数据时另一台设备必须等待传输完成后才能进行数据处理。以下代码展示了全双工full-duplex音频录制与播放即同时进行音频录制和播放而不需要等待一个操作完成后再进行另一个操作。

import pyaudio

RECORD_SECONDS = 5
CHUNK = 1024
RATE = 16000

p = pyaudio.PyAudio()
# frames_per_buffer设置音频每个缓冲区的大小
stream = p.open(format=p.get_format_from_width(2),
                channels=1,
                rate=RATE,
                input=True,
                output=True,
                frames_per_buffer=CHUNK)

print('recording')
for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
    # read读取音频然后writer播放音频
    stream.write(stream.read(CHUNK))
print('done')

stream.close()
p.terminate()

1.3 回调函数的使用

在前面的代码中PyAudio执行音频播放或录制是以阻塞主线程的方式进行的这意味着代码无法同时处理其他任务。为了解决这一问题PyAudio提供了回调函数使得程序在进行音频输入和输出时能够以非阻塞的方式进行操作即处理音频流的同时处理其他任务。PyAudio回调函数是在单独的线程中执行的当音频流数据可用时回调函数会被自动调用并以立即对音频数据进行处理。PyAudio回调函数具有固定的参数接口函数介绍如下

def callback(in_data,       # 录制的音频数据的字节流如果没有录音则为None
            frame_count,    # 每个缓冲区中的帧数本次读取的数据量
            time_info,      # 有关音频流时间信息的字典
            status_flags)   # 音频流状态的标志位

以下代码展示了以回调函数的形式播放音频。

import wave
import time
import pyaudio

filepath = "demo.wav"
with wave.open(filepath, 'rb') as wf:
    # 当音频流数据可用时回调函数会被自动调用
    def callback(in_data, frame_count, time_info, status):
        # 读取了指定数量的音频帧数据
        data = wf.readframes(frame_count)
        # pyaudio.paContinue为常量表示继续进行音频流的处理
        # 根据需要更改为pyaudio.paAbort或pyaudio.paComplete等常量来控制处理流程的中断和结束
        return (data, pyaudio.paContinue)

    p = pyaudio.PyAudio()

    # stream_callback设置回调函数
    stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
                    channels=wf.getnchannels(),
                    rate=wf.getframerate(),
                    output=True,
                    stream_callback=callback)
    
    # 判断音频流是否处于活动状态
    while stream.is_active():
        time.sleep(0.1)
    stream.close()

    p.terminate()

以下代码展示了如何运用回调函数实现音频录制与播放的全双工模式。在超时情况下通过调用stream.close()来关闭音频流并释放相关资源。一旦音频流被关闭将无法再传输音频数据。若想实现录音过程中暂停一段时间后再继续录音可使用stream.stop_stream()来暂停音频流的数据传输即暂时停止音频的读取和写入但仍保持流对象处于打开状态。随后可通过调用stream.start_stream()来重新启动音频流的数据传输。

import time
import pyaudio

# 录音时长
DURATION = 5 

def callback(in_data, frame_count, time_info, status):
    # in_data为麦克风输入的音频流
    return (in_data, pyaudio.paContinue)

p = pyaudio.PyAudio()
stream = p.open(format=p.get_format_from_width(2),
                channels=1,
                rate=16000,
                input=True,
                output=True,
                stream_callback=callback)

start = time.time()
#  当音频流处于活动状态且录音时间未达到设定时长时
while stream.is_active() and (time.time() - start) < DURATION:
    time.sleep(0.1)

# 超过时长关闭音频流
stream.close()
p.terminate()

1.4 设备管理

PyAudio提供了host Api和device Api来获取音频设备但host Api和device Api代表了不同的层级和功能。具体如下

• host Api是对底层音频系统的抽象表示系统上可用的音频接口提供了与底层音频设备交互的功能。每个host Api都有自己的特点和支持的功能集如使用的数据格式、采样率等。常见的host Api包括ALSA、PulseAudio、CoreAudio等。
• device Api是指具体的音频输入或输出设备如麦克风、扬声器或耳机等。每个音频设备都属于一个特定的音频host Api并具有不同的参数配置例如采样率、缓冲区大小等。

本文主要对更为常用的device Api进行介绍PyAudio中关于device Api的函数有如下

1. get_device_info_by_index(index)
   通过整数型索引获取指定设备的详细信息。该函数返回一个包含设备信息的字典包括设备名称、输入/输出通道数、支持的采样率范围等。

2. get_default_input_device_info()
   获取默认输入设备的详细信息。该函数返回一个包含设备信息的字典。

3. get_default_output_device_info()
   获取默认输出设备的详细信息。该函数返回一个包含设备信息的字典。

4. get_device_count()
   获取计算机上可用音频设备的数量这些设备可以是麦克风、扬声器、音频接口等。

其中默认设备为当前操作系统的音频默认设备可以通过操作系统音频控制页面更改默认音频输入输出设备。下面代码展示了这些函数的使用

import pyaudio

# 获取指定设备的详细信息
def get_device_info_by_index(index):
    p = pyaudio.PyAudio()
    device_info = p.get_device_info_by_index(index)
    p.terminate()
    return device_info

# 获取默认输入设备的详细信息
def get_default_input_device_info():
    p = pyaudio.PyAudio()
    default_input_info = p.get_default_input_device_info()
    p.terminate()
    return default_input_info

# 获取默认输出设备的详细信息
def get_default_output_device_info():
    p = pyaudio.PyAudio()
    default_output_info = p.get_default_output_device_info()
    p.terminate()
    return default_output_info

# 获取计算机上可用音频设备的数量
def get_device_count():
    p = pyaudio.PyAudio()
    device_count = p.get_device_count()
    p.terminate()
    return device_count

# 示例用法
index = 0
print("可用音频设备数量", get_device_count())
print("设备{}的信息{}".format(index, get_device_info_by_index(index)))
print("默认录音设备的信息", get_default_input_device_info())
print("默认播放设备的信息", get_default_output_device_info())

对于以上代码如返回的默认播放设备信息字典如下

默认播放设备的信息
{'index': 3,
 'structVersion': 2,
 'name': '扬声器 (Realtek High Definition Au',
 'hostApi': 0,
 'maxInputChannels': 0,
 'maxOutputChannels': 2,
 'defaultLowInputLatency': 0.09,
 'defaultLowOutputLatency': 0.09,
 'defaultHighInputLatency': 0.18,
 'defaultHighOutputLatency': 0.18,
 'defaultSampleRate': 44100.0}

该设备也是系统当前默认的音频设备其中各个参数的含义如下

• 'index': 3设备的索引号用于在设备列表中唯一标识该设备。
• 'structVersion': 2设备信息结构的版本号用于指示该设备信息的数据结构版本。
• 'name': '扬声器 (Realtek High Definition Au'设备的名称表示该设备是一个 Realtek High Definition 型号的扬声器。
• 'hostApi': 0设备声卡驱动模式来自于PortAudio如果想详细了解见pyaudio声卡信息中hostApi。
• 'maxInputChannels': 0设备支持的最大输入通道数这里为0表示该设备没有输入功能不支持录音。
• 'maxOutputChannels': 2设备支持的最大输出通道数这里为2表示该设备支持2个输出通道即可以播放立体声音频。
• 'defaultLowInputLatency': 0.09默认低输入延迟以秒为单位表示从音频输入信号进入设备所需的最小时间。
• 'defaultLowOutputLatency': 0.09默认低输出延迟以秒为单位表示从设备输出信号到达音频输出所需的最小时间。
• 'defaultHighInputLatency': 0.18默认高输入延迟以秒为单位表示从音频输入信号进入设备所需的最大时间。
• 'defaultHighOutputLatency': 0.18默认高输出延迟以秒为单位表示从设备输出信号到达音频输出所需的最大时间。
• 'defaultSampleRate': 44100.0默认采样率表示设备支持的默认音频采样率为44100赫兹Hz。这是音频设备在单位时间内采样的样本数影响声音的质量和频率范围。

如果想指定设备进行音频录制或录制则在open函数中指定设备的索引代码如下

import pyaudio

p = pyaudio.PyAudio()

# 获取可用的设备数量
device_count = p.get_device_count()

# 遍历设备打印设备信息和索引
for i in range(device_count):
    device_info = p.get_device_info_by_index(i)
    print(f"Device {i}: {device_info['name']}")

# 选择所需的录音设备的索引
input_device_index = 1 
# 选择所需的播放设备的索引
output_device_index = 2 

# 打开音频流并指定设备
stream = p.open(format=p.get_format_from_width(2),
                channels=1,
                rate=16000,
                input=True,
                output=True,
                input_device_index = input_device_index,
                output_device_index = output_device_index)

# 操作输出设备和录音设备
# ...

2 SpeechRecognition

2.1 SpeechRecognition介绍与安装

SpeechRecognition是一个用于语音识别的Python库支持多个语音识别引擎以将音频转换为文本。SpeechRecognition开源仓库地址为speech_recognition。基于PyAudio库SpeechRecognition封装了更加方面和全面的音频录制函数。本文主要介绍利用SpeechRecognition录制音频。使用SpeechRecognition进行音频录制需要Python3.8及以上环境以及最低PyAudio 0.2.11版本。在安装PyAudio后SpeechRecognition安装命令如下

pip install SpeechRecognition

SpeechRecognition主要的类有

AudioData

AudioData类是用于表示语音数据主要参数和函数如下

参数

• frame_data音频字节流数据
• sample_rate音频采样率
• sample_width: 音频的采样位数

函数

• get_segment返回指定时间段内的音频数据的AudioData对象
• get_raw_data返回音频数据的原始字节流
• get_wav_data返回音频数据的wav格式字节流
• get_aiff_data返回音频数据的aiff格式字节流
• get_flac_data返回音频数据的flac格式字节流

Microphone

Microphone类是封装PyAudio用于驱动麦克风设备功能的类因此构造参数与PyAudio主要参数和函数如下

参数

• device_index麦克风设备的索引号不指定将采用PyAudio的默认音频输入设置
• format采样格式为16位整数不指定将采用PyAudio的默认音频输入设置
• SAMPLE_WIDTH音频的采样位数 不指定将采用PyAudio的默认音频输入设置
• SAMPLE_RATE采样率不指定将采用PyAudio的默认音频输入设置
• CHUNK每个缓冲区中存储的帧数默认为1024
• audio PyAudio对象
• stream调用PyAudio的open函数打开的音频流

函数

• get_pyaudio用来获取PyAudio的版本号并调用PyAudio库
• list_microphone_names返回当前系统中所有可用的麦克风设备的名称列表
• list_working_microphones返回当前系统中所有正在工作的麦克风设备的名称列表。麦克风设备是否运行的评定方式为对于某设备尝试录制一段短暂的音频然后检查是否成功录制到了具有一定音频能量的音频数据。

Recognizer类

Recognizer类是用于语音识别的主要类它提供了一系列参数和函数来处理音频输入主要参数和函数如下

参数

• energy_threshold = 300: 用于录制最低音频能量基于音频均方根RMS计算能量
• dynamic_energy_threshold = True: 是否使用动态能量阈值
• dynamic_energy_adjustment_damping = 0.15: 能量阈值调整的阻尼系数
• dynamic_energy_ratio = 1.5: 动态能量比率
• pause_threshold = 0.8: 在一段完整短语被认为结束之前非语音音频的持续时间以秒为单位
• operation_timeout = None: 内部操作例如API请求开始后超时的时间以秒为单位如果不设置超时时间则为None
• phrase_threshold = 0.3: 认为一段语音至少需要的持续时间以秒为单位低于该值的语音将被忽略用于过滤噪声
• non_speaking_duration = 0.5: 非语音音频的持续时间以秒为单位

函数

• record从一个音频源中读取数据
• adjust_for_ambient_noise用于在录制音频之前自动根据麦克风的环境噪声水平调整energy_threshold参数
• listen音频录制结果返回AudioData类
• listen_in_background用于在后台录制音频并调用回调函数

Recognizer类的listen函数每次录音分为三个阶段

1. 录音起始
   这一阶段意味着开始录音但是没有声音输入。如果当前获得的声音片段能量值低于energy_threshold则认为没有声音输入。一旦当前获得的声音片段能量值高于energy_threshold则进入下一阶段。该阶段将最多保存non_speaking_duration长度的音频片段。如果dynamic_energy_threshold为True则会根据环境动态调整energy_threshold。
   listen函数提供输入参数timeout以控制该阶段时长如果录音处于该阶段timeout秒则停止录音返回错误提示timeout默认为None。

2. 录音中
   这一阶段意味着已有声音输入。如果声音片段能量值低于energy_threshold连续超过pause_threshold秒则结束录音。在这一阶段energy_threshold一直是固定值并不会进行动态调整。
   listen函数提供输入参数phrase_time_limit以控制该阶段最大时长如果录音处于该阶段phrase_time_limit秒则结束录音。

3. 录音结束
   在这一阶段中如果录音中阶段获得的声音片段时间不超过phrase_threshold秒则不返回录音结果且进入下一次录音起始阶段。如果超过phrase_threshold秒则将音频片段转为音频流以AudioData对象返回。

2.2 示例代码

音频录制

import speech_recognition as sr

# 创建一个Recognizer对象用于语音识别
r = sr.Recognizer()

# 设置相关阈值
r.non_speaking_duration = 0.3
r.pause_threshold = 0.5

# 创建一个Microphone对象设置采样率为16000
# 构造函数所需参数device_index=None, sample_rate=None, chunk_size=1024
msr = sr.Microphone(sample_rate=16000)

# 打开麦克风
with msr as source:
    # 如果想连续录音该段代码使用for循环
    # 进行环境噪音适应duration为适应时间不能小于0.5
    # 如果无噪声适应要求该段代码可以注释
    r.adjust_for_ambient_noise(source, duration=0.5)
    print("开始录音")
    
    # 使用Recognizer监听麦克风录音
    # phrase_time_limit=None表示不设置时间限制
    audio = r.listen(source, phrase_time_limit=None)
    
    print("录音结束")
    
    # 将录音数据写入.wav格式文件
    with open("microphone-results.wav", "wb") as f:
        # audio.get_wav_data()获得wav格式的音频二进制数据
        f.write(audio.get_wav_data())
    
    # 将录音数据写入.raw格式文件
    with open("microphone-results.raw", "wb") as f:
        f.write(audio.get_raw_data())

    # 将录音数据写入.aiff格式文件
    with open("microphone-results.aiff", "wb") as f:
        f.write(audio.get_aiff_data())

    # 将录音数据写入.flac格式文件
    with open("microphone-results.flac", "wb") as f:
        f.write(audio.get_flac_data())

音频文件读取

# 导入speech_recognition库别名为sr
import speech_recognition as sr

# 创建一个Recognizer对象r用于语音识别
r = sr.Recognizer()

# 设置音频文件路径
filepath = "demo.wav"

# 使用AudioFile打开音频文件作为音频源
with sr.AudioFile(filepath) as source:
    # 使用record方法记录从音频源中提取的2秒音频从第1秒开始
    audio = r.record(source, offset=1, duration=2)
    
    # 创建一个文件用于保存提取的音频数据
    with open("microphone-results.wav", "wb") as f:
        # 将提取的音频数据写入文件
        f.write(audio.get_wav_data())

回调函数的使用

import time
import speech_recognition as sr

# 这是从后台线程调用的回调函数
def callback(recognizer, audio):
    # recognizer是Recognizer对象的实例。audio是从麦克风捕获到的音频数据
    print(type(audio))

r = sr.Recognizer()
m = sr.Microphone()
with m as source:
    # 我们只需要在开始监听之前校准一次
    r.adjust_for_ambient_noise(source)

# 在后台开始监听
stop_listening = r.listen_in_background(m, callback)

# 进行一些无关的计算持续5秒钟
for _ in range(50):
    # 即使主线程正在做其他事情我们仍然在监听
    time.sleep(0.1)

# 调用此函数请求停止后台监听
stop_listening(wait_for_stop=False)

麦克风设备查看

import speech_recognition as sr

# 获取麦克风设备名称列表
def list_microphone_names():
    mic_list = sr.Microphone.list_microphone_names()
    for index, mic_name in enumerate(mic_list):
        print("Microphone {}: {}".format(index, mic_name))
    print("\n")


# 获取可用的工作麦克风列表
def list_working_microphones():
    mic_list = sr.Microphone.list_working_microphones()
    for index, mic_name in mic_list.items():
        print("Microphone {}: {}".format(index, mic_name))
    print("\n")


# 获得pyaudio对象
def get_pyaudio():
    audio = sr.Microphone.get_pyaudio().PyAudio()
    # 获取默认音频输入设备信息
    print(audio.get_default_input_device_info())
    print("\n")
    return audio


print("所有麦克风列表")
list_microphone_names()
print("可运行麦克风列表")
list_working_microphones()
print("默认音频输入设备信息")
get_pyaudio()

3 语音识别示例应用

本示例给出一个基于SpeechRecognition库和Whisper语音识别库的非流式语音识别示例应用。一般来说语音识别分为流式语音识别和非流式语音识别

• 流式语音识别是指在语音输入过程中实时进行语音识别即边接收语音数据边输出识别结果实现实时性较高的语音识别。在流式语音识别中语音被分割成一小段一小段的流可以通过连续发送这些流来实时地获取识别结果。随着语音输入的增加流式语音识别也可以优化输出部分结果。流式语音识别准确率相对较低但是实时性强适用于需要快速响应的场景例如实时语音助手、电话客服、会议记录等。技术上流式语音识别需要实时处理音频流要求算法具有低延迟和高吞吐量的特点通常使用各种优化策略来提高实时性能。
• 非流式语音识别是指等待语音输入结束后将完整的语音输入一次性进行分析和识别。非流式语音识别精度高适用于一些不需要实时响应的场景或一次性识别整段语音的场景如指令识别、语音转写、语音搜索、语音翻译等。技术上非流式语音识别注重语音的整体准确性和语义理解通常采用复杂的模型和算法来提高识别准确率。

Whisper是OpenAI开源的通用多语言语音识别模型库。Whisper使用了一个序列到序列的Transformer模型支持多国语言语音识别其英语的识别水平与人类接近。关于Whisper的安装和使用可参考Whisper开源仓库或参考文章Whisper语音转文字手把手教程。所提供的语音识别示例实现了简单的语音起始和结束检测并进行相应的语音识别和结果展示程序代码结构如下

.
├── asr.py 语音识别类
├── record.py 录音类
└── run.py 界面类

安装SpeechRecognition库和Whisper库后运行run.py文件即可打开示例应用。

界面类

界面类提供了一个基于PyQt5编写的简单应用界面如下所示。当界面初始化时会同时初始化录音类和语音识别类。点击开始录音按钮后程序将实现自动循环监听说话音频的开始和结束。每次说话结束后程序会自动进行语音识别并将识别结果显示在界面中。点击停止按钮则会等待录音结束并停止语音监听。

# run.py
from PyQt5 import QtGui
from PyQt5.QtWidgets import *
from PyQt5.QtCore import QSize, Qt
import sys
from record import AudioHandle


class Window(QMainWindow):
    """
    界面类
    """

    def __init__(self):
        super().__init__()
        # --- 设置标题
        self.setWindowTitle('语音识别demo')
        # --- 设置窗口尺寸
        # 获取系统桌面尺寸
        desktop = app.desktop()
        # 设置界面初始尺寸
        self.width = int(desktop.screenGeometry().width() * 0.3)
        self.height = int(0.5 * self.width)
        self.resize(self.width, self.height)
        # 设置窗口最小值
        self.minWidth = 300
        self.setMinimumSize(QSize(self.minWidth, int(0.5 * self.minWidth)))

        # --- 创建组件
        self.showBox = QTextEdit()
        self.showBox.setReadOnly(True)
        self.startBtn = QPushButton("开始录音")
        self.stopBtn = QPushButton("停止录音")
        self.stopBtn.setEnabled(False)

        # --- 组件初始化
        self.initUI()

        # --- 初始化音频类
        self.ahl = AudioHandle()
        # 连接用于传递信息的信号
        self.ahl.infoSignal.connect(self.showInfo)
        self.showInfo("<font color='blue'>{}</font>".format("程序已初始化"))

    def initUI(self) -> None:
        """
        界面初始化
        """
        # 设置整体布局
        mainLayout = QVBoxLayout()
        mainLayout.addWidget(self.showBox)
        # 设置底部水平布局
        blayout = QHBoxLayout()
        blayout.addWidget(self.startBtn)
        blayout.addWidget(self.stopBtn)
        mainLayout.addLayout(blayout)

        mainWidget = QWidget()
        mainWidget.setLayout(mainLayout)
        self.setCentralWidget(mainWidget)

        # 设置事件
        self.startBtn.clicked.connect(self.record)
        self.stopBtn.clicked.connect(self.record)

    def record(self) -> None:
        """
        录音控制
        """
        sender = self.sender()
        if sender.text() == "开始录音":
            self.stopBtn.setEnabled(True)
            self.startBtn.setEnabled(False)
            # 开启录音线程
            self.ahl.start()
        elif sender.text() == "停止录音":
            self.stopBtn.setEnabled(False)
            # waitDialog用于等待录音停止
            waitDialog = QProgressDialog("正在停止录音...", None, 0, 0)
            waitDialog.setWindowTitle("请等待")
            waitDialog.setWindowModality(Qt.ApplicationModal)
            waitDialog.setCancelButton(None)
            waitDialog.setRange(0, 0)

            # 设置 Marquee 模式
            waitDialog.setWindowFlag(Qt.WindowContextHelpButtonHint, False)
            waitDialog.setWindowFlag(Qt.WindowCloseButtonHint, False)
            waitDialog.setWindowFlag(Qt.WindowMaximizeButtonHint, False)
            waitDialog.setWindowFlag(Qt.WindowMinimizeButtonHint, False)
            waitDialog.setWindowFlag(Qt.WindowTitleHint, False)
            # 关闭对话框边框
            waitDialog.setWindowFlags(self.windowFlags() | Qt.FramelessWindowHint)

            # 连接关闭信号即ahl线程结束则waitDialog关闭
            self.ahl.finished.connect(waitDialog.accept)
            # 结束录音线程
            self.ahl.stop()
            if self.ahl.isRunning():
                # 显示对话框
                waitDialog.exec_()

            # 关闭对话框
            self.ahl.finished.disconnect(waitDialog.accept)
            waitDialog.close()

            self.startBtn.setEnabled(True)

    def showInfo(self, text: str) -> None:
        """
        信息展示函数
        :param text: 输入文字可支持html
        """
        self.showBox.append(text)
        if not self.ahl.running:
            self.stopBtn.click()

    def closeEvent(self, event: QtGui.QCloseEvent):
        """
        重写退出事件
        :param event: 事件对象
        """
        # 点击停止按钮
        if self.ahl.running:
            self.stopBtn.click()
        del self.ahl
        event.accept()


if __name__ == '__main__':
    app = QApplication(sys.argv)
    ex = Window()
    # 获取默认图标
    default_icon = app.style().standardIcon(QStyle.SP_MediaVolume)

    # 设置窗口图标为默认图标
    ex.setWindowIcon(default_icon)

    ex.show()
    sys.exit(app.exec_())

录音类

录音类可以用于监听麦克风输入的音频并调用语音识别类进行识别。通过设置采样率、适应环境时长、录音最长时长等参数实现自动判断说话开始和结束的功能。同时通过PyQt5的信号机制在界面上展示不同类型的信息包括警告信息和识别结果。

# record.py
import speech_recognition as sr
from PyQt5.QtCore import QThread, pyqtSignal
import time, os
import numpy as np
from asr import ASR


class AudioHandle(QThread):
    """
    录音控制类
    """

    # 用于展示信息的pyqt信号
    infoSignal = pyqtSignal(str)

    def __init__(self, sampleRate: int = 16000, adjustTime: int = 1, phraseLimitTime: int = 5,
                 saveAudio: bool = False, hotWord: str = ""):
        """
        :param sampleRate: 采样率
        :param adjustTime: 适应环境时长/s
        :param phraseLimitTime: 录音最长时长/s
        :param saveAudio: 是否保存音频
        :param hotWord: 热词数据
        """
        super(AudioHandle, self).__init__()
        self.sampleRate = sampleRate
        self.duration = adjustTime
        self.phraseTime = phraseLimitTime
        # 用于设置运行状态
        self.running = False
        self.rec = sr.Recognizer()
        # 麦克风对象
        self.mic = sr.Microphone(sample_rate=self.sampleRate)
        # 语音识别模型对象
        # hotWord为需要优先识别的热词
        # 输入"秦剑 无憾"表示优先匹配该字符串中的字符
        self.asr = ASR(prompt=hotWord)
        self.saveAudio = saveAudio
        self.savePath = "output"

    def run(self) -> None:
        self.listen()

    def stop(self) -> None:
        self.running = False

    def setInfo(self, text: str, type: str = "info") -> None:
        """
        展示信息
        :param text: 文本
        :param type: 文本类型
        """
        nowTime = time.strftime("%H:%M:%S", time.localtime())
        if type == "info":
            self.infoSignal.emit("<font color='blue'>{} {}</font>".format(nowTime, text))
        elif type == "text":
            self.infoSignal.emit("<font color='green'>{} {}</font>".format(nowTime, text))
        else:
            self.infoSignal.emit("<font color='red'>{} {}</font>".format(nowTime, text))

    def listen(self) -> None:
        """
        语音监听函数
        """
        try:
            with self.mic as source:
                self.setInfo("录音开始")
                self.running = True
                while self.running:
                    # 设备监控
                    audioIndex = self.mic.audio.get_default_input_device_info()['index']
                    workAudio = self.mic.list_working_microphones()
                    if len(workAudio) == 0 or audioIndex not in workAudio:
                        self.setInfo("未检测到有效音频输入设备", type='warning')
                        break
                    self.rec.adjust_for_ambient_noise(source, duration=self.duration)
                    self.setInfo("正在录音")
                    # self.running为否无法立即退出该函数如果想立即退出则需要重写该函数
                    audio = self.rec.listen(source, phrase_time_limit=self.phraseTime)
                    # 将音频二进制数据转换为numpy类型
                    audionp = self.bytes2np(audio.frame_data)
                    if self.saveAudio:
                        self.saveWav(audio)
                    # 判断音频rms值是否超过经验阈值如果没超过表明为环境噪声
                    if np.sqrt(np.mean(audionp ** 2)) < 0.02:
                        continue
                    self.setInfo("音频正在识别")
                    # 识别语音
                    result = self.asr.predict(audionp)
                    self.setInfo(f"识别结果为{result}", "text")
        except Exception as e:
            self.setInfo(e, "warning")
        finally:
            self.setInfo("录音停止")
            self.running = False

    def bytes2np(self, inp: bytes, sampleWidth: int = 2) -> np.ndarray:
        """
        将音频二进制数据转换为numpy类型
        :param inp: 输入音频二进制流
        :param sampleWidth: 音频采样宽度
        :return: 音频numpy数组
        """

        # 使用np.frombuffer函数将字节序列转换为numpy数组
        tmp = np.frombuffer(inp, dtype=np.int16 if sampleWidth == 2 else np.int8)
        # 确保tmp为numpy数组
        tmp = np.asarray(tmp)

        # 获取tmp数组元素的数据类型信息
        i = np.iinfo(tmp.dtype)
        # 计算tmp元素的绝对最大值
        absmax = 2 ** (i.bits - 1)
        # 计算tmp元素的偏移量
        offset = i.min + absmax

        # 将tmp数组元素转换为浮点型并进行归一化
        array = np.frombuffer((tmp.astype(np.float32) - offset) / absmax, dtype=np.float32)

        # 返回转换后的numpy数组
        return array

    def saveWav(self, audio: sr.AudioData) -> None:
        """
        保存语音结果
        :param audio: AudioData音频对象
        """
        nowTime = time.strftime("%H_%M_%S", time.localtime())
        os.makedirs(self.savePath, exist_ok=True)
        with open("{}/{}.wav".format(self.savePath, nowTime), 'wb') as f:
            f.write(audio.get_wav_data())

语音识别类

语音识别类利用Whisper进行语音识别。在使用Whisper进行语音识别时可以通过设置initial_prompt参数来指定初始提示。initial_prompt参数是一个字符串用于在模型生成文本之前提供一些初始的上下文信息。将这些信息传递给Whisper模型可以帮助它更好地理解任务的背景和上下文。通过设置适当的initial_prompt可以引导模型产生与特定主题相关的响应或者在对话中提供一些先验知识。例如热点词汇识别结果为简体字还是繁体字。initial_prompt并不是必需的参数如果没有适当的初始提示可以选择不使用它让模型完全自由生成响应。但是要注意的是如果输入的语音为环境噪声或者使用的是小型Whisper模型initial_prompt的设置可能会导致语音识别输出结果为initial_prompt。

Whisper提供了5种型号的模型其中4种支持纯英文版本以平衡速度和准确性。Whisper模型越大精度越高速度越慢本文默认使用small型号的模型。以下是这些可用模型的型号名称、大致的显存要求和相对速度

型号	参数量	仅英文模型	多语言模型	所需显存	相对速度
tiny	39M	tiny.en	tiny	~1GB	~32x
base	74M	base.en	base	~1GB	~16x
small	244M	small.en	small	~2GB	~6x
medium	769M	medium.en	medium	~5GB	~2x
large	1550M	N/A	large	~10GB	1x

# asr.py
import whisper
import numpy as np


class ASR:
    """
    语音识别模型类
    """

    def __init__(self, modelType: str = "small", prompt: str = ""):
        """
        :param modelType: whisper模型类型
        :param prompt: 提示词
        """
        # 模型默认使用cuda运行没gpu跑模型很慢。
        # 使用device="cpu"即可改为cpu运行
        self.model = whisper.load_model(modelType, device="cuda")
        # prompt作用就是提示模型输出指定类型的文字
        # 这里使用简体中文就是告诉模型尽可能输出简体中文的识别结果
        self.prompt = "简体中文" + prompt

    def predict(self, audio: np.ndarray) -> str:
        """
        语音识别
        :param audio: 输入的numpy音频数组
        :return: 输出识别的字符串结果
        """

        # prompt在whisper中用法是作为transformer模型交叉注意力模块的初始值。transformer为自回归模型会逐个生成识别文字
        # 如果输入的语音为空initial_prompt的设置可能会导致语音识别输出结果为initial_prompt
        result = self.model.transcribe(audio.astype(np.float32), initial_prompt=self.prompt)
        return result["text"]

4 参考

• 数字音频基础­­­­­­­­­­从PCM说起
• portaudio
• PyAudio
• pyaudio声卡信息中hostApi
• speech_recognition
• Whisper
• Whisper语音转文字手把手教程