Alibaba Cloud Model Studio:CosyVoice 音声合成 Python SDK

• ノンストリーミング：完全なテキストを一度に送信し、完全な音声を直接返すブロッキング呼び出しです。短いテキストの音声合成に最も適しています。

• 単方向ストリーミング：完全なテキストを一度に送信し、コールバック関数を介して音声データ (チャンク単位の場合もあります) を配信する非ブロッキング呼び出しです。低レイテンシーが求められる短いテキストのシナリオに最も適しています。

• 双方向ストリーミング：テキストを複数のセグメントで送信し、コールバック関数を介してリアルタイムで段階的に合成された音声を配信する非ブロッキング呼び出しです。低レイテンシーが求められる長いテキストのシナリオに最も適しています。

# coding=utf-8

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

# Singapore リージョンと Beijing リージョンの API キーは異なります。API キーの取得：https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数が設定されていない場合は、次の行の "sk-xxx" をご自身の Model Studio API キーに置き換えてください。
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下は Singapore リージョンの URL です。Beijing リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# モデル
model = "cosyvoice-v3-flash"
# 音声
voice = "longanyang"

# SpeechSynthesizer をインスタンス化し、コンストラクターでモデルや音声などのリクエストパラメーターを渡します
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# 合成用のテキストを送信し、バイナリオーディオを取得します
audio = synthesizer.call("How is the weather today?")
# 最初のテキスト送信には WebSocket 接続の確立が必要なため、初回パケットレイテンシーには接続セットアップ時間が含まれます
print('[Metric] requestId: {}, first packet latency: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))

# オーディオをローカルファイルに保存します
with open('output.mp3', 'wb') as f:
    f.write(audio)

# 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

# Singapore リージョンと Beijing リージョンの API キーは異なります。API キーの取得：https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数が設定されていない場合は、次の行の "sk-xxx" をご自身の Model Studio API キーに置き換えてください。
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下は Singapore リージョンの URL です。Beijing リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# モデル
model = "cosyvoice-v3-flash"
# 音声
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: {}, first packet latency: {} ms'.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 をインスタンス化し、コンストラクターでモデルや音声などのリクエストパラメーターを渡します
synthesizer = SpeechSynthesizer(
    model=model,
    voice=voice,
    callback=callback,
)

# 合成用のテキストを送信し、on_data コールバックメソッドを通じてリアルタイムでバイナリオーディオを取得します
synthesizer.call("How is the weather today?")

# 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

# Singapore リージョンと Beijing リージョンの API キーは異なります。API キーの取得：https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数が設定されていない場合は、次の行の "sk-xxx" をご自身の Model Studio API キーに置き換えてください。
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

# 以下は Singapore リージョンの URL です。Beijing リージョンのモデルを使用する場合は、URL を次のように置き換えてください：wss://dashscope.aliyuncs.com/api-ws/v1/inference
dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# モデル
model = "cosyvoice-v3-flash"
# 音声
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())
        # タスク完了後 (on_complete コールバックがトリガーされた後)、get_first_package_delay を呼び出してレイテンシーを取得できます
        # 最初のテキスト送信には WebSocket 接続の確立が必要なため、初回パケットレイテンシーには接続セットアップ時間が含まれます
        print('[Metric] requestId: {}, first packet latency: {} ms'.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._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 をインスタンス化し、コンストラクターでモデルや音声などのリクエストパラメーターを渡します
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()