Zenn
🎤

google text-to-speechの使い方メモ

に公開
texttospeech
tech

公式ドキュメントはこちらを参考
https://cloud.google.com/text-to-speech/docs/samples?hl=ja

最も基本的な使い方

PCのマイクを使って録音及びスピーカーを使って再生するためのライブラリにはsounddeviceを使っています。
（世の中にはpydubを使うサンプルコードも多いですが、pydubは古いためpython3.12や3.13では依存エラーが出やすいです）

io.BytesIOを使えば、一時ファイルに書き出さずに済みます。

from google.cloud import texttospeech
import sounddevice as sd
import soundfile as sf
import io

client = texttospeech.TextToSpeechClient()

synthesis_input = texttospeech.SynthesisInput(text="はい喜んで、あなた方のため。はい謹んで、あなた方のため。に〜")

voice = texttospeech.VoiceSelectionParams(
    language_code="ja-JP", ssml_gender=texttospeech.SsmlVoiceGender.NEUTRAL
)

audio_config = texttospeech.AudioConfig(
    audio_encoding=texttospeech.AudioEncoding.MP3
)

response = client.synthesize_speech(
    input=synthesis_input, voice=voice, audio_config=audio_config
)


audio_data = response.audio_content

with io.BytesIO(audio_data) as buffer:
    data, samplerate = sf.read(buffer)
    sd.play(data, samplerate)
    sd.wait()

ストリーミングさせながら再生

音声をチャンクに分割し、キューに蓄積させながら変換しています。

from google.cloud import texttospeech
import sounddevice as sd
import soundfile as sf
import io

from queue import Queue
import asyncio

text_buff: Queue[str] = Queue()

for text in ["はい喜んで","あなた方のため", "はい謹んで", "あなた方のため", "に"]:
    text_buff.put(text)

async def generate_text():
    """テキストのリストを非同期的に生成して返すジェネレーター関数。"""
    while not text_buff.empty():
        yield text_buff.get()


client = texttospeech.TextToSpeechClient()

voice = texttospeech.VoiceSelectionParams(
    language_code="ja-JP", ssml_gender=texttospeech.SsmlVoiceGender.MALE
)

audio_config = texttospeech.AudioConfig(
    audio_encoding=texttospeech.AudioEncoding.MP3
)

async def synthesize_speech():
    """テキストを音声に変換し、音声データを非同期的に生成するジェネレーター関数。"""
    async for t in generate_text():
        synthesis_input = texttospeech.SynthesisInput(text=t)

        response = client.synthesize_speech(
            input=synthesis_input, voice=voice, audio_config=audio_config
        )
        audio_data = response.audio_content

        yield audio_data



async def play_audio(audio_data):
    """バイトデータの音声を再生する非同期関数。
    
    Args:
        audio_data: 再生する音声データのバイト列
    """
    with io.BytesIO(audio_data) as buffer:
        data, samplerate = sf.read(buffer)
    sd.play(data, samplerate)
    await asyncio.sleep(3)

async def main():
    """メイン実行関数。音声合成と再生を行う。"""
    async for audio_data in synthesize_speech():
        await play_audio(audio_data)

asyncio.run(main())

StreamingAudioを使う

ジェネレーターで送られてくるテキストを順次音声に変換して再生します。

取り扱いしやすいよう、ユーティリティファイルと実行ファイルの2つに分けました。

  • add_audio_data：テキストをStreamingAudioを使ってオーディオデータに変換します。オーディオデータはキューに順次格納されます

  • streaming_play_audio：上のadd_audio_dataで作成されたオーディオデータをキューから順次取り出し、sounddeviceを使って再生します。

utilities/google_tts

import sounddevice as sd
import numpy as np
from typing import Generator
from google.cloud import texttospeech

# 音声生成クライアント
googletts_client = texttospeech.TextToSpeechClient()

# See https://cloud.google.com/text-to-speech/docs/voices for all voices.
voice = texttospeech.VoiceSelectionParams(
    language_code="ja-JP",
    name="ja-JP-Chirp3-HD-Aoede"
)

streaming_audio_config = texttospeech.StreamingAudioConfig(
    audio_encoding=texttospeech.AudioEncoding.PCM, # ストリーミングはPCMとOGG_OPUSのみサポート
    sample_rate_hertz=24000
)

streaming_config = texttospeech.StreamingSynthesizeConfig(
    voice=voice,
    streaming_audio_config=streaming_audio_config
)

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(generator: Generator[str, None, None]):
    for text in generator:
        yield texttospeech.StreamingSynthesizeRequest(input=texttospeech.StreamingSynthesisInput(text=text))


def play_audio(audio_data: bytes):
    """google ttsの音声データを再生する"""
    audio_array = np.frombuffer(audio_data, dtype=np.int16)
    audio_normalized = audio_array / 32768.0
    sd.play(audio_normalized, 24000)
    sd.wait()

sample.py

from queue import Queue
import threading
import time
import itertools

from utilities.google_tts import googletts_client, config_request, request_generator, play_audio


audio_buffer: Queue[bytes] = Queue() # 音声データバッファ
generation_done = threading.Event() # 音声生成完了フラグ

textset = """吾輩は猫である。名前はまだ無い。
どこで生れたかとんと見当がつかぬ。何でも薄暗いじめじめした所でニャーニャー泣いていた事だけは記憶している。
吾輩はここで始めて人間というものを見た。しかもあとで聞くとそれは書生という人間中で一番獰悪な種族であったそうだ。
この書生というのは時々我々を捕つかまえて煮にて食うという話である。
しかしその当時は何という考もなかったから別段恐しいとも思わなかった。
ただ彼の掌に載せられてスーと持ち上げられた時何だかフワフワした感じがあったばかりである。
"""

def text_chunk_generator():
    # テキストを2〜5文字または「。」で区切る
    chunks = []
    start_index = 0
    count = 1

    while True:
        chunk = textset[start_index:start_index+count]
        if chunk.strip() and chunk != "。":
            chunks.append(chunk.replace("。", ""))
        start_index += count
        count = 1 if count == 5 else count + 1
        if start_index >= len(textset):
            break
    
    for text in chunks:
        print(text)
        yield text
        time.sleep(0.3)



def add_audio_data():
    streaming_responses = googletts_client.streaming_synthesize(
        itertools.chain([config_request], request_generator(text_chunk_generator()))
    )
    for response in streaming_responses:
        audio_data = response.audio_content
        audio_buffer.put(audio_data)

    # 音声生成完了を通知
    generation_done.set()
    print("音声生成完了")

def streaming_play_audio():
    audio_data = bytearray()
    
    while not generation_done.is_set() or not audio_buffer.empty():
        if not audio_buffer.empty():
            chunk = audio_buffer.get()
            audio_data.extend(chunk)
        else:
            # バッファが空で、かつデータが溜まっている場合は再生
            if len(audio_data) > 0:
                try:
                    play_audio(audio_data)
                    # 再生したデータをクリア
                    audio_data = bytearray()
                except Exception as e:
                    print(f"再生エラー: {e}")
    
    # 最後に残ったデータを再生
    if len(audio_data) > 0:
        try:
            play_audio(audio_data)
            print("すべての再生完了")
        except Exception as e:
            print(f"最終再生エラー: {e}")


if __name__ == "__main__":
    # 音声生成スレッドを開始
    threading.Thread(target=add_audio_data).start()
    # 再生スレッドを開始
    threading.Thread(target=streaming_play_audio).start()

Discussion