SDK de Python de CosyVoice

# 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. Sustituya {WorkspaceId} por el ID real de su espacio de trabajo. El valor de region 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 = "cosyvoice-v3-plus"
# Voz
voice = "longanyang"

# 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
#
# pyaudio installation instructions:
# For macOS, run the following commands:
#   brew install portaudio
#   pip install pyaudio
# For Debian/Ubuntu, run the following commands:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   or
#   pip install pyaudio
# For CentOS, run the following commands:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# For Microsoft Windows, run the following command:
#   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. Sustituya {WorkspaceId} por el ID real de su espacio de trabajo. El valor de region 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 = "cosyvoice-v3-plus"
# Voz
voice = "longanyang"

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 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 de recepción: itere sobre 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 a sintetizar en modo streaming
        for text in test_text:
            session.send_text(text)
            time.sleep(0.1)
        # Finalice la síntesis 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()

    # La primera conexión WebSocket se establece cuando se envía el primer texto, por lo que la latencia del primer paquete incluye el tiempo de establecimiento de la conexión
    print('[Metric] first package delay: {} ms'.format(session.first_package_delay))