CosyVoice Python SDK

# coding=utf-8

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

# If you have not configured environment variables, replace the next line with your API key: dashscope.api_key = "YOUR_API_KEY"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# Model
model = "cosyvoice-v3-flash"
# Voice
voice = "longanyang"

# Instantiate SpeechSynthesizer and pass request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(model=model, voice=voice)
# Send text for synthesis and get binary audio
audio = synthesizer.call("How is the weather today?")
# First request includes WebSocket connection setup time
print('[Metric] requestId: {}, first-package delay: {} ms'.format(
  synthesizer.get_last_request_id(),
  synthesizer.get_first_package_delay()))

# Save audio to local file
with open('output.mp3', 'wb') as f:
  f.write(audio)

# coding=utf-8

import os
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

# If you have not configured environment variables, replace the next line with your API key: dashscope.api_key = "YOUR_API_KEY"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# Model
model = "cosyvoice-v3-flash"
# Voice
voice = "longanyang"


# Define callback interface
class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    self.file = open("output.mp3", "wb")
    print("Connection established: " + get_timestamp())

  def on_complete(self):
    print("Speech synthesis completed, all results received: " + get_timestamp())
    # Call get_first_package_delay only after on_complete triggers
    # First request includes WebSocket connection setup time
    print('[Metric] requestId: {}, first-package delay: {} ms'.format(
      synthesizer.get_last_request_id(),
      synthesizer.get_first_package_delay()))

  def on_error(self, message: str):
    print(f"Speech synthesis error: {message}")

  def on_close(self):
    print("Connection closed: " + get_timestamp())
    self.file.close()

  def on_event(self, message):
    pass

  def on_data(self, data: bytes) -> None:
    print(get_timestamp() + " Binary audio length: " + str(len(data)))
    self.file.write(data)


callback = Callback()

# Instantiate SpeechSynthesizer and pass request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(
  model=model,
  voice=voice,
  callback=callback,
)

# Send text for synthesis and retrieve binary audio in real time through the on_data method of the callback interface
synthesizer.call("How is the weather today?")

# coding=utf-8
#
# PyAudio installation instructions:
# For macOS, run:
#   brew install portaudio
#   pip install pyaudio
# For Debian/Ubuntu, run:
#   sudo apt-get install python-pyaudio python3-pyaudio
#   or
#   pip install pyaudio
# For CentOS, run:
#   sudo yum install -y portaudio portaudio-devel && pip install pyaudio
# For Microsoft Windows, run:
#   python -m pip install pyaudio

import os
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

# If you have not configured environment variables, replace the next line with your API key: dashscope.api_key = "YOUR_API_KEY"
dashscope.api_key = os.environ.get('DASHSCOPE_API_KEY')

dashscope.base_websocket_api_url='wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference'

# Model
model = "cosyvoice-v3-flash"
# Voice
voice = "longanyang"


# Define callback interface
class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    print("Connection established: " + 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("Speech synthesis completed, all results received: " + get_timestamp())

  def on_error(self, message: str):
    print(f"Speech synthesis error: {message}")

  def on_close(self):
    print("Connection closed: " + get_timestamp())
    # Stop player
    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() + " Binary audio length: " + str(len(data)))
    self._stream.write(data)


callback = Callback()

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.",
]

# Instantiate SpeechSynthesizer and pass request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(
  model=model,
  voice=voice,
  format=AudioFormat.PCM_22050HZ_MONO_16BIT,
  callback=callback,
)


# Stream text for synthesis. Retrieve binary audio in real time through the on_data method of the callback interface
for text in test_text:
  synthesizer.streaming_call(text)
  time.sleep(0.1)
# End streaming speech synthesis
synthesizer.streaming_complete()

# First request includes WebSocket connection setup time
print('[Metric] requestId: {}, first-package delay: {} ms'.format(
  synthesizer.get_last_request_id(),
  synthesizer.get_first_package_delay()))

# coding=utf-8

import dashscope
from dashscope.audio.tts_v2 import *
import json
from datetime import datetime


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


# If you have not configured your API key in an environment variable, replace YOUR_API_KEY with your actual key
# dashscope.api_key = "YOUR_API_KEY"

model = "cosyvoice-v3-flash"
# Voice
voice = "longanyang"


# Define callback interface
class Callback(ResultCallback):
  _player = None
  _stream = None

  def on_open(self):
    self.file = open("output.mp3", "wb")
    print("Connection established: " + get_timestamp())

  def on_complete(self):
    print("Speech synthesis completed, all results received: " + get_timestamp())

  def on_error(self, message: str):
    print(f"Speech synthesis error: {message}")

  def on_close(self):
    print("Connection closed: " + get_timestamp())
    self.file.close()

  def on_event(self, message):
    json_data = json.loads(message)
    if json_data['payload'] and json_data['payload']['output'] and json_data['payload']['output']['sentence']:
      sentence = json_data['payload']['output']['sentence']
      print(f'sentence: {sentence}')
      # Get sentence index
      # index = sentence['index']
      words = sentence['words']
      if words:
        for word in words:
          print(f'word: {word}')
          # Example: word: {'text': 'T', 'begin_index': 0, 'end_index': 1, 'begin_time': 80, 'end_time': 200}

  def on_data(self, data: bytes) -> None:
    print(get_timestamp() + " Binary audio length: " + str(len(data)))
    self.file.write(data)


callback = Callback()

# Instantiate SpeechSynthesizer and pass request parameters such as model and voice in the constructor
synthesizer = SpeechSynthesizer(
  model=model,
  voice=voice,
  callback=callback,
  additional_params={'word_timestamp_enabled': True}
)

# Send text for synthesis and retrieve binary audio in real time through the on_data method of the callback interface
synthesizer.call("How is the weather today?")
# First request includes WebSocket connection setup time
print('[Metric] requestId: {}, first-package delay: {} ms'.format(
  synthesizer.get_last_request_id(),
  synthesizer.get_first_package_delay()))

class Callback(ResultCallback):
  def on_open(self) -> None:
    print('Connection successful')

  def on_data(self, data: bytes) -> None:
    # Implement logic to handle binary audio data
    pass

  def on_complete(self) -> None:
    print('Synthesis complete')

  def on_error(self, message) -> None:
    print('Error: ', message)

  def on_close(self) -> None:
    print('Connection closed')


callback = Callback()