【说站】Python+pyaudio实现音频控制示例详解

pip install PyAudio

pip3 install PyAudio-0.2.11-cp38-cp38-win_amd64.whl

/*
* 提示：该行代码过长，系统自动注释不进行高亮。一键复制会移除系统注释 
* #!/usr/bin/env python3#-*- coding:utf-8 -*-#------------- 音频设备操作模块 -------------------##   功能:   录制/获取音频流/播放音频#   时间：  2021-09-13##-------------------------------------------------- import sys ,pyaudio, wavefrom tqdm import tqdm class UacAudioInAndOut:def __init__(self):"""功能:   录音参数初始化创建vad检测模块对象参数:   /返回值: /"""self.input_format_dict = {"S8_LE":16, "S16_LE":8, "S24_LE":4, "S32_LE":2}self.framerate_list = [8000, 11025, 16000, 22050, 32000, 44100, 48000,88200, 96000, 176400, 192000, 352800, 384000] def _inforPrintf(self, infor_content):"""功能:   检测操作系统，使用正确编码输出打印信息参数:   infor_content: 信息内容返回值: /"""if sys.platform != "linux" and sys.platform != "darwin":infor_content = str(infor_content).encode("gbk","ignore").decode("gbk")print(infor_content) def GetAllDevInfor(self):"""功能:   显示支持设备信息参数:   /返回值: /"""PA = pyaudio.PyAudio()self._inforPrintf("----------------------< 本机支持设备 >------------------------------")for dev_index in range(PA.get_device_count()):self._inforPrintf("\n-------------------------------------------------------")for key in PA.get_device_info_by_index(dev_index):self._inforPrintf("%s:%s"%(key, str(PA.get_device_info_by_index(dev_index)[key])))self._inforPrintf("========================================================")  def GetUacDevInfor(self, devKeywordOrIndex=None):"""功能:   获取UAC设备信息参数:   devKeywordOrIndex: 设备名称关键字或索引返回值: dic 设备信息字典False 设备信息获取失败"""PA = pyaudio.PyAudio()if devKeywordOrIndex == None:self._inforPrintf("\033[0;36;31m[UacAudioInAndOut] 未设设备, 当前使用默认设备\033[0m")return PA.get_default_input_device_info()if str(devKeywordOrIndex).isdigit():devKeywordOrIndex = int(devKeywordOrIndex)return PA.get_device_info_by_index(devKeywordOrIndex) uac_infor_list = []for uac_index in range(PA.get_device_count()):if PA.get_device_info_by_index(uac_index).get("name").find(str(devKeywordOrIndex)) >= 0:uac_infor_list.append(PA.get_device_info_by_index(uac_index)) if len(uac_infor_list) > 1:self._inforPrintf("\033[0;36;33m[UacAudioInAndOut] UAC 设备有多个，\请修正关键字, 当前设备如下: %s\033[0m"%str(uac_infor_list))return Falseelse:return uac_infor_list.pop() def is_framerate_supported(self, setFramerate, UacAudioInHandle,load_parame_dict, input_or_output="input"):"""功能:   判断当配置在指定设备中是否支持参数:   setFramerate:   设置采样率UacAudioInHandle:   设备句柄load_parame_dict:   加载字典input_or_output:输入/输出功能返回值: bool True/False"""try:if input_or_output == "input":UacAudioInHandle.is_format_supported(rate=float(setFramerate),input_device=load_parame_dict['index'],input_channels=load_parame_dict['setInputChannels'],input_format=load_parame_dict['_setInputFormat'])else:UacAudioInHandle.is_format_supported(rate=float(setFramerate),output_device=load_parame_dict['index'],output_channels=load_parame_dict['maxOutputChannels'],output_format=UacAudioInHandle.get_format_from_width(load_parame_dict['setOutputFormat']))return Trueexcept:return False def LoadUacAudioInDevice(self, maxStreamDuration=1000, setInputChannels=None,setInputFormat=None, devKeywordOrIndex=None):"""功能:   加载音频获取设备参数:   maxStreamDuration=1000 默认一段流时长setInputChannels:   通道数setInputFormat: 位宽devKeywordOrIndex:录音设备关键字/索引返回值:成功: UacAudioInHandle, StreamHandle, load_parame_dict失败: False"""try:load_parame_dict = {}uac_infor_dict = self.GetUacDevInfor(devKeywordOrIndex)if not setInputFormat:_Format = "S16_LE"self._inforPrintf("\033[0;36;33m[UacAudioInAndOut] 未设置位宽，使用默认 S16_LE \033[0m")else:_Format = setInputFormatsetInputFormat = self.input_format_dict[_Format] if not setInputChannels or int(setInputChannels) > uac_infor_dict["maxInputChannels"]:setInputChannels = uac_infor_dict["maxInputChannels"]self._inforPrintf("\033[0;36;33m[UacAudioInAndOut] 输入通道未设置/超出当前设备最大值,使用默认最大通道 %s\\033[0m"%setInputChannels)else:setInputChannels = int(setInputChannels)dev_index = uac_infor_dict["index"]load_parame_dict["index"]=dev_indexload_parame_dict["setInputFormat"] = _Formatload_parame_dict["_setInputFormat"] = setInputFormatload_parame_dict["setInputChannels"] = setInputChannelsUacAudioInHandle = pyaudio.PyAudio()for setInputFramerate in self.framerate_list:if self.is_framerate_supported(setInputFramerate, UacAudioInHandle, load_parame_dict):load_parame_dict["setInputFramerate"] = setInputFrameratebreak#计算数据大小一段CHUNK_SIZE = int(setInputFramerate * maxStreamDuration / 1000)load_parame_dict["CHUNK_SIZE"] = CHUNK_SIZEself._inforPrintf("\033[0;36;38m[UacAudioInAndOut] 加载参数: %s\033[0m"%str(load_parame_dict))#加载设备StreamHandle = UacAudioInHandle.open(format=load_parame_dict['_setInputFormat'],channels=load_parame_dict['setInputChannels'],rate=load_parame_dict['setInputFramerate'],input=True,input_device_index=load_parame_dict['index'],start=False,frames_per_buffer=int(CHUNK_SIZE))#开始流获取StreamHandle.start_stream()return UacAudioInHandle, StreamHandle, load_parame_dictexcept:self._inforPrintf("\033[0;36;31m[UacAudioInAndOut] Uac AudioIn 加载失败\033[0m")return False, False, False def LoadUacAudioOutDevice(self, devKeywordOrIndex):"""功能:   加载音频输出设备参数:   /返回值: UacAudioInHandle 或 False"""try:uac_infor_dict = self.GetUacDevInfor(devKeywordOrIndex)UacAudioInHandle = pyaudio.PyAudio()return UacAudioInHandle, uac_infor_dictexcept:return False  def GetUacAudioInStream(self, StreamHandle, CHUNK_SIZE):"""功能:   开始采集声卡音频生成音频流参数:   UacAudioInHandle:   设备句柄StreamHandle:   流句柄返回值  chunk_data 流数据"""return StreamHandle.read(CHUNK_SIZE, exception_on_overflow=False) #防止溢出  def UacAudioOutPlay(self, playWavFile, Repeat=None, Pdict=None, devKeywordOrIndex=None,):"""功能:   可以循环播放指定文件参数:   playWavFile:播放文件路径Repeat: 循环播放次数CustomizeAudioParam:自定义播放参数返回值: /"""UacAudioInHandle, uac_infor_dict = self.LoadUacAudioOutDevice(devKeywordOrIndex)self._inforPrintf(str(uac_infor_dict).encode("gbk","ignore").decode("gbk"))self._inforPrintf("\033[1;36;34m[UacAudioInAndOut] 指定设备: %s\t播放文件: %s\t循环总数: %s\\033[0m"%(devKeywordOrIndex, playWavFile,Repeat))try:chunk=1024pfb = wave.open(playWavFile, 'rb')setOutputFormat = pfb.getsampwidth()setOutputChannels = pfb.getnchannels()setOutputFramerate = pfb.getframerate()uac_infor_dict['setOutputFormat'] = setOutputFormat if setOutputChannels > uac_infor_dict["maxOutputChannels"]:self._inforPrintf("\033[0;36;31m[UacAudioInAndOut] 当前通道数，在该设备上不支持, \设备最大通道数: %s\033[0m"%uac_infor_dict["maxOutputChannels"])return Falseif not self.is_framerate_supported(setOutputFramerate, UacAudioInHandle, uac_infor_dict, "output"):self._inforPrintf("\033[0;36;31m[UacAudioInAndOut] 当前文件采样率，在该设备上不支持,\设备默认采样率: %s\033[0m"%uac_infor_dict["defaultSampleRate"])return Falseelse:uac_infor_dict["defaultSampleRate"] = setOutputFrameratestream = UacAudioInHandle.open(output_device_index=uac_infor_dict['index'],format=UacAudioInHandle.get_format_from_width(setOutputFormat),channels=setOutputChannels,rate=setOutputFramerate,output=True) if Repeat == "Dead_cycle":self._inforPrintf("\033[1;36;33m[UacAudioInAndOut] Dead cycle play !!! \033[0m")while True:if type(Pdict) == dict and Pdict["play status"] == "stop":breakpfb = wave.open(playWavFile, 'rb')while True:data = pfb.readframes(chunk)if not data:breakstream.write(data)else:for index in tqdm(range(int(Repeat))):if type(Pdict) == dict and Pdict["play status"] == "stop":breakpfb = wave.open(playWavFile, 'rb')while True:data = pfb.readframes(chunk)if not data:breakstream.write(data) stream.stop_stream()stream.close()self.CloseAudioDevice(UacAudioInHandle)return Trueexcept:stream.stop_stream()stream.close()return False  def UacAudioInRecord(self, saveWavFile, recordTime, #单位秒setInputChannels=None,setInputFormat=None,devKeywordOrIndex=None):"""功能:   录制音频文件参数:   recordTime: 录音时长, 单位(s)setInputFramerate:  采样率setInputChannels:   通道数setInputFormat: 位宽devKeywordOrIndex:  录音设备索引返回值: /"""maxStreamDuration=1000load_parame_dict = {}UacAudioInHandle, StreamHandle, load_parame_dict = self.LoadUacAudioInDevice(maxStreamDuration,setInputChannels,setInputFormat,devKeywordOrIndex)if not UacAudioInHandle or not StreamHandle:self._inforPrintf("\033[0;36;31m[UacAudioInAndOut] 录音失败\033[0m")return False self._inforPrintf("\033[1;36;34m[UacAudioInAndOut] 录音 -> 文件名: %s 时长: %s\\033[0m"%(saveWavFile,recordTime))self._inforPrintf(load_parame_dict["CHUNK_SIZE"])data_list = []for recordTime_index in range(int(recordTime)):data = Nonedata = StreamHandle.read(load_parame_dict["CHUNK_SIZE"], exception_on_overflow=False)data_list.append(data)StreamHandle.stop_stream()StreamHandle.close()self.CloseAudioDevice(UacAudioInHandle)with wave.open(saveWavFile, "wb") as wavfb:wavfb.setnchannels(load_parame_dict["setInputChannels"])wavfb.setsampwidth(UacAudioInHandle.get_sample_size(load_parame_dict["_setInputFormat"]))wavfb.setframerate(load_parame_dict["setInputFramerate"])wavfb.writeframes(b''.join(data_list)) """功能:   关闭音频流设备参数:   UacAudioInHandle返回值: bool True/False"""try:StreamHandle.stop_stream()StreamHandle.close()self.CloseAudioDevice()return Trueexcept:return False def CloseAudioDevice(self, UacAudioDeviceHandle):"""功能:   释放 Audio 设备参数:   UacAudioDeviceHandle返回值: bool True/False"""try:UacAudioDeviceHandle.terminate()return Trueexcept:return False  if __name__=="__main__":asv = UacAudioInAndOut()asv.GetAllDevInfor()#asv.UacAudioOutPlay(sys.argv[1], int(sys.argv[2]), None, sys.argv[3])asv.UacAudioInRecord(sys.argv[1], sys.argv[2])
*/