全部产品
云市场
云游戏

Python SDK

更新时间:2020-08-12 16:40:27

本文介绍如何使用阿里云智能语音服务提供的Python SDK,包括SDK的安装方法及SDK代码示例。

前提条件

使用SDK前,请先阅读接口说明,详情请参见接口说明

下载安装

说明

  • SDK仅支持Python3.4及以上版本,暂不支持Python 2。

  • 确认已安装Python包管理工具setuptools,如果没有安装,请在Linux终端使用以下命令安装:

      pip install setuptools
  1. 下载Python SDK

  2. 安装SDK。

    在Linux终端的SDK目录下,执行以下命令:

     # 打包
     python setup.py bdist_egg
     # 安装
     python setup.py install

关键接口

  • NlsClient:语音处理客户端,利用该客户端可以进行一句话识别、实时语音识别和语音合成的语音处理任务。该客户端为线程安全,建议全局仅创建一个实例。

  • SpeechSynthesizer:语音合成处理类,通过该接口设置请求参数,发送请求。非线程安全。

    • start方法:建立与服务端的连接。默认参数ping_interval表示自动发送ping命令的间隔,ping_timeout表示等待ping命令响应的pong消息的超时时间,需要满足ping_interval 大于 ping_timeout。

    • wait_completed方法:等待服务端合成完毕或者合成超时。

    • close方法:关闭与服务端的网络链接。

  • SpeechSynthesizerCallback:回调事件函数集合的类,合成结果、异常等回调的统一入口。

    • on_binary_data_received方法:客户端接收到合成音频数据的回调。

    • on_completed方法:客户端接收到合成结束的回调。

    • on_completed方法:客户端接收到异常错误的回调。

    • on_channel_closed方法:客户端接收到断开网络链接成功的回调。

SDK调用注意事项

  • NlsClient对象创建一次可以重复使用。

  • 一个合成任务对应一个SpeechSynthesizer对象。例如有N个文本需要合成,则要进行N次语音合成任务,创建N个SpeechSynthesizer对象。

  • SpeechSynthesizerCallback对象和SpeechSynthesizer对象是一一对应的,不能将一个实现的SpeechSynthesizerCallback对象设置到多个SpeechSynthesizer对象中,否则无法将识别任务区分开。

示例代码

说明

  • 示例使用SDK内置的默认语音合成服务的外网访问服务端URL,如果您使用上海阿里云ECS并想使用内网访问服务端URL,则在创建NlsClient对象时,设置内网访问的URL:

    synthesizer=client.create_synthesizer(callback,"ws://nls-gateway.cn-shanghai-internal.aliyuncs.com/ws/v1")ynthesizer=client.create_synthesizer(callback,"ws://nls-gateway.cn-shanghai-internal.aliyuncs.com/ws/v1")
  • 示例将合成的音频保存在文件中,如果您需要播放音频且对实时性要求较高,建议使用流式播放,即边接收语音数据边播放,减少延时。

# -*- coding: utf-8 -*-
import threading
import ali_speech
from ali_speech.callbacks import SpeechSynthesizerCallback
from ali_speech.constant import TTSFormat
from ali_speech.constant import TTSSampleRate
class MyCallback(SpeechSynthesizerCallback):
    # 参数name用于指定保存音频的文件。
    def __init__(self, name):
        self._name = name
        self._fout = open(name, 'wb')
    def on_binary_data_received(self, raw):
        print('MyCallback.on_binary_data_received: %s' % len(raw))
        self._fout.write(raw)
    def on_completed(self, message):
        print('MyCallback.OnRecognitionCompleted: %s' % message)
        self._fout.close()
    def on_task_failed(self, message):
        print('MyCallback.OnRecognitionTaskFailed-task_id:%s, status_text:%s' % (
            message['header']['task_id'], message['header']['status_text']))
        self._fout.close()
    def on_channel_closed(self):
        print('MyCallback.OnRecognitionChannelClosed')
def process(client, appkey, token, text, audio_name):
    callback = MyCallback(audio_name)
    synthesizer = client.create_synthesizer(callback)
    synthesizer.set_appkey(appkey)
    synthesizer.set_token(token)
    synthesizer.set_voice('xiaoyun')
    synthesizer.set_text(text)
    synthesizer.set_format(TTSFormat.WAV)
    synthesizer.set_sample_rate(TTSSampleRate.SAMPLE_RATE_16K)
    synthesizer.set_volume(50)
    synthesizer.set_speech_rate(0)
    synthesizer.set_pitch_rate(0)
    try:
        ret = synthesizer.start()
        if ret < 0:
            return ret
        synthesizer.wait_completed()
    except Exception as e:
        print(e)
    finally:
        synthesizer.close()
def process_multithread(client, appkey, token, number):
    thread_list = []
    for i in range(0, number):
        text = "这是线程" + str(i) + "的合成。"
        audio_name = "sy_audio_" + str(i) + ".wav"
        thread = threading.Thread(target=process, args=(client, appkey, token, text, audio_name))
        thread_list.append(thread)
        thread.start()
    for thread in thread_list:
        thread.join()
if __name__ == "__main__":
    client = ali_speech.NlsClient()
    # 设置输出日志信息的级别:DEBUG、INFO、WARNING、ERROR。
    client.set_log_level('INFO')
    appkey = '您的appkey'
    token = '您的token'
    text = "今天是周一,天气挺好的。"
    audio_name = 'sy_audio.wav'
    process(client, appkey, token, text, audio_name)
    # 多线程示例
    # process_multithread(client, appkey, token, 2)