wakeforge / src /backends /gcp.py
eoinedge's picture
Upload folder using huggingface_hub
5838d6a verified
Raw
History Blame Contribute Delete
4.27 kB
"""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)