SDK de Python de Qwen-Audio-TTS

# 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

# Las API Key de las regiones de Singapur y Pekín son diferentes. Obtenga su API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Si no se ha configurado la variable de entorno, sustituya la siguiente línea por su API Key de Model Studio: api_key="sk-xxx"
# La siguiente configuración corresponde a la región de Singapur (ap-southeast-1). Sustituya "{WorkspaceId}" por el ID real de su espacio de trabajo. La configuración varía según la región.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Modelo
model = "qwen-audio-3.0-tts-flash"
# Voz
voice = "longanhuan_v3.6"

# Envíe el texto completo de una sola vez y reciba el audio binario fragmento a fragmento
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
#
# Instrucciones de instalación de pyaudio:
# Para macOS, ejecute los siguientes comandos:
#   brew install portaudio
#   pip install pyaudio
# Para Debian/Ubuntu, ejecute los siguientes comandos:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   o
#   pip install pyaudio
# Para CentOS, ejecute los siguientes comandos:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# Para Microsoft Windows, ejecute el siguiente comando:
#   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

# Las API Key de las regiones de Singapur y Pekín son diferentes. Obtenga su API Key: https://www.alibabacloud.com/help/model-studio/get-api-key
# Si no se ha configurado la variable de entorno, sustituya la siguiente línea por su API Key de Model Studio: api_key="sk-xxx"
# La siguiente configuración corresponde a la región de Singapur (ap-southeast-1). Sustituya "{WorkspaceId}" por el ID real de su espacio de trabajo. La configuración varía según la región.
client = dashscope.Client(
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    workspace="{WorkspaceId}",
    region="ap-southeast-1",
)

# Modelo
model = "qwen-audio-3.0-tts-flash"
# Voz
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.",
]

# Cree una sesión de síntesis por streaming bidireccional: la conexión se establece automáticamente al entrar en el contexto with
with client.audio.tts.connect(
        model=model,
        voice=voice,
        format="pcm",
        sample_rate=22050,
) as session:
    # Inicialice el reproductor
    player = pyaudio.PyAudio()
    stream = player.open(format=pyaudio.paInt16, channels=1, rate=22050, output=True)

    # Hilo receptor: itere los datos de audio y reprodúzcalos en tiempo real
    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)
        # Envíe el texto para la síntesis por streaming
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # Finalice la síntesis de voz por streaming
        session.stop()
        receiver.result()

    print("Speech synthesis completed, all results received: " + get_timestamp())
    # Detenga el reproductor
    stream.stop_stream()
    stream.close()
    player.terminate()

    # El primer envío de texto requiere establecer una conexión WebSocket, por lo que la latencia del primer paquete incluye el tiempo de establecimiento de la conexión
    print('[Metric] first packet latency: {} ms'.format(session.first_package_delay))