CosyVoice Python SDK

# coding=utf-8

import os
import json
import dashscope
from dashscope.audio.tts_v2 import *

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# 新加坡地區和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# 若沒有配置環境變數，請用百鍊API Key將下行替換為：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

# 模型
model = "cosyvoice-v3-plus"
# 音色
voice = "longanyang"

# 定義回調介面
class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        self.file = open("output.mp3", "wb")
        print("串連建立：" + get_timestamp())

    def on_complete(self):
        print("語音合成完成，所有合成結果已被接收：" + get_timestamp())
        # 當任務完成（on_complete 回調觸發）後，才可調用 get_first_package_delay 擷取延遲
        # 首次發送文本時需建立 WebSocket 串連，因此首包延遲會包含串連建立的耗時
        print('[Metric] requestId為：{}，首包延遲為：{}毫秒'.format(
            synthesizer.get_last_request_id(),
            synthesizer.get_first_package_delay()))

    def on_error(self, message: str):
        print(f"語音合成出現異常：{message}")

    def on_close(self):
        print("串連關閉：" + get_timestamp())
        self.file.close()

    def on_event(self, message):
        # 解析服務端事件，擷取輸出資訊
        data = json.loads(message)
        output = data.get('payload', {}).get('output', {})
        event_type = output.get('type', '')
        original_text = output.get('original_text', '')
        if event_type:
            print(f"事件類型: {event_type}, 原始文本: {original_text}")

    def on_data(self, data: bytes) -> None:
        print(get_timestamp() + " 二進位音頻長度為：" + str(len(data)))
        self.file.write(data)

callback = Callback()

# 執行個體化SpeechSynthesizer，並在構造方法中傳入模型（model）、音色（voice）等請求參數
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    callback=callback,
)

# 發送待合成文本，在回調介面的on_data方法中即時擷取二進位音頻
synthesizer.call("今天天氣怎麼樣？")

# coding=utf-8
#
# pyaudio安裝說明：
# 如果是macOS作業系統，執行如下命令：
#   brew install portaudio
#   pip install pyaudio
# 如果是Debian/Ubuntu作業系統，執行如下命令：
#   sudo apt-get install python-pyaudio python3-pyaudio
#   或者
#   pip install pyaudio
# 如果是CentOS作業系統，執行如下命令：
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# 如果是Microsoft Windows，執行如下命令：
#   python -m pip install pyaudio

import os
import time
import pyaudio
import json
import dashscope
from dashscope.api_entities.dashscope_response import SpeechSynthesisResponse
from dashscope.audio.tts_v2 import *

from datetime import datetime

def get_timestamp():
    now = datetime.now()
    formatted_timestamp = now.strftime("[%Y-%m-%d %H:%M:%S.%f]")
    return formatted_timestamp

# 新加坡地區和北京地區的API Key不同。擷取API Key：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
# 若沒有配置環境變數，請用百鍊API Key將下行替換為：dashscope.api_key = "sk-xxx"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下為新加坡地區的配置，調用時請將"{WorkspaceId}"替換為真實的業務空間ID，各地區的配置不同。
dashscope.base_websocket_api_url='wss://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference'

# 模型
model = "cosyvoice-v3-plus"
# 音色
voice = "longanyang"

# 定義回調介面
class Callback(ResultCallback):
    _player = None
    _stream = None

    def on_open(self):
        print("串連建立：" + get_timestamp())
        self._player = pyaudio.PyAudio()
        self._stream = self._player.open(
            format=pyaudio.paInt16, channels=1, rate=22050, output=True
        )

    def on_complete(self):
        print("語音合成完成，所有合成結果已被接收：" + get_timestamp())

    def on_error(self, message: str):
        print(f"語音合成出現異常：{message}")

    def on_close(self):
        print("串連關閉：" + get_timestamp())
        # 停止播放器
        self._stream.stop_stream()
        self._stream.close()
        self._player.terminate()

    def on_event(self, message):
        # 解析服務端事件，擷取輸出資訊
        data = json.loads(message)
        output = data.get('payload', {}).get('output', {})
        event_type = output.get('type', '')
        original_text = output.get('original_text', '')
        if event_type:
            print(f"事件類型: {event_type}, 原始文本: {original_text}")

    def on_data(self, data: bytes) -> None:
        print(get_timestamp() + " 二進位音頻長度為：" + str(len(data)))
        self._stream.write(data)

callback = Callback()

test_text = [
    "流式文本語音合成SDK，",
    "可以將輸入的文本",
    "合成為語音位元據，",
    "相比於非流式語音合成，",
    "流式合成的優勢在於即時性",
    "更強。使用者在輸入文本的同時",
    "可以聽到接近同步的語音輸出，",
    "極大地提升了互動體驗，",
    "減少了使用者等待時間。",
    "適用於調用大規模",
    "語言模型（LLM），以",
    "流式輸入文本的方式",
    "進行語音合成的情境。",
]

# 執行個體化SpeechSynthesizer，並在構造方法中傳入模型（model）、音色（voice）等請求參數
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    format=AudioFormat.PCM_22050HZ_MONO_16BIT,
    callback=callback,
)

# 流式發送待合成文本。在回調介面的on_data方法中即時擷取二進位音頻
for text in test_text:
    synthesizer.streaming_call(text)
    time.sleep(0.1)
# 結束流式語音合成
synthesizer.streaming_complete()

# 首次發送文本時需建立 WebSocket 串連，因此首包延遲會包含串連建立的耗時
print('[Metric] requestId為：{}，首包延遲為：{}毫秒'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))