Qwen-Audio-TTS Python SDK

# coding=utf-8

import os

from datetime import datetime

import dashscope

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

# API キーは、シンガポールリージョンと北京リージョンで異なります。API キーの取得： https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください： api_key="sk-xxx"
# 以下はシンガポール (ap-southeast-1) リージョンの設定です。呼び出す際に、"{WorkspaceId}" を実際のワークスペース ID に置き換えてください。設定はリージョンによって異なります。
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# モデル
model = "qwen-audio-3.0-tts-flash"
# 音声
voice = "longanhuan_v3.6"

# 完全なテキストを一度に送信し、バイナリオーディオをチャンク単位で受信します
with open("output.mp3", "wb") as f:
    for chunk in client.audio.tts.stream("How is the weather today?", model=model, voice=voice):
        print(get_timestamp() + " Binary audio length: " + str(len(chunk)))
        f.write(chunk)
print("Speech synthesis completed, all results received: " + get_timestamp())

# 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
from concurrent.futures import ThreadPoolExecutor

import dashscope

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 キーは、シンガポールリージョンと北京リージョンで異なります。API キーの取得： https://www.alibabacloud.com/help/model-studio/get-api-key
# 環境変数を設定していない場合は、次の行を Model Studio API キーに置き換えてください： api_key="sk-xxx"
# 以下はシンガポール (ap-southeast-1) リージョンの設定です。呼び出す際に、"{WorkspaceId}" を実際のワークスペース ID に置き換えてください。設定はリージョンによって異なります。
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# モデル
model = "qwen-audio-3.0-tts-flash"
# 音声
voice = "longanhuan_v3.6"

test_text = [
    "Streaming text-to-speech SDK,",
    "converts input text",
    "into binary audio data.",
    "Compared with non-streaming speech synthesis,",
    "streaming synthesis offers better real-time performance.",
    "Users hear near-synchronous audio output while typing,",
    "greatly improving interaction experience",
    "and reducing wait time.",
    "Ideal for large language model (LLM) integration,",
    "where text is streamed for speech synthesis.",
]

# 双方向ストリーミング合成セッションを作成します：with コンテキストに入ると接続が自動的に確立されます
with client.audio.tts.connect(
        model=model,
        voice=voice,
        format="pcm",
        sample_rate=22050,
) as session:
    # プレーヤーを初期化します
    player = pyaudio.PyAudio()
    stream = player.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)

    # 受信スレッド：音声データを反復処理し、リアルタイムで再生します
    def receive_audio():
        for chunk in session.audio_chunks():
            print(get_timestamp() + " Binary audio length: " + str(len(chunk)))
            stream.write(chunk)

    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_audio)
        # ストリーミング合成のためにテキストを送信します
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # ストリーミング音声合成を終了します
        session.stop()
        receiver.result()

    print("Speech synthesis completed, all results received: " + get_timestamp())
    # プレーヤーを停止します
    stream.stop_stream()
    stream.close()
    player.terminate()

    # 最初のテキスト送信では WebSocket 接続を確立する必要があるため、初回パケットレイテンシーには接続確立時間が含まれます
    print('[Metric] first packet latency: {} ms'.format(session.first_package_delay))