| """Google Cloud Text-to-Speech backend using an API key (REST API). |
| |
| Uses the public REST endpoints so the only requirement is an API key, |
| which is the simplest credential to provide inside a Hugging Face Space |
| (store it as a Space secret named ``GCP_TTS_API_KEY``): |
| |
| * ``GET /v1/voices`` — list available voices |
| * ``POST /v1/text:synthesize`` — synthesize speech |
| |
| Requesting ``LINEAR16`` at 16 kHz returns ready-to-use PCM WAV bytes. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import base64 |
| from typing import List, Optional |
|
|
| import requests |
|
|
| from ..audio import read_wav_bytes |
| from .base import SynthesisResult, TTSBackend, Voice |
|
|
| _BASE_URL = "https://texttospeech.googleapis.com/v1" |
| _TIMEOUT = 30 |
|
|
|
|
| class GCPTTSBackend(TTSBackend): |
| source = "google_cloud_tts" |
|
|
| def __init__( |
| self, |
| api_key: str, |
| language_prefixes: List[str], |
| max_voices_per_locale: int, |
| sample_rate_hz: int, |
| prefer_voice_types: Optional[List[str]] = None, |
| ) -> None: |
| self.api_key = api_key |
| self.language_prefixes = [p.lower() for p in language_prefixes] |
| self.max_voices_per_locale = max_voices_per_locale |
| self.sample_rate_hz = sample_rate_hz |
| self.prefer_voice_types = prefer_voice_types or [ |
| "chirp", |
| "neural", |
| "wavenet", |
| "studio", |
| "journey", |
| "standard", |
| ] |
| self._voices: List[Voice] = [] |
|
|
| |
|
|
| def prepare(self) -> None: |
| response = requests.get( |
| f"{_BASE_URL}/voices", |
| params={"key": self.api_key}, |
| timeout=_TIMEOUT, |
| ) |
| if response.status_code != 200: |
| raise RuntimeError( |
| f"Google TTS voices request failed ({response.status_code}): " |
| f"{response.text[:200]}" |
| ) |
|
|
| raw_voices = response.json().get("voices", []) |
| by_locale: dict[str, List[Voice]] = {} |
| for item in raw_voices: |
| name = item.get("name", "") |
| for locale in item.get("languageCodes", []): |
| prefix = locale.split("-")[0].lower() |
| if prefix not in self.language_prefixes: |
| continue |
| by_locale.setdefault(locale, []).append( |
| Voice(name=name, language_code=locale, description="google_cloud_tts") |
| ) |
|
|
| def rank(voice: Voice) -> tuple[int, str]: |
| lower = voice.name.lower() |
| for idx, preferred in enumerate(self.prefer_voice_types): |
| if preferred in lower: |
| return idx, voice.name |
| return 999, voice.name |
|
|
| selected: List[Voice] = [] |
| for locale in sorted(by_locale): |
| ordered = sorted(by_locale[locale], key=rank) |
| selected.extend(ordered[: self.max_voices_per_locale]) |
|
|
| if not selected: |
| raise RuntimeError("Google TTS returned no voices matching the requested locales.") |
| self._voices = selected |
|
|
| def voices(self) -> List[Voice]: |
| return list(self._voices) |
|
|
| def synthesize(self, text: str, voice: Voice) -> SynthesisResult: |
| payload = { |
| "input": {"text": text}, |
| "voice": { |
| "languageCode": voice.language_code, |
| "name": voice.name, |
| }, |
| "audioConfig": { |
| "audioEncoding": "LINEAR16", |
| "sampleRateHertz": self.sample_rate_hz, |
| }, |
| } |
| response = requests.post( |
| f"{_BASE_URL}/text:synthesize", |
| params={"key": self.api_key}, |
| json=payload, |
| timeout=_TIMEOUT, |
| ) |
| if response.status_code != 200: |
| raise RuntimeError( |
| f"Google TTS synthesize failed ({response.status_code}): " |
| f"{response.text[:200]}" |
| ) |
|
|
| audio_b64 = response.json().get("audioContent") |
| if not audio_b64: |
| raise RuntimeError("Google TTS response contained no audioContent.") |
|
|
| audio, sample_rate = read_wav_bytes(base64.b64decode(audio_b64)) |
| return SynthesisResult(audio=audio, sample_rate_hz=sample_rate) |
|
|
