Create app.py

app.py

@@ -0,0 +1,314 @@
+import io
+import os
+import subprocess
+import time
+import wave
+import numpy as np
+import gradio as gr
+import requests
+import torch
+from functools import lru_cache
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import FileResponse, JSONResponse, Response
+from zipvoice.luxvoice import LuxTTS
+
+from pydantic import BaseModel
+
+GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "")
+
+# Tracked so /health can report uptime and request volume - useful for
+# correlating a slow day with "how long has this container been running"
+# or "how many requests has it handled" without digging through raw logs.
+START_TIME = time.time()
+speak_request_count = 0
+
+
+class ChatBody(BaseModel):
+    messages: list
+
+# Init Model (weights already downloaded at build time)
+device = "cuda" if torch.cuda.is_available() else "cpu"
+lux_tts = LuxTTS("YatharthS/LuxTTS", device=device, threads=2)
+
+
+@lru_cache(maxsize=8)
+def get_encoded_prompt(audio_prompt, ref_duration, rms):
+    # encode_prompt() is deterministic for a given (file, duration, rms)
+    # combo, and me.wav never changes between calls from the chat bot -
+    # so we only need to pay this cost once instead of on every message.
+    return lux_tts.encode_prompt(audio_prompt, duration=ref_duration, rms=rms)
+
+
+# Warm the cache at startup so the very first real request doesn't
+# also have to pay the reference-encoding cost.
+try:
+    get_encoded_prompt("me.wav", 5, 0.01)
+    print("Reference prompt pre-encoded and cached.")
+except Exception as e:
+    print(f"Could not pre-warm reference prompt cache: {e}")
+
+
+def run_tts(
+    text,
+    audio_prompt=None,
+    rms=0.01,
+    ref_duration=5,
+    t_shift=0.9,
+    num_steps=4,
+    speed=0.8,
+    return_smooth=False,
+):
+    # Shared core used by both the Gradio UI (infer) and the /speak
+    # FastAPI route, so there's one code path for actual generation.
+    if not audio_prompt:
+        audio_prompt = "me.wav"
+
+    start_time = time.time()
+
+    encoded_prompt = get_encoded_prompt(audio_prompt, ref_duration, rms)
+
+    final_wav = lux_tts.generate_speech(
+        text,
+        encoded_prompt,
+        num_steps=int(num_steps),
+        t_shift=t_shift,
+        speed=speed,
+        return_smooth=return_smooth,
+    )
+
+    duration = round(time.time() - start_time, 2)
+
+    final_wav = final_wav.cpu().squeeze(0).numpy()
+    final_wav = (np.clip(final_wav, -1.0, 1.0) * 32767).astype(np.int16)
+
+    return final_wav, duration
+
+
+def wav_bytes_from_array(int16_array, sample_rate=48000):
+    # Wraps a raw int16 PCM array in a proper WAV header, in-memory -
+    # no temp file on disk, so nothing to serve back over a second
+    # HTTP round trip.
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)
+        wf.setframerate(sample_rate)
+        wf.writeframes(int16_array.tobytes())
+    return buf.getvalue()
+
+
+def compress_to_aac(wav_bytes, bitrate="40k"):
+    # AAC-in-fragmented-MP4 instead of Opus-in-Ogg: Safari/iOS/WebKit
+    # never added Ogg container support (confirmed still true as of
+    # 2026), so Opus playback silently fails on iPad/iPhone even
+    # though it works fine on Chrome (laptop, Android). AAC in MP4
+    # plays everywhere - Safari, Chrome, Firefox, Android - with one
+    # code path instead of detecting device/browser and branching.
+    # "frag_keyframe+empty_moov+default_base_moof" produces a
+    # streamable fragmented MP4 that doesn't need to seek back to
+    # write a moov atom at the end, since we're writing to a pipe.
+    # Runs entirely via pipes, no temp files.
+    proc = subprocess.run(
+        [
+            "ffmpeg", "-i", "pipe:0",
+            "-c:a", "aac", "-b:a", bitrate, "-vn",
+            "-movflags", "frag_keyframe+empty_moov+default_base_moof",
+            "-f", "mp4", "pipe:1",
+        ],
+        input=wav_bytes,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=True,
+    )
+    return proc.stdout
+
+
+def infer(
+    text,
+    audio_prompt,
+    rms,
+    ref_duration,
+    t_shift,
+    num_steps,
+    speed,
+    return_smooth,
+):
+    if not text:
+        return None, "Please provide text."
+
+    final_wav, duration = run_tts(
+        text, audio_prompt, rms, ref_duration, t_shift, num_steps, speed, return_smooth,
+    )
+
+    stats_msg = f"✨ Generation complete in **{duration}s**."
+    return (48000, final_wav), stats_msg
+
+
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🎙️ Papa Jerry Voice (LuxTTS)")
+
+    gr.Markdown(
+        """
+        > **Note:** This demo runs on a **2-core CPU**, so expect slower inference.  
+        > **Tip:** If words get cut off, lower **Speed** or increase **Ref Duration**.
+        """
+    )
+
+    with gr.Row():
+        with gr.Column():
+            input_text = gr.Textbox(
+                label="Text to Synthesize",
+                value="Hey, what's up? I'm feeling really great!",
+            )
+            input_audio = gr.Audio(
+                label="Reference Audio (.wav)",
+                type="filepath",
+                value="me.wav",
+            )
+
+            with gr.Row():
+                rms_val = gr.Number(value=0.01, label="RMS (Loudness)")
+                ref_duration_val = gr.Number(
+                    value=5,
+                    label="Reference Duration (sec)",
+                    info="Lower = faster. Set ~1000 if you hear artifacts.",
+                )
+                t_shift_val = gr.Number(value=0.9, label="T-Shift")
+
+            with gr.Row():
+                steps_val = gr.Slider(1, 10, value=4, step=1, label="Num Steps")
+                speed_val = gr.Slider(
+                    0.5, 2.0, value=0.8, step=0.1,
+                    label="Speed (Lower = Longer / Clearer)",
+                )
+                smooth_val = gr.Checkbox(label="Return Smooth", value=False)
+
+            btn = gr.Button("Generate Speech", variant="primary")
+
+        with gr.Column():
+            audio_out = gr.Audio(label="Result")
+            status_text = gr.Markdown("Ready to generate...")
+
+    btn.click(
+        fn=infer,
+        inputs=[
+            input_text, input_audio, rms_val, ref_duration_val,
+            t_shift_val, steps_val, speed_val, smooth_val,
+        ],
+        outputs=[audio_out, status_text],
+        api_name="predict",
+    )
+
+app = FastAPI()
+
+@app.get("/")
+def read_index():
+    return FileResponse("index.html")
+
+@app.get("/speak")
+def speak(text: str):
+    # GET + query param (instead of POST + JSON body) so the browser's
+    # <audio> element can set this URL directly as its src and stream
+    # the response progressively, playing as bytes arrive instead of
+    # the page having to fetch() the whole body into a blob first.
+    #
+    # Also compresses to Opus before sending - measured egress out of
+    # this Space is throttled to roughly 20-30 KB/s regardless of
+    # payload size, and raw WAV needs ~96 KB/s to play in real time.
+    # Opus at 32kbps needs ~4 KB/s, so it actually fits the available
+    # bandwidth instead of just being smaller.
+    if not text:
+        raise HTTPException(status_code=400, detail="No text provided")
+
+    global speak_request_count
+    speak_request_count += 1
+    request_start = time.time()
+
+    final_wav, duration = run_tts(text)
+    wav_bytes = wav_bytes_from_array(final_wav)
+
+    try:
+        audio_bytes = compress_to_aac(wav_bytes)
+        media_type = "audio/mp4"
+    except Exception as e:
+        print(f"AAC compression failed, falling back to raw wav: {e}")
+        audio_bytes = wav_bytes
+        media_type = "audio/wav"
+
+    total_seconds = round(time.time() - request_start, 2)
+    uptime_hours = round((time.time() - START_TIME) / 3600, 2)
+    print(
+        f"[/speak] request #{speak_request_count} | "
+        f"generation={duration}s | total={total_seconds}s | "
+        f"wav={len(wav_bytes)}B -> {media_type}={len(audio_bytes)}B | "
+        f"container_uptime={uptime_hours}h"
+    )
+
+    return Response(
+        content=audio_bytes,
+        media_type=media_type,
+        headers={
+            "X-Generation-Seconds": str(duration),
+            # Lets the browser reuse an already-downloaded clip for the
+            # same exact text (e.g. a "Replay Voice" click) instead of
+            # re-requesting and regenerating it from scratch.
+            "Cache-Control": "public, max-age=86400",
+        },
+    )
+
+
+@app.get("/health")
+def health():
+    uptime_seconds = round(time.time() - START_TIME, 1)
+    return JSONResponse({
+        "status": "ok",
+        "device": device,
+        "uptime_seconds": uptime_seconds,
+        "uptime_hours": round(uptime_seconds / 3600, 2),
+        "speak_requests_served": speak_request_count,
+    })
+
+
+@app.get("/{filename}")
+def read_static_file(filename: str):
+    # Serves any other flat file next to index.html/app.py in the repo -
+    # me.jpg, facts.json, etc. This only matches a single path segment
+    # (no slashes), so it won't intercept the /voice/... Gradio routes.
+    # IMPORTANT: this catch-all must stay registered AFTER /speak (and
+    # any other single-segment route) - FastAPI matches routes in
+    # registration order, so an earlier catch-all would shadow it.
+    if os.path.isfile(filename):
+        return FileResponse(filename)
+    raise HTTPException(status_code=404, detail="Not found")
+
+
+@app.post("/chat")
+def chat(body: ChatBody):
+    # Proxies the Groq call server-side so GROQ_API_KEY never has to
+    # live in index.html or be visible via "View Page Source".
+    # Plain (non-async) def so FastAPI runs this blocking network call
+    # in a threadpool instead of freezing the whole server on it.
+    if not GROQ_API_KEY:
+        raise HTTPException(status_code=500, detail="GROQ_API_KEY not configured on server")
+
+    r = requests.post(
+        "https://api.groq.com/openai/v1/chat/completions",
+        headers={
+            "Content-Type": "application/json",
+            "Authorization": f"Bearer {GROQ_API_KEY}",
+        },
+        json={
+            "model": "llama-3.3-70b-versatile",
+            "max_tokens": 300,
+            "messages": body.messages,
+        },
+        timeout=30,
+    )
+    return JSONResponse(content=r.json(), status_code=r.status_code)
+
+
+app = gr.mount_gradio_app(app, demo, path="/voice")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=7860)