本文介紹如何使用阿里云智能語音服務提供的Python SDK,包括SDK的安裝方法及SDK代碼示例。
前提條件
在使用SDK前,請先閱讀接口說明,詳情請參見接口說明。
下載安裝
SDK僅支持Python3,暫不支持Python2。
請確認已安裝Python包管理工具setuptools。如果沒有安裝,請在終端使用以下命令安裝:
pip install setuptools
下載Python SDK。
從Github獲取Python SDK,或直接下載streamInputTts-github-python。
安裝SDK依賴。
進入SDK根目錄使用如下命令安裝SDK依賴:
python -m pip install -r requirements.txt
安裝SDK。
依賴安裝完成后使用如下命令安裝SDK:
python -m pip install .
安裝完成后通過以下代碼導入SDK。
# -*- coding: utf-8 -*- import nls
上述命令均需要在SDK根目錄中執行。
多線程和多并發
在CPython中,由于存在全局解釋器鎖,同一時刻只有一個線程可以執行Python代碼(雖然某些性能導向的庫可能會去除此限制)。如果您想更好地利用多核心計算機的計算資源,推薦你使用multiprocessing或concurrent.futures.ProcessPoolExecutor。 如果你想要同時運行多個I/O密集型任務,則多線程仍然是一個合適的模型。
如果單解釋器有太多線程,將會在線程間切換造成更多消耗,有可能會導致SDK出現錯誤。不建議使用超過200線程,推薦使用multiprocessing技術或者手動使用腳本創建多個解釋器。
關鍵接口
語音合成對應的類為NlsSpeechSynthesizer,其核心方法如下:
1. 初始化(__init__)
參數說明
參數 | 類型 | 參數說明 |
url | String | 網關WebSocket URL地址,默認為 |
appkey | String | Appkey,獲取方式請參見管理項目。 |
long_tts | bool | 語音合成方式,取值說明如下:
|
token | String | 訪問Token,詳情可參見獲取Token概述。 |
on_metainfo | Function | 如果start方法中通過ex參數傳遞enable_subtitle,則會返回對應字幕信息?;卣{參數包含以下兩種:
其中,用戶自定義參數為下方callback_args字段中返回的參數內容。 |
on_data | Function | 當存在合成數據后的回調參數。回調參數包含以下兩種:
其中,用戶自定義參數為下方callback_args字段中返回的參數內容。 |
on_error | Function | 當SDK或云端出現錯誤時的回調參數?;卣{參數包含以下兩種:
其中,用戶自定義參數為下方callback_args字段中返回的參數內容。 |
on_close | Function | 當和云端連接斷開時的回調參數?;卣{參數為用戶自定義參數,即用戶自定義參數為下方callback_args字段中返回的參數內容。 |
callback_args | List | 用戶自定義參數列表,列表中的內容會打包(pack)成List數據結構傳遞給各個回調的最后一個參數。 |
返回值:無
2. start
同步開始語音合成,如果wait_complete為True(默認),則會阻塞直到所有音頻合成完畢(on_completed返回之后)返回,否則會立即返回。
參數說明
參數 | 類型 | 參數說明 |
text | String | 要合成的文字。 |
aformat | String | 合成出來音頻的格式,支持輸出PCM、WAV和MP3編碼格式數據,默認為pcm。 |
voice | String | 發音人,默認為xiaoyun。 |
sample_rate | Integer | 識別音頻采樣率,默認值:16000 Hz。 |
volume | Integer | 音量大小,取值范圍0~100,默認值:50。 |
speech_rate | Integer | 語速,取值范圍-500~500,默認值:0。 |
pitch_rate | Integer | 語調,取值范圍-500~500,默認值:0。 |
wait_complete | Boolean | 是否阻塞到合成完成。 |
start_timeout | Integer | 和云端連接建立超時,默認值:10秒。 |
completed_timeout | Integer | 從連接建立到合成完成超時,默認值:60秒 |
ping_interval | Integer | Ping包發送間隔,默認值:8秒。無需間隔可設置為0或None。 |
ping_timeout | Integer | 是否檢查Pong包超時,默認值:None。None為不檢查Pong包是否超時。 |
ex | Dict | 用戶提供的額外參數,該字典內容會以 |
返回值:無
3. shutdown
強行關閉當前請求,重復調用無副作用。
參數說明:無
返回值:無
代碼示例
本示例中將合成的音頻保存在文件中,如果您需要播放音頻且對實時性要求較高,建議使用流式播放,即邊接收語音數據邊播放,減少延時。
本示例中使用SDK內置的默認外網訪問服務端URL,如果您使用阿里云上海地域的ECS,并需要通過內網訪問服務端URL,請使用如下URL:
URL="wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
。
import time
import threading
import sys
import nls
URL="wss://nls-gateway-cn-shanghai.aliyuncs.com/ws/v1"
TOKEN="yourToken" #參考http://bestwisewords.com/document_detail/450255.html獲取token
APPKEY="yourAppkey" #獲取Appkey請前往控制臺:https://nls-portal.console.aliyun.com/applist
TEXT='大壯正想去摘取花瓣,誰知阿麗和阿強突然內訌,阿麗拿去手槍向樹干邊的阿強射擊,兩聲槍響,阿強直接倒入水中'
#以下代碼會根據上述TEXT文本反復進行語音合成
class TestTts:
def __init__(self, tid, test_file):
self.__th = threading.Thread(target=self.__test_run)
self.__id = tid
self.__test_file = test_file
def start(self, text):
self.__text = text
self.__f = open(self.__test_file, "wb")
self.__th.start()
def test_on_metainfo(self, message, *args):
print("on_metainfo message=>{}".format(message))
def test_on_error(self, message, *args):
print("on_error args=>{}".format(args))
def test_on_close(self, *args):
print("on_close: args=>{}".format(args))
try:
self.__f.close()
except Exception as e:
print("close file failed since:", e)
def test_on_data(self, data, *args):
try:
self.__f.write(data)
except Exception as e:
print("write data failed:", e)
def test_on_completed(self, message, *args):
print("on_completed:args=>{} message=>{}".format(args, message))
def __test_run(self):
print("thread:{} start..".format(self.__id))
tts = nls.NlsSpeechSynthesizer(url=URL,
token=TOKEN,
appkey=APPKEY,
on_metainfo=self.test_on_metainfo,
on_data=self.test_on_data,
on_completed=self.test_on_completed,
on_error=self.test_on_error,
on_close=self.test_on_close,
callback_args=[self.__id])
print("{}: session start".format(self.__id))
r = tts.start(self.__text, voice="ailun")
print("{}: tts done with result:{}".format(self.__id, r))
def multiruntest(num=500):
for i in range(0, num):
name = "thread" + str(i)
t = TestTts(name, "tests/test_tts.pcm")
t.start(TEXT)
nls.enableTrace(True)
multiruntest(1)