Python+pyaudio實現音頻控制示例詳解
簡介
PyAudio是一個跨平臺的音頻處理工具包,使用該工具包可以在Python程序中播放和錄制音頻,也可以產生wav文件等
安裝
pip install PyAudio
註意:使用該命令安裝時可能會報錯,報錯內容如下:
針對該問題,我們使用whl文件進行安裝,首先在網址下面找到以下文件並下載,根據自己的python版本及計算機系統下載相應文件即可。
下載完成後,切換到文件所在目錄,使用如下命令安裝即可
pip3 install PyAudio-0.2.11-cp38-cp38-win_amd64.whl
pyaudio控制指定設備,錄制音頻/采集音頻流/播放音頻
#!/usr/bin/env python3 #-*- coding:utf-8 -*- #------------- 音頻設備操作模塊 ------------------- # # 功能: 錄制/獲取音頻流/播放音頻 # 時間: 2021-09-13 # #-------------------------------------------------- import sys ,pyaudio, wave from 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 False else: 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 True except: 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 = setInputFormat setInputFormat = 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_index load_parame_dict["setInputFormat"] = _Format load_parame_dict["_setInputFormat"] = setInputFormat load_parame_dict["setInputChannels"] = setInputChannels UacAudioInHandle = pyaudio.PyAudio() for setInputFramerate in self.framerate_list: if self.is_framerate_supported(setInputFramerate, UacAudioInHandle, load_parame_dict): load_parame_dict["setInputFramerate"] = setInputFramerate break #計算數據大小一段 CHUNK_SIZE = int(setInputFramerate * maxStreamDuration / 1000) load_parame_dict["CHUNK_SIZE"] = CHUNK_SIZE self._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_dict except: 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_dict except: 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=1024 pfb = 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 False if 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 False else: uac_infor_dict["defaultSampleRate"] = setOutputFramerate stream = 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": break pfb = wave.open(playWavFile, 'rb') while True: data = pfb.readframes(chunk) if not data: break stream.write(data) else: for index in tqdm(range(int(Repeat))): if type(Pdict) == dict and Pdict["play status"] == "stop": break pfb = wave.open(playWavFile, 'rb') while True: data = pfb.readframes(chunk) if not data: break stream.write(data) stream.stop_stream() stream.close() self.CloseAudioDevice(UacAudioInHandle) return True except: 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=1000 load_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 = None data = 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 True except: return False def CloseAudioDevice(self, UacAudioDeviceHandle): """ 功能: 釋放 Audio 設備 參數: UacAudioDeviceHandle 返回值: bool True/False """ try: UacAudioDeviceHandle.terminate() return True except: 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])
以上就是Python+pyaudio實現音頻控制示例詳解的詳細內容,更多關於Python pyaudio音頻控制的資料請關註WalkonNet其它相關文章!
推薦閱讀:
- python 錄制系統聲音的示例
- PyTorch中的CUDA的操作方法
- 基於Pytorch實現的聲音分類實例代碼
- 如何使用 Python為你的在線會議創建一個假的攝像頭
- Python幹貨實戰之八音符醬小遊戲全過程詳解