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, 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("33[0;36;31m[UacAudioInAndOut] 未設裝置, 當前使用預設裝置33[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("33[0;36;33m[UacAudioInAndOut] UAC 裝置有多個，
                    請修正關鍵字, 當前裝置如下: %s33[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("33[0;36;33m[UacAudioInAndOut] 未設定位寬，使用預設 S16_LE 33[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("33[0;36;33m[UacAudioInAndOut] 輸入通道未設定/超出當前裝置最大值,使用預設最大通道 %s
                                                                                    33[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("33[0;36;38m[UacAudioInAndOut] 載入引數: %s33[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("33[0;36;31m[UacAudioInAndOut] Uac AudioIn 載入失敗33[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("33[1;36;34m[UacAudioInAndOut] 指定裝置: %st播放檔案: %st迴圈總數: %s
                                            33[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("33[0;36;31m[UacAudioInAndOut] 當前通道數，在該裝置上不支援, 
                                裝置最大通道數: %s33[0m"%uac_infor_dict["maxOutputChannels"])
                return False
            if not self.is_framerate_supported(setOutputFramerate, UacAudioInHandle, uac_infor_dict, "output"):
                self._inforPrintf("33[0;36;31m[UacAudioInAndOut] 當前檔案取樣率，在該裝置上不支援,
                                    裝置預設取樣率: %s33[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("33[1;36;33m[UacAudioInAndOut] Dead cycle play !!! 33[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("33[0;36;31m[UacAudioInAndOut] 錄音失敗33[0m")
            return False

        self._inforPrintf("33[1;36;34m[UacAudioInAndOut] 錄音 -> 檔名: %s 時長: %s
                                            33[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])