Chirp 3: 即時カスタム音声

import requests, os, json

def create_instant_custom_voice_key(
    access_token, project_id, reference_audio_bytes, consent_audio_bytes
):
    url = "https://texttospeech.googleapis.com/v1beta1/voices:generateVoiceCloningKey"

    request_body = {
        "reference_audio": {
            "audio_config": {"audio_encoding": "LINEAR16", "sample_rate_hertz": 24000},
            "content": reference_audio_bytes,
        },
        "voice_talent_consent": {
            "audio_config": {"audio_encoding": "LINEAR16", "sample_rate_hertz": 24000},
            "content": consent_audio_bytes,
        },
        "consent_script": "I am the owner of this voice and I consent to Google using this voice to create a synthetic voice model.",
        "language_code": "en-US",
    }

    try:
        headers = {
            "Authorization": f"Bearer {access_token}",
            "x-goog-user-project": project_id,
            "Content-Type": "application/json; charset=utf-8",
        }

        response = requests.post(url, headers=headers, json=request_body)
        response.raise_for_status()

        response_json = response.json()
        return response_json.get("voiceCloningKey")

    except requests.exceptions.RequestException as e:
        print(f"Error making API request: {e}")
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON response: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

import requests, os, json, base64
from IPython.display import Audio, display

def synthesize_text_with_cloned_voice(access_token, project_id, voice_key, text):
    url = "https://texttospeech.googleapis.com/v1beta1/text:synthesize"

    request_body = {
        "input": {
            "text": text
        },
        "voice": {
            "language_code": "en-US",
            "voice_clone": {
                "voice_cloning_key": voice_key,
            }
        },
        "audioConfig": {
            "audioEncoding": "LINEAR16",
            "sample_rate_hertz": 24000
        }
    }

    try:
        headers = {
            "Authorization": f"Bearer {access_token}",
            "x-goog-user-project": project_id,
            "Content-Type": "application/json; charset=utf-8"
        }

        response = requests.post(url, headers=headers, json=request_body)
        response.raise_for_status()

        response_json = response.json()
        audio_content = response_json.get("audioContent")

        if audio_content:
            display(Audio(base64.b64decode(audio_content), rate=24000))
        else:
            print("Error: Audio content not found in the response.")
            print(response_json)

    except requests.exceptions.RequestException as e:
        print(f"Error making API request: {e}")
    except json.JSONDecodeError as e:
        print(f"Error decoding JSON response: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

from google.cloud import texttospeech
from google.cloud.texttospeech_v1beta1.services.text_to_speech import client


def perform_voice_cloning(
    voice_cloning_key: str,
    transcript: str,
    language_code: str,
    synthesis_output_path: str,
    tts_client: client.TextToSpeechClient,
) -> None:
  """Perform voice cloning and write output to a file.

  Args:
    voice_cloning_key: The voice cloning key.
    transcript: The transcript to synthesize.
    language_code: The language code.
    synthesis_output_path: The synthesis audio output path.
    tts_client: The TTS client to use.
  """
  voice_clone_params = texttospeech.VoiceCloneParams(
      voice_cloning_key=voice_cloning_key
  )
  voice = texttospeech.VoiceSelectionParams(
      language_code=language_code, voice_clone=voice_clone_params
  )
  request = texttospeech.SynthesizeSpeechRequest(
      input=texttospeech.SynthesisInput(text=transcript),
      voice=voice,
      audio_config=texttospeech.AudioConfig(
          audio_encoding=texttospeech.AudioEncoding.LINEAR16,
          sample_rate_hertz=24000,
      ),
  )
  response = tts_client.synthesize_speech(request)
  with open(synthesis_output_path, 'wb') as out:
    out.write(response.audio_content)
    print(f'Audio content written to file {synthesis_output_path}.')


if __name__ == '__main__':
  client = texttospeech.TextToSpeechClient()
  with open('voice_cloning_key.txt', 'r') as f:
    key = f.read()
  perform_voice_cloning(
      voice_cloning_key=key,
      transcript='Hello world!',
      language_code='en-US',
      synthesis_output_path='/tmp/output.wav',
      tts_client=client,
  )

import io
import wave
from google.cloud import texttospeech
from google.cloud.texttospeech_v1beta1.services.text_to_speech import client


def perform_voice_cloning_with_simulated_streaming(
    voice_cloning_key: str,
    simulated_streamed_text: list[str],
    language_code: str,
    synthesis_output_path: str,
    tts_client: client.TextToSpeechClient,
) -> None:
  """Perform voice cloning for a given reference audio, voice talent consent, and consent script.

  Args:
    voice_cloning_key: The voice cloning key.
    simulated_streamed_text: The list of transcripts to synthesize, where each
      item represents a chunk of streamed text. This is used to simulate
      streamed text input and is not meant to be representative of real-world
      streaming usage.
    language_code: The language code.
    synthesis_output_path: The path to write the synthesis audio output to.
    tts_client: The TTS client to use.
  """
  voice_clone_params = texttospeech.VoiceCloneParams(
      voice_cloning_key=voice_cloning_key
  )
  streaming_config = texttospeech.StreamingSynthesizeConfig(
      voice=texttospeech.VoiceSelectionParams(
          language_code=language_code, voice_clone=voice_clone_params
      ),
      streaming_audio_config=texttospeech.StreamingAudioConfig(
          audio_encoding=texttospeech.AudioEncoding.PCM,
          sample_rate_hertz=24000,
      ),
  )
  config_request = texttospeech.StreamingSynthesizeRequest(
      streaming_config=streaming_config
  )

  # Request generator. Consider using Gemini or another LLM with output
  # streaming as a generator.
  def request_generator():
    yield config_request
    for text in simulated_streamed_text:
      yield texttospeech.StreamingSynthesizeRequest(
          input=texttospeech.StreamingSynthesisInput(text=text)
      )

  streaming_responses = tts_client.streaming_synthesize(request_generator())
  audio_buffer = io.BytesIO()
  for response in streaming_responses:
    print(f'Audio content size in bytes is: {len(response.audio_content)}')
    audio_buffer.write(response.audio_content)

  # Write collected audio outputs to a WAV file.
  with wave.open(synthesis_output_path, 'wb') as wav_file:
    wav_file.setnchannels(1)
    wav_file.setsampwidth(2)
    wav_file.setframerate(24000)
    wav_file.writeframes(audio_buffer.getvalue())
    print(f'Audio content written to file {synthesis_output_path}.')


if __name__ == '__main__':
  client = texttospeech.TextToSpeechClient()
  with open('voice_cloning_key.txt', 'r') as f:
    key = f.read()
  perform_voice_cloning_with_simulated_streaming(
      voice_cloning_key=key,
      simulated_streamed_text=[
          'Hello world!',
          'This is the second text chunk.',
          'This simulates streaming text for synthesis.',
      ],
      language_code='en-US',
      synthesis_output_path='streaming_output.wav',
      tts_client=client,
  )