Python使用PyAudio实现音频录制与播放功能详解

远山来客

发布时间：2026-10-15 | 3人浏览过

原创

本文介绍如何利用Python的PyAudio库结合wave模块实现音频录制与播放功能。文章详细讲解了环境配置、回调函数处理、流式录制原理，并封装了AudioRecord类以简化操作。同时介绍了使用pynput库进行键盘监听以实现录音中断，最后展示了基于PyQt5构建可视化界面的思路与部分代码实现。

近期在使用屏幕录制软件时产生了一个想法：能否利用Python开发一个屏幕录制工具，以此锻炼编程实践能力。为此，计划撰写一系列教程，涵盖以下核心模块：

屏幕视频录制实现

音频录制功能开发

音视频合成处理

基于PyQt5的可视化界面构建

上述四个部分构成了完整的项目框架，期望能尽快完善。在前一篇文章中，已利用OpenCV完成了屏幕录制部分，本文将延续该系列，重点讲解如何使用Python进行音频录制。

应用平台

Windows 10
Python 3.7

音频录制部分

音频录制在原理上与视频录制类似，同样以数据帧的形式进行录制和保存。本部分将使用第三方库PyAudio配合内置的wave模块来编写核心代码：

pip install PyAudio

若遇到安装失败的情况，可点击此处下载对应的.whl文件。注意，cp37代表Python 3.7环境，64代表64位操作系统。若下载的版本不匹配，安装将会失败。下载完成后，在CMD窗口中进入.whl文件所在目录，执行pip install PyAudio-xx.whl即可完成安装。

音频录制的核心代码逻辑如下：


from pyaudio import PyAudio, paInt16, paContinue, paComplete

# 设置固定参数
chunk = 1024  # 每个缓冲区的帧数
format_sample = paInt16  # 采样位数
channels = 2  # 声道： 1，单声道；2，双声道
fps = 44100  # 采样频率
# 这里采用回调的方式录制音频
def callback(in_data, frame_count, time_info, status):
    """录制回调函数"""
    wf.writeframes(in_data)
    if xx:  # 当某某条件满足时
        return in_data, paContinue
    else:
        return in_data, paComplete
# 实例化PyAudio
p = PyAudio()
stream = p.open(format=format_sample,
				channels=channels,
				rate=fps,
                frames_per_buffer=chunk,
                input=True,
                input_device_index=None,  # 输入设备索引, None为默认设备
                stream_callback=callback   # 回调函数
                )
# 开始流录制
stream.start_stream()
# 判断流是否活跃
while stream.is_active():
	time.sleep(0.1)    # 0.1为灵敏度
# 录制完成，关闭流及实例
stream.stop_stream()
stream.close()
p.terminate()

采用流式处理并结合回调函数进行录制。首先需要定义保存音频文件的逻辑，使用wave模块新建一个音频二进制文件：

import wave
wf = wave.open('test.wav', 'wb')
wf.setnchannels(channels)
wf.setsampwidth(p.get_sample_size(format_sample))
wf.setframerate(fps)

为了便于后续代码的复用与集成，将上述逻辑封装为一个类。

from pyaudio import PyAudio
class AudioRecord(PyAudio):
    def __init__(self,):

完整源码详见文末补充。

音频播放部分

播放部分的代码结构与录制部分差异不大，核心实现如下：

wf = wave.open('test.wav', 'rb')
def callback(in_data, frame_count, time_info, status):
	data = wf.readframes(frame_count)
	return data, paContinue
stream = p.open(format=p.get_format_from_width(wf.getsampwidth()),
				channels=wf.getnchannels(),
                rate=wf.getframerate(),
				output=True,
				output_device_index=output_device_index,  # 输入设备索引
				stream_callback=callback  # 输出用回调函数
                )
stream.start_stream()
while stream.is_active():
	time.sleep(0.1)

目前测试确认.wav和.mp3格式可以正常录制及播放，其他类型的音频格式用户可以自行调用代码进行测试验证。

GUI窗口所需属性值代码部分

考虑到GUI界面需要更人性化地输出和输入数据，编写了相关代码，内容包含获取音频时长以及查询系统可用的输入设备和输出设备。

# 音频时长
duration = wf.getnframes() / wf.getframerate()

# 获取系统目前已安装的输入输出设备
dev_info = self.get_device_info_by_index(i)
default_rate = int(dev_info['defaultSampleRate'])
if not dev_info['hostApi'] and default_rate == fps and '映射器' not in dev_info['name']:
	if dev_info['maxInputChannels']:
		print('输入设备：', dev_info['name'])
	elif dev_info['maxOutputChannels']:
		print('输出设备：', dev_info['name'])

pynput监听键盘

在此部分代码中，暂时使用pynput库监听键盘事件，以实现对录音过程的暂停或终止处理。可以调用前文提到的键盘监听代码。

def hotkey(self):
    """热键监听"""
    with keyboard.Listener(on_press=self.on_press) as listener:
        listener.join()

def on_press(self, key):
    try:
        if key.char == 't':  # t键，录制结束，保存音频
            self.flag = True
        elif key.char == 'k':  # k键，录制中止，删除文件
            self.flag = True
            self.kill = True
    except Exception as e:
        print(e)

该功能与上一篇文章中的实现类似，此处不再赘述。

总结

以上介绍了如何使用PyAudio调用Windows音频设备进行录制及播放。通过本项目，学习了类的定义及其继承关系等知识。目前的用法仅展示了PyAudio功能的冰山一角，还有更多高级特性值得进一步探索！

撰写日期：2021年12月20日

源码:

import wave
import time
from pathlib import Path
from threading import Thread
from pyaudio import PyAudio, paInt16, paContinue, paComplete
from pynput import keyboard  # pip install pynput

class AudioRecord(PyAudio):
    def __init__(self, channels=2):
        super().__init__()
        self.chunk = 1024  # 每个缓冲区的帧数
        self.format_sample = paInt16  # 采样位数
        self.channels = channels  # 声道： 1，单声道；2，双声道
        self.fps = 44100  # 采样频率
        self.input_dict = None
        self.output_dict = None
        self.stream = None
        self.filename = '~test.wav'
        self.duration = 0   # 音频时长
        self.flag = False
        self.kill = False
    def __call__(self, filename):
        """重载文件名"""
        self.filename = filename
    def callback_input(self, in_data, frame_count, time_info, status):
        """录制回调函数"""
        self.wf.writeframes(in_data)
        if not self.flag:
            return in_data, paContinue
        else:
            return in_data, paComplete
    def callback_output(self, in_data, frame_count, time_info, status):
        """播放回调函数"""
        data = self.wf.readframes(frame_count)
        return data, paContinue
    def open_stream(self, name):
        """打开录制流"""
        input_device_index = self.get_device_index(name, True) if name else None
        return self.open(format=self.format_sample,
                         channels=self.channels,
                         rate=self.fps,
                         frames_per_buffer=self.chunk,
                         input=True,
                         input_device_index=input_device_index,  # 输入设备索引
                         stream_callback=self.callback_input
                         )
    def audio_record_run(self, name=None):
        """音频录制"""
        self.wf = self.save_audio_file(self.filename)
        self.stream = self.open_stream(name)
        self.stream.start_stream()
        while self.stream.is_active():
            time.sleep(0.1)
        self.wf.close()
        if self.kill:
            Path(self.filename).unlink()
        self.duration = self.get_duration(self.wf)
        print(self.duration)
        self.terminate_run()
    def run(self, filename=None, name=None, record=True):
        """音频录制线程"""
        thread_1 = Thread(target=self.hotkey, daemon=True)
        if record:
            # 录制
            if filename:
                self.filename = filename
            thread_2 = Thread(target=self.audio_record_run, args=(name,))
            # 播放
            if not filename:
                raise Exception('未输入音频文件名，不能播放，请输入后再试！')
            thread_2 = Thread(target=self.read_audio, args=(filename, name,))
        thread_1.start()
        thread_2.start()
    def read_audio(self, filename, name=None):
        """音频播放"""
        output_device_index = self.get_device_index(name, False) if name else None
        with wave.open(filename, 'rb') as self.wf:
            self.duration = self.get_duration(self.wf)
            self.stream = self.open(format=self.get_format_from_width(self.wf.getsampwidth()),
                                    channels=self.wf.getnchannels(),
                                    rate=self.wf.getframerate(),
                                    output=True,
                                    output_device_index=output_device_index,  # 输出设备索引
                                    stream_callback=self.callback_output
                                    )
            self.stream.start_stream()
            while self.stream.is_active():
                time.sleep(0.1)
    @staticmethod
    def get_duration(wf):
        """获取音频时长"""
        return round(wf.getnframes() / wf.getframerate(), 2)
    def get_in_out_devices(self):
        """获取系统输入输出设备"""
        self.input_dict = {}
        self.output_dict = {}
        for i in range(self.get_device_count()):
            dev_info = self.get_device_info_by_index(i)
            default_rate = int(dev_info['defaultSampleRate'])
            if not dev_info['hostApi'] and default_rate == self.fps and '映射器' not in dev_info['name']:
                if dev_info['maxInputChannels']:
                    self.input_dict[dev_info['name']] = i
                elif dev_info['maxOutputChannels']:
                    self.output_dict[dev_info['name']] = i
    def get_device_index(self, name, input_in=True):
        """获取选定设备索引"""
        if input_in and self.input_dict:
            return self.input_dict.get(name, -1)
        elif not input_in and self.output_dict:
            return self.output_dict.get(name, -1)
    def save_audio_file(self, filename):
        """音频文件保存"""
        wf = wave.open(filename, 'wb')
        wf.setnchannels(self.channels)
        wf.setsampwidth(self.get_sample_size(self.format_sample))
        wf.setframerate(self.fps)
        return wf
    def terminate_run(self):
        """结束流录制或流播放"""
        if self.stream:
            self.stream.stop_stream()
            self.stream.close()
        self.terminate()
    def hotkey(self):
        """热键监听"""
        with keyboard.Listener(on_press=self.on_press) as listener:
            listener.join()
    def on_press(self, key):
        try:
            if key.char == 't':  # t键，录制结束，保存音频
                self.flag = True
            elif key.char == 'k':  # k键，录制中止，删除文件
                self.kill = True
        except Exception as e:
            print(e)
if __name__ == '__main__':
    audio_record = AudioRecord()
    audio_record.get_in_out_devices()
    # 录制
    print(audio_record.input_dict)
    audio_record.run('test.mp3')
    # 播放
    print(audio_record.output_dict)
    audio_record.run('test.mp3', record=False)

至此，关于Python使用PyAudio制作录音工具的文章介绍完毕。更多相关Python PyAudio录音工具内容，请搜索本站历史文章或继续浏览后续相关文章，希望大家以后多多支持本站！