1. 程式人生 > 其它 >Python使用websocket呼叫語音識別,語音轉文字

Python使用websocket呼叫語音識別,語音轉文字

@

目錄

0. 太長不看系列,直接使用

1.2官網註冊後拿到APISecret和APIKey,直接複製文章2.5demo程式碼,儲存為real_time_audio_recognition.py,在命令列執行

python real_time_audio_recognition.py -client_secret=您的client_secret -client_id=您的client_id -file_path=test.wav --audio_format=wav --sample_rate=16000

使用中有任何問題,歡迎留言提問。

1. Python呼叫標貝科技語音識別websocket介面,實現語音轉文字

1.1 環境準備:

Python 3

1.2 獲取許可權

標貝科技 https://ai.data-baker.com/#/index

填寫邀請碼fwwqgs,每日免費呼叫量還可以翻倍

1.2.1 登入

點選產品地址進行登入,支援簡訊、密碼、微信三種方式登入。

1.2.2 建立新應用

登入後進入【首頁概覽】,各位開發者可以進行建立多個應用。包括一句話識別、長語音識別、錄音檔案識別;線上合成、離線合成、長文字合成。

1.2.3 選擇服務

進入【已建立的應用】,左側選擇您需呼叫的AI技術服務,右側展示對應服務頁面概覽(您可查詢用量、管理套餐、購買服務量、自主獲取授權、預警管理)。

1.2.4 獲取Key&Secret

通過服務 / 授權管理,獲取對應引數,進行開發配置(獲取訪問令牌token

拿到Key和Secret就可以正式使用啦!

2. 程式碼實現

2.1 獲取access_token

在拿到Key和Secret後,我們還需要呼叫授權介面獲取access_token,這個access_token有效時長是24小時。

# 獲取access_token用於鑑權
def get_access_token(client_secret, client_id):
    grant_type = "client_credentials"
    url = "https://openapi.data-baker.com/oauth/2.0/token?grant_type={}&client_secret={}&client_id={}" \
        .format(grant_type, client_secret, client_id)

    try:
        response = requests.post(url)
        response.raise_for_status()
    except Exception as e:
        print(response.text)
        raise Exception
    else:
        access_token = json.loads(response.text).get('access_token')
        return access_token

2.2 準備資料

需要根據介面要求設定引數,並且對音訊資料進行分割

# 準備資料
def prepare_data(args, access_token):
    # 讀取音訊檔案
    with open(args.file_path, 'rb') as f:
        file = f.read()

    # 填寫Header資訊
    audio_format = args.audio_format
    sample_rate = args.sample_rate

    splited_data = [str(base64.b64encode(file[i:i + 5120]), encoding='utf-8') for i in range(0, len(file), 5120)]
    asr_params = {"audio_format": audio_format, "sample_rate": int(sample_rate), "speech_type": 1}

    json_list = []
    for i in range(len(splited_data)):
        if i != len(splited_data) - 1:
            asr_params['req_idx'] = i
        else:
            asr_params['req_idx'] = -len(splited_data) + 1
        asr_params["audio_data"] = splited_data[i]
        data = {"access_token": access_token, "version": "1.0", "asr_params": asr_params}

        json_list.append(json.dumps(data))

    return json_list

2.3 配置介面引數

client_secret和client_id:在文章1.2的官網獲取,必填

file_save_path:檔案儲存路徑,必填

audio_format:音訊型別,預設wav格式

sample_rate:取樣率,預設16000Hz

# 獲取命令列輸入引數
def get_args():
    parser = argparse.ArgumentParser(description='ASR')
    parser.add_argument('-client_secret', type=str, required=True)
    parser.add_argument('-client_id', type=str, required=True)
    parser.add_argument('-file_path', type=str, required=True)
    parser.add_argument('--audio_format', type=str, default='wav')
    parser.add_argument('--sample_rate', type=str, default='16000')
    args = parser.parse_args()

    return args

2.4 建立websocket客戶端

class Client:
    def __init__(self, data, uri):
        self.data = data
        self.uri = uri

    #建立連線
    def connect(self):
        ws_app = websocket.WebSocketApp(uri,
                                        on_open=self.on_open,
                                        on_message=self.on_message,
                                        on_error=self.on_error,
                                        on_close=self.on_close)
        ws_app.run_forever()

    # 建立連線後傳送訊息
    def on_open(self, ws):
        print("sending..")
        for i in range(len(self.data)):
            ws.send(self.data[i])

    # 接收訊息
    def on_message(self, ws, message):
        code = json.loads(message).get("code")
        if code != 90000:
            # 列印介面錯誤
            print(message)
        if json.loads(message).get('end_flag') == 1:
            print(json.loads(message).get('asr_text'))

    # 列印錯誤
    def on_error(slef, ws, error):
        print("error: ", str(error))

    # 關閉連線
    def on_close(ws):
        print("client closed.")

2.5 完整demo

import argparse
import json
import base64
import requests
import websocket


class Client:
    def __init__(self, data, uri):
        self.data = data
        self.uri = uri

    #建立連線
    def connect(self):
        ws_app = websocket.WebSocketApp(uri,
                                        on_open=self.on_open,
                                        on_message=self.on_message,
                                        on_error=self.on_error,
                                        on_close=self.on_close)
        ws_app.run_forever()

    # 建立連線後傳送訊息
    def on_open(self, ws):
        print("sending..")
        for i in range(len(self.data)):
            ws.send(self.data[i])

    # 接收訊息
    def on_message(self, ws, message):
        code = json.loads(message).get("code")
        if code != 90000:
            # 列印介面錯誤
            print(message)
        if json.loads(message).get('end_flag') == 1:
            print(json.loads(message).get('asr_text'))

    # 列印錯誤
    def on_error(slef, ws, error):
        print("error: ", str(error))

    # 關閉連線
    def on_close(ws):
        print("client closed.")


# 準備資料
def prepare_data(args, access_token):
    # 讀取音訊檔案
    with open(args.file_path, 'rb') as f:
        file = f.read()

    # 填寫Header資訊
    audio_format = args.audio_format
    sample_rate = args.sample_rate

    splited_data = [str(base64.b64encode(file[i:i + 5120]), encoding='utf-8') for i in range(0, len(file), 5120)]
    asr_params = {"audio_format": audio_format, "sample_rate": int(sample_rate), "speech_type": 1}

    json_list = []
    for i in range(len(splited_data)):
        if i != len(splited_data) - 1:
            asr_params['req_idx'] = i
        else:
            asr_params['req_idx'] = -len(splited_data) + 1
        asr_params["audio_data"] = splited_data[i]
        data = {"access_token": access_token, "version": "1.0", "asr_params": asr_params}

        json_list.append(json.dumps(data))

    return json_list


# 獲取命令列輸入引數
def get_args():
    parser = argparse.ArgumentParser(description='ASR')
    parser.add_argument('-client_secret', type=str, required=True)
    parser.add_argument('-client_id', type=str, required=True)
    parser.add_argument('-file_path', type=str, required=True)
    parser.add_argument('--audio_format', type=str, default='wav')
    parser.add_argument('--sample_rate', type=str, default='16000')
    args = parser.parse_args()

    return args


# 獲取access_token用於鑑權
def get_access_token(client_secret, client_id):
    grant_type = "client_credentials"
    url = "https://openapi.data-baker.com/oauth/2.0/token?grant_type={}&client_secret={}&client_id={}" \
        .format(grant_type, client_secret, client_id)

    try:
        response = requests.post(url)
        response.raise_for_status()
    except Exception as e:
        print(response.text)
        raise Exception
    else:
        access_token = json.loads(response.text).get('access_token')
        return access_token


if __name__ == '__main__':
    try:
        args = get_args()

        # 獲取access_token
        client_secret = args.client_secret
        client_id = args.client_id
        access_token = get_access_token(client_secret, client_id)

        # 準備資料
        data = prepare_data(args, access_token)

        uri = "wss://openapi.data-baker.com/asr/realtime"
        # 建立Websocket連線
        client = Client(data, uri)
        client.connect()
    except Exception as e:
        print(e)

2.5 執行

複製所有程式碼,確定音訊為wav格式,取樣率為16K,在命令列執行

python real_time_audio_recognition.py -client_secret=您的client_secret -client_id=您的client_id -file_path=test.wav --audio_format=wav --sample_rate=16000

填寫邀請碼fwwqgs,每日免費呼叫量還可以翻倍