en-stt-tts / client_example.py
StrangerInAStrangeLand's picture
Upload client_example.py with huggingface_hub
c8ee307 verified
Raw History Blame Contribute Delete
5.24 kB
#!/usr/bin/env python3
"""Minimal, dependency-light client for the English STT + TTS Space.
Calls the two stable named endpoints:
POST/GET /gradio_api/call/transcribe (api_name="/transcribe")
POST/GET /gradio_api/call/speak (api_name="/speak")
gradio_client hides the two-step enqueue->stream protocol, so use it unless you need to
speak HTTP directly (the raw form is documented at the bottom of this file).
Environment (never hardcode credentials):
EN_SPEECH_SPACE Space id "user/space" or a full https URL. REQUIRED.
HF_TOKEN Optional Hugging Face token. Sending one matters: an anonymous call
draws the shared anonymous ZeroGPU allowance (~2 min/day) instead of
your account quota, and fails with "ZeroGPU quota exceeded" on an
HTTP 200 response. If unset, gradio_client falls back to the token
saved by `hf auth login` (~/.cache/huggingface/token).
Examples:
EN_SPEECH_SPACE=user/en-stt-tts HF_TOKEN=hf_xxx \
python client_example.py audio speech.wav
EN_SPEECH_SPACE=user/en-stt-tts HF_TOKEN=hf_xxx \
python client_example.py tts "The quick brown fox." --voice bf_emma --speed 1.1
"""
from __future__ import annotations
import argparse
import os
import sys
from pathlib import Path
def get_client():
"""Build an authenticated gradio_client for the Space named in EN_SPEECH_SPACE."""
from gradio_client import Client
space = (os.environ.get("EN_SPEECH_SPACE") or "").strip()
if not space:
sys.exit("Set EN_SPEECH_SPACE to the Space id (e.g. user/en-stt-tts) or its URL.")
# token=None makes gradio_client use the locally saved HF token, if any.
token = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
return Client(space, token=token, verbose=False)
def transcribe(client, audio_path: str, max_seconds: float = 300.0):
"""Upload an audio file, return (transcript, status)."""
from gradio_client import handle_file
transcript, status = client.predict(
handle_file(audio_path), # -> first positional input (gr.Audio, filepath)
max_seconds, # -> second positional input (gr.Slider)
api_name="/transcribe",
)
return transcript, status
def speak(client, text: str, voice: str = "af_heart", speed: float = 1.0,
max_chars: int = 5000):
"""Synthesise text, return (local_wav_path, status).
gradio_client downloads the generated WAV to a local temp dir and returns its path.
"""
wav_path, status = client.predict(
text, # gr.Textbox
voice, # gr.Dropdown (value, e.g. "af_heart")
speed, # gr.Slider (0.5 - 2.0)
max_chars, # gr.Slider (maximum text length)
api_name="/speak",
)
return wav_path, status
def main() -> int:
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
p_audio = sub.add_parser("audio", help="transcribe an audio file")
p_audio.add_argument("file")
p_audio.add_argument("--max-seconds", type=float, default=300.0)
p_tts = sub.add_parser("tts", help="synthesise speech from text")
p_tts.add_argument("text")
p_tts.add_argument("--voice", default="af_heart")
p_tts.add_argument("--speed", type=float, default=1.0)
p_tts.add_argument("--max-chars", type=int, default=5000)
p_tts.add_argument("--out", default=None, help="copy the WAV here (optional)")
args = ap.parse_args()
client = get_client()
if args.cmd == "audio":
path = Path(args.file)
if not path.is_file():
sys.exit(f"No such file: {path}")
transcript, status = transcribe(client, str(path), args.max_seconds)
print(f"status : {status}")
print(f"transcript : {transcript!r}")
return 0 if transcript else 1
wav_path, status = speak(client, args.text, args.voice, args.speed, args.max_chars)
print(f"status : {status}")
print(f"wav : {wav_path}")
if args.out:
Path(args.out).write_bytes(Path(wav_path).read_bytes())
print(f"copied : {args.out}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
# ---------------------------------------------------------------------------------------
# Raw HTTP, no gradio_client (for reference — the enqueue + SSE dance):
#
# import json, os, requests
# base = "https://<user>-<space>.hf.space"
# hdr = {"Authorization": f"Bearer {os.environ['HF_TOKEN']}"}
# r = requests.post(f"{base}/gradio_api/call/transcribe",
# json={"data": ["<local path or FileData>", 300]}, headers=hdr)
# event_id = r.json()["event_id"]
# for line in requests.get(f"{base}/gradio_api/call/transcribe/{event_id}",
# headers=hdr, stream=True).iter_lines():
# ... # lines look like: b'data: ["transcript", "status"]'
#
# A refusal (quota, queue, auth) arrives as an *error envelope inside the SSE payload*
# with HTTP 200 — check for an "error" key before trusting the result.