調(diào)用科大訊飛語音聽寫，使用 Python 實(shí)現(xiàn)語音識別，將實(shí)時(shí)語音轉(zhuǎn)換為文字。

參考這篇[博客]()實(shí)現(xiàn)的錄音，首先在官網(wǎng)下載了關(guān)于語音聽寫的 SDK ，然后在文件夾內(nèi)新建了兩個(gè) .py 文件，分別是 get_audio.py 和 iat_demo.py ，并且新建了一個(gè)存放錄音的文件夾 audios ，文件夾內(nèi)存放錄音文件 input.wav ，我的整個(gè)文件目錄如下：

          
            asr_SDK(文件名)
├─ Readme.html
├─ audios
│    └─ input.wav（存放音頻）
├─ bin
│    ├─ gm_continuous_digit.abnf
│    ├─ ise_cn
│    ├─ ise_en
│    ├─ msc
│    ├─ msc.dll （因?yàn)槲沂?2位的python，所以用的這個(gè)動態(tài)鏈接庫）
│    ├─ msc_x64.dll
│    ├─ source.txt
│    ├─ userwords.txt
│    └─ wav
├─ doc
├─ get_audio.py
├─ iat_demo.py
├─ include
├─ libs
├─ release.txt
└─ samples
          
        

一、錄音

這里使用的是 pyaudio 進(jìn)行錄音，需要下載相關(guān)的輪子，具體可參考我的另一篇博客。然后根據(jù)自己的需要進(jìn)行了修改， gt_audio.py 全部代碼如下：

          
            
import pyaudio # 這個(gè)需要自己下載輪子
import wave
in_path = "./audios/input.wav" # 存放錄音的路徑

def get_audio(filepath):
    aa = str(input("是否開始錄音？   （y/n）"))
    if aa == str("y") :
        CHUNK = 1024
        FORMAT = pyaudio.paInt16
        CHANNELS = 1                # 聲道數(shù)
        RATE = 11025                # 采樣率
        RECORD_SECONDS = 5          # 錄音時(shí)間
        WAVE_OUTPUT_FILENAME = filepath
        p = pyaudio.PyAudio()

        stream = p.open(format=FORMAT,
                        channels=CHANNELS,
                        rate=RATE,
                        input=True,
                        frames_per_buffer=CHUNK)

        print("*"*5, "開始錄音：請?jiān)?秒內(nèi)輸入語音", "*"*5)
        frames = []
        for i in range(0, int(RATE / CHUNK * RECORD_SECONDS)):
            data = stream.read(CHUNK)
            frames.append(data)
        print("*"*5, "錄音結(jié)束\n")

        stream.stop_stream()
        stream.close()
        p.terminate()

        wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
        wf.setnchannels(CHANNELS)
        wf.setsampwidth(p.get_sample_size(FORMAT))
        wf.setframerate(RATE)
        wf.writeframes(b''.join(frames))
        wf.close()
    elif aa == str("否"):
        exit()
    else:
        print("語音錄入失敗，請重新開始")
        get_audio(in_path)

          
        

錄音的保持是可循環(huán)的，每當(dāng)重新錄音，都會覆蓋前一次的音頻。

二、語音識別

直接使用的是科大訊飛官網(wǎng)語音聽寫的 web API 關(guān)于 Python 的例子，在此基礎(chǔ)上進(jìn)行了相關(guān)的調(diào)整，自動識別錄音轉(zhuǎn)換為文字。 iat_demo.py 的全部代碼如下：

          
            import websocket
import requests
import datetime
import hashlib
import base64
import hmac
import json
import os, sys
import re
from urllib.parse import urlencode
import logging
import time
import ssl
import wave
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
from pyaudio import PyAudio,paInt16
from get_audio import get_audio # 導(dǎo)入錄音.py文件

input_filename = "input.wav"               # 麥克風(fēng)采集的語音輸入
input_filepath = "./audios/"              # 輸入文件的path
in_path = input_filepath + input_filename



type = sys.getfilesystemencoding()

path_pwd = os.path.split(os.path.realpath(__file__))[0]
os.chdir(path_pwd)

try:
    import thread
except ImportError:
    import _thread as thread

logging.basicConfig()

STATUS_FIRST_FRAME = 0  # 第一幀的標(biāo)識
STATUS_CONTINUE_FRAME = 1  # 中間幀標(biāo)識
STATUS_LAST_FRAME = 2  # 最后一幀的標(biāo)識

framerate = 8000
NUM_SAMPLES = 2000
channels = 1
sampwidth = 2
TIME = 2

global wsParam


class Ws_Param(object):
    # 初始化
    def __init__(self, host):
        self.Host = host
        self.HttpProto = "HTTP/1.1"
        self.HttpMethod = "GET"
        self.RequestUri = "/v2/iat"
        self.APPID = "5d312675" # 在控制臺-我的應(yīng)用-語音聽寫（流式版）獲取APPID
        self.Algorithm = "hmac-sha256"
        self.url = "wss://" + self.Host + self.RequestUri

        # 采集音頻 錄音
        get_audio("./audios/input.wav")

        # 設(shè)置測試音頻文件，流式聽寫一次最多支持60s，超過60s會引起超時(shí)等錯(cuò)誤。
        self.AudioFile = r"./audios/input.wav"

        self.CommonArgs = {"app_id": self.APPID}
        self.BusinessArgs = {"domain":"iat", "language": "zh_cn","accent":"mandarin"}

    def create_url(self):
        url = 'wss://ws-api.xfyun.cn/v2/iat'
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))
        APIKey = 'a6aabfcca4ae28f9b6a448f705b7e432' # 在控制臺-我的應(yīng)用-語音聽寫（流式版）獲取APIKey
        APISecret = 'e649956e14eeb085d1b0dce77a671131' # 在控制臺-我的應(yīng)用-語音聽寫（流式版）獲取APISecret

        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + "/v2/iat " + "HTTP/1.1"
        signature_sha = hmac.new(APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()
        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')

        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
            APIKey, "hmac-sha256", "host date request-line", signature_sha)
        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
        v = {
            "authorization": authorization,
            "date": date,
            "host": "ws-api.xfyun.cn"
        }
        url = url + '?' + urlencode(v)
        return url

# 收到websocket消息的處理 這里我對json解析進(jìn)行了一些更改 打印簡短的一些信息
def on_message(ws, message):
    msg = json.loads(message) # 將json對象轉(zhuǎn)換為python對象 json格式轉(zhuǎn)換為字典格式
    try:
        code = msg["code"]
        sid = msg["sid"]
        
        if code != 0:
            errMsg = msg["message"]
            print("sid:%s call error:%s code is:%s\n" % (sid, errMsg, code))
        else:
            result = msg["data"]["result"]["ws"]
            # 以json格式顯示 
            data_result = json.dumps(result, ensure_ascii=False, sort_keys=True, indent=4, separators=(',', ': ')) 
            print("sid:%s call success!" % (sid))
            print("result is:%s\n" % (data_result))
    except Exception as e:
        print("receive msg,but parse exception:", e)


# 收到websocket錯(cuò)誤的處理
def on_error(ws, error):
    print("### error:", error)


# 收到websocket關(guān)閉的處理
def on_close(ws):
    print("### closed ###")


# 收到websocket連接建立的處理
def on_open(ws):
    def run(*args):
        frameSize = 1280  # 每一幀的音頻大小
        intervel = 0.04  # 發(fā)送音頻間隔(單位:s)
        status = STATUS_FIRST_FRAME  # 音頻的狀態(tài)信息，標(biāo)識音頻是第一幀，還是中間幀、最后一幀
        with open(wsParam.AudioFile, "rb") as fp:
            while True:
                buf = fp.read(frameSize)

                # 文件結(jié)束
                if not buf:
                    status = STATUS_LAST_FRAME
                # 第一幀處理
                # 發(fā)送第一幀音頻，帶business 參數(shù)
                # appid 必須帶上，只需第一幀發(fā)送
                if status == STATUS_FIRST_FRAME:

                    d = {"common": wsParam.CommonArgs,
                         "business": wsParam.BusinessArgs,
                         "data": {"status": 0, "format": "audio/L16;rate=16000",
                                   "audio": str(base64.b64encode(buf),'utf-8'),
                                  "encoding": "raw"}}
                    d = json.dumps(d)
                    ws.send(d)
                    status = STATUS_CONTINUE_FRAME
                # 中間幀處理
                elif status == STATUS_CONTINUE_FRAME:
                    d = {"data": {"status": 1, "format": "audio/L16;rate=16000",
                                   "audio": str(base64.b64encode(buf),'utf-8'),
                                  "encoding": "raw"}}
                    ws.send(json.dumps(d))
                # 最后一幀處理
                elif status == STATUS_LAST_FRAME:
                    d = {"data": {"status": 2, "format": "audio/L16;rate=16000",
                                  "audio": str(base64.b64encode(buf),'utf-8'),
                                  "encoding": "raw"}}
                    ws.send(json.dumps(d))
                    time.sleep(1)
                    break
                # 模擬音頻采樣間隔
                time.sleep(intervel)
        ws.close()

    thread.start_new_thread(run, ())


if __name__ == "__main__":
    wsParam = Ws_Param("ws-api.xfyun.cn") #流式聽寫 域名
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()
    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)
    ws.on_open = on_open
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})


          
        

三、啟動程序

在程序文件夾內(nèi)，右鍵點(diǎn)擊 iat_demo ，選擇 Edit with IDLE -> Edit with IDLE3.7(32 bit) 打開，然后使用 F5 快速啟動，啟動的時(shí)候如果提示沒有哪個(gè)第三方包，自行安裝。啟動成功后根據(jù)提示進(jìn)行操作，這是我修改 json 解析之前操作的打印結(jié)果：

相關(guān)代碼下載