File size: 5,235 Bytes
c8ee307
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#!/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.