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

语音识别技术的快速发展为实现更多智能化应用提供了无限可能。本文旨在介绍一个基于Python实现的简易音频录制与语音识别应用。文章简要介绍相关技术的应用，重点放在音频录制方面，而语音识别则关注于调用相关的语音识别库。本文将首先概述一些音频基础概念，然后详细讲解如何利用PyAudio库和SpeechRecognition库实现音频录制功能。最后，构建一个简单的语音识别示例应用，该应用程序可以实时监听音频的开始和结束，并将录制的音频数据传输至Whisper语音识别库进行语音识别，最终将识别结果输出到基于PyQt5搭建的简易页面中。

本文所有代码见：Python-Study-Notes

文章目录

• 0 音频基础概念
• 1 PyAudio
• • 1.1 PyAudio介绍与安装
  • 1.2 音频录制与播放
  • • 1.2.1 音频播放
    • 1.2.2 音频录制
    • 1.2.3 全双工音频录制与播放
  • 1.3 回调函数的使用
  • 1.4 设备管理
• 2 SpeechRecognition
• • 2.1 SpeechRecognition介绍与安装
  • 2.2 示例代码
• 3 语音识别示例应用
• 4 参考

0 音频基础概念

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

采样频率

音频信号通常是连续的模拟波形，为了存储它们，需要将其离散化。这通过采样来实现，即在固定时间间隔内测量声音信号的幅度。采样的过程就是抽取模拟信号各点的频率值。采样率越高即1秒内抽取的数据点越多，音频音质就越好，但同时也增加了存储和处理成本。奈奎斯特-香农（Nyquist–Shannon）采样定理强调采样频率必须高于信号最大频率的两倍，以确保从采样值中完全恢复原始模拟信号。在音频信号采样领域，常使用两个主要的采样频率：16kHz和44.1kHz。
举例来说，16kHz表示每秒采样16000次，而人类言语声音频率范围在200Hz到8kHz，16kHz的采样频率已足够捕捉人类语音频率特征，同时减轻了音频数据存储和处理的负担。因此常用语音采样频率为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/WAEV（Waveform Audio File Format），FLAC（Free Lossless Audio Codec），AIFF（Audio Interchange File Format）等。有损压缩常见的音频编码格式有：MP3（MPEG Audio Layer III），AAC（Advanced Audio Coding），WMA（Windows Media Audio）等。

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

音频与视频概念对比

下面的代码展示了读取音频内容为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~32678，soundfile库会除以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型号的模型。以下是这些可用模型的型号名称、大致的显存要求和相对速度：

# 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语音转文字手把手教程