Alibaba Cloud Model Studio:CosyVoice speech synthesis Python SDK

# coding=utf-8

import dashscope
from dashscope.audio.tts_v2 import *

# API キーを環境変数として設定していない場合は、「your-api-key」を実際の API キーに置き換えてください。
# dashscope.api_key = "your-api-key"

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

# SpeechSynthesizer をインスタンス化し、コンストラクターでモデルや音声などのリクエストパラメーターを渡します。
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# 合成するテキストを送信し、バイナリ音声データを取得します。
audio = synthesizer.call("今日の天気はどうですか？")
# 初めてテキストを送信する際には、WebSocket 接続を確立する必要があります。そのため、初回パケット遅延には接続確立にかかる時間が含まれます。
print('[メトリック] リクエスト ID: {}, 初回パケット遅延: {} 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 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 キーを環境変数として設定していない場合は、「your-api-key」を実際の API キーに置き換えてください。
# dashscope.api_key = "your-api-key"

# モデル
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('[メトリック] リクエスト ID: {}, 初回パケット遅延: {} 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):
        pass

    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("今日の天気はどうですか？")

# 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 time
import pyaudio
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 キーを環境変数として設定していない場合は、「your-api-key」を実際の API キーに置き換えてください。
# dashscope.api_key = "your-api-key"

# モデル
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())

    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):
        pass

    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()

# 初めてテキストを送信する際には、WebSocket 接続を確立する必要があります。そのため、初回パケット遅延には接続確立にかかる時間が含まれます。
print('[メトリック] リクエスト ID: {}, 初回パケット遅延: {} ms'.format(
    synthesizer.get_last_request_id(),
    synthesizer.get_first_package_delay()))

class Callback(ResultCallback):
    def on_open(self) -> None:
        print('接続に成功しました')
    
    def on_data(self, data: bytes) -> None:
        # 合成されたバイナリ音声結果を受信するロジックを実装します。

    def on_complete(self) -> None:
        print('合成が完了しました')

    def on_error(self, message) -> None:
        print('例外が発生しました: ', message)

    def on_close(self) -> None:
        print('接続が閉じられました')


callback = Callback()