Spaces:
Running on Zero
Running on Zero
| """FastAPI wrapper for the Urdu S2S live MVP.""" | |
| from __future__ import annotations | |
| from collections.abc import Callable | |
| from io import BytesIO | |
| import html | |
| from pathlib import Path | |
| import shutil | |
| from typing import Any | |
| from uuid import uuid4 | |
| from .asr_providers import FasterWhisperASRProvider | |
| from .live_providers import OpenAIBridgeProvider, OpenAICompatibleChatClient, OpenAIReplyProvider | |
| from .pipeline import SpeechToSpeechPipeline | |
| from .providers import ( | |
| CandidateCsvLookup, | |
| LookupASRProvider, | |
| LookupBridgeProvider, | |
| LookupReplyProvider, | |
| LookupTTSProvider, | |
| resolve_path, | |
| ) | |
| from .schemas import ASRResult, BridgeResult, ReplyResult, SpeechToSpeechRequest, SpeechToSpeechResult, TTSResult | |
| from .tts_providers import ChatterboxPraxyTTSProvider | |
| from .tracing import to_jsonable | |
| ROOT = Path(__file__).resolve().parents[2] | |
| DEFAULT_BASELINE_CSV = ( | |
| ROOT / "reports/evals/urdu_s2s_reference.csv" | |
| ) | |
| DEFAULT_UPLOAD_DIR = ROOT / "artifacts/api_uploads" | |
| DEFAULT_TTS_OUTPUT_DIR = ROOT / "artifacts/api_tts_outputs" | |
| DEFAULT_PRAXY_ANCHOR = ROOT / "data/processed/voice_anchors/chatterbox_praxy_v1/bench_025.wav" | |
| def build_demo_html(*, default_mode: str) -> str: | |
| escaped_mode = html.escape(default_mode) | |
| return f"""<!doctype html> | |
| <html lang="en"> | |
| <head> | |
| <meta charset="utf-8" /> | |
| <meta name="viewport" content="width=device-width, initial-scale=1" /> | |
| <title>Urdu S2S</title> | |
| <style> | |
| :root {{ | |
| color-scheme: dark; | |
| font-family: Inter, ui-sans-serif, system-ui, -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif; | |
| background: #101114; | |
| color: #f4f5f7; | |
| }} | |
| body {{ margin: 0; min-height: 100vh; display: grid; place-items: center; }} | |
| main {{ width: min(820px, calc(100vw - 32px)); }} | |
| h1 {{ font-size: 28px; margin: 0 0 8px; letter-spacing: 0; }} | |
| p {{ color: #b7bdc8; line-height: 1.5; }} | |
| form, pre {{ border: 1px solid #2b3038; border-radius: 8px; background: #171a20; }} | |
| form {{ display: grid; gap: 14px; padding: 18px; }} | |
| label {{ display: grid; gap: 8px; color: #d8dce3; }} | |
| input, select, button {{ | |
| font: inherit; | |
| border-radius: 6px; | |
| border: 1px solid #343b46; | |
| background: #0f1115; | |
| color: #f4f5f7; | |
| padding: 10px 12px; | |
| }} | |
| button {{ cursor: pointer; background: #f4f5f7; color: #101114; font-weight: 700; }} | |
| button:disabled {{ opacity: .65; cursor: progress; }} | |
| audio {{ width: 100%; margin-top: 14px; }} | |
| pre {{ padding: 14px; overflow: auto; white-space: pre-wrap; }} | |
| </style> | |
| </head> | |
| <body> | |
| <main> | |
| <h1>Urdu Speech-to-Speech</h1> | |
| <p>Upload a WAV prompt. The live GPU mode returns an Urdu assistant reply synthesized with the Praxy voice anchor.</p> | |
| <form id="s2s-form"> | |
| <label>Audio file <input id="audio" name="audio" type="file" accept="audio/*" required /></label> | |
| <label>Mode | |
| <select id="mode" name="mode"> | |
| <option value="{escaped_mode}">{escaped_mode}</option> | |
| <option value="live_tts">live_tts</option> | |
| <option value="text_only">text_only</option> | |
| </select> | |
| </label> | |
| <label>Roman Urdu hint <input id="prompt" name="prompt_roman_urdu" placeholder="optional" /></label> | |
| <button id="submit" type="submit">Generate Response</button> | |
| </form> | |
| <audio id="player" controls hidden></audio> | |
| <pre id="result">Ready.</pre> | |
| </main> | |
| <script> | |
| const form = document.getElementById("s2s-form"); | |
| const result = document.getElementById("result"); | |
| const button = document.getElementById("submit"); | |
| const player = document.getElementById("player"); | |
| form.addEventListener("submit", async (event) => {{ | |
| event.preventDefault(); | |
| button.disabled = true; | |
| result.textContent = "Generating response..."; | |
| player.hidden = true; | |
| const data = new FormData(); | |
| data.append("audio", document.getElementById("audio").files[0]); | |
| const params = new URLSearchParams({{ | |
| mode: document.getElementById("mode").value, | |
| prompt_roman_urdu: document.getElementById("prompt").value, | |
| }}); | |
| try {{ | |
| const response = await fetch(`/s2s?${{params.toString()}}`, {{ method: "POST", body: data }}); | |
| const payload = await response.json(); | |
| result.textContent = JSON.stringify(payload, null, 2); | |
| if (response.ok && payload.tts_audio_url) {{ | |
| player.src = payload.tts_audio_url; | |
| player.hidden = false; | |
| }} | |
| }} catch (error) {{ | |
| result.textContent = String(error); | |
| }} finally {{ | |
| button.disabled = false; | |
| }} | |
| }}); | |
| </script> | |
| </body> | |
| </html> | |
| """ | |
| class PlaceholderTTSProvider: | |
| """TTS adapter for text-only live mode.""" | |
| provider = "placeholder_tts" | |
| model = "not_synthesized_text_only" | |
| def synthesize( | |
| self, | |
| request: SpeechToSpeechRequest, | |
| reply: ReplyResult, | |
| bridge: BridgeResult, | |
| ) -> TTSResult: | |
| return TTSResult( | |
| audio_path=Path(""), | |
| provider=self.provider, | |
| model=self.model, | |
| voice="", | |
| duration_seconds=None, | |
| ) | |
| def build_lookup_pipeline(candidate_csv: Path) -> tuple[SpeechToSpeechPipeline, CandidateCsvLookup]: | |
| lookup = CandidateCsvLookup(candidate_csv) | |
| return ( | |
| SpeechToSpeechPipeline( | |
| asr_provider=LookupASRProvider(lookup), | |
| reply_provider=LookupReplyProvider(lookup), | |
| bridge_provider=LookupBridgeProvider(lookup), | |
| tts_provider=LookupTTSProvider(lookup), | |
| ), | |
| lookup, | |
| ) | |
| def run_baseline_lookup(candidate_csv: Path, benchmark_id: str) -> SpeechToSpeechResult: | |
| pipeline, lookup = build_lookup_pipeline(candidate_csv) | |
| row = lookup.get(benchmark_id) | |
| return pipeline.run( | |
| SpeechToSpeechRequest( | |
| request_id=benchmark_id, | |
| audio_path=Path(row.get("input_audio_path", "")), | |
| metadata={"candidate_csv": str(candidate_csv)}, | |
| ) | |
| ) | |
| def build_live_text_only_pipeline( | |
| *, | |
| chat_client: OpenAICompatibleChatClient | None = None, | |
| whisper_model_factory: Any | None = None, | |
| whisper_model: str = "large-v3", | |
| whisper_language: str = "ur", | |
| whisper_device: str = "cpu", | |
| whisper_compute_type: str = "int8", | |
| ) -> SpeechToSpeechPipeline: | |
| chat_client = chat_client or OpenAICompatibleChatClient() | |
| return SpeechToSpeechPipeline( | |
| asr_provider=FasterWhisperASRProvider( | |
| model_name=whisper_model, | |
| language=whisper_language, | |
| device=whisper_device, | |
| compute_type=whisper_compute_type, | |
| model_factory=whisper_model_factory, | |
| ), | |
| reply_provider=OpenAIReplyProvider(chat_client=chat_client), | |
| bridge_provider=OpenAIBridgeProvider(chat_client=chat_client), | |
| tts_provider=PlaceholderTTSProvider(), | |
| ) | |
| def build_live_tts_pipeline( | |
| *, | |
| output_audio_path: Path, | |
| voice_prompt_audio_path: Path = DEFAULT_PRAXY_ANCHOR, | |
| chat_client: OpenAICompatibleChatClient | None = None, | |
| whisper_model_factory: Any | None = None, | |
| whisper_model: str = "large-v3", | |
| whisper_language: str = "ur", | |
| whisper_device: str = "cpu", | |
| whisper_compute_type: str = "int8", | |
| chatterbox_device: str = "cuda", | |
| chatterbox_t3_model: str = "v3", | |
| chatterbox_model_loader: Any | None = None, | |
| chatterbox_wav_writer: Any | None = None, | |
| chatterbox_duration_reader: Any | None = None, | |
| ) -> SpeechToSpeechPipeline: | |
| chat_client = chat_client or OpenAICompatibleChatClient() | |
| return SpeechToSpeechPipeline( | |
| asr_provider=FasterWhisperASRProvider( | |
| model_name=whisper_model, | |
| language=whisper_language, | |
| device=whisper_device, | |
| compute_type=whisper_compute_type, | |
| model_factory=whisper_model_factory, | |
| ), | |
| reply_provider=OpenAIReplyProvider(chat_client=chat_client), | |
| bridge_provider=OpenAIBridgeProvider(chat_client=chat_client), | |
| tts_provider=ChatterboxPraxyTTSProvider( | |
| output_audio_path=output_audio_path, | |
| voice_prompt_audio_path=voice_prompt_audio_path, | |
| device=chatterbox_device, | |
| t3_model=chatterbox_t3_model, | |
| model_loader=chatterbox_model_loader, | |
| wav_writer=chatterbox_wav_writer, | |
| duration_reader=chatterbox_duration_reader, | |
| ), | |
| ) | |
| def run_live_text_only( | |
| *, | |
| audio_path: Path, | |
| request_id: str, | |
| prompt_roman_urdu: str = "", | |
| chat_client: OpenAICompatibleChatClient | None = None, | |
| whisper_model_factory: Any | None = None, | |
| whisper_model: str = "large-v3", | |
| whisper_language: str = "ur", | |
| whisper_device: str = "cpu", | |
| whisper_compute_type: str = "int8", | |
| ) -> SpeechToSpeechResult: | |
| pipeline = build_live_text_only_pipeline( | |
| chat_client=chat_client, | |
| whisper_model_factory=whisper_model_factory, | |
| whisper_model=whisper_model, | |
| whisper_language=whisper_language, | |
| whisper_device=whisper_device, | |
| whisper_compute_type=whisper_compute_type, | |
| ) | |
| return pipeline.run( | |
| SpeechToSpeechRequest( | |
| request_id=request_id, | |
| audio_path=audio_path, | |
| metadata={"prompt_roman_urdu": prompt_roman_urdu}, | |
| ) | |
| ) | |
| def run_live_tts( | |
| *, | |
| audio_path: Path, | |
| request_id: str, | |
| output_audio_path: Path, | |
| prompt_roman_urdu: str = "", | |
| voice_prompt_audio_path: Path = DEFAULT_PRAXY_ANCHOR, | |
| chat_client: OpenAICompatibleChatClient | None = None, | |
| whisper_model_factory: Any | None = None, | |
| whisper_model: str = "large-v3", | |
| whisper_language: str = "ur", | |
| whisper_device: str = "cpu", | |
| whisper_compute_type: str = "int8", | |
| chatterbox_device: str = "cuda", | |
| chatterbox_t3_model: str = "v3", | |
| chatterbox_model_loader: Any | None = None, | |
| chatterbox_wav_writer: Any | None = None, | |
| chatterbox_duration_reader: Any | None = None, | |
| ) -> SpeechToSpeechResult: | |
| pipeline = build_live_tts_pipeline( | |
| output_audio_path=output_audio_path, | |
| voice_prompt_audio_path=voice_prompt_audio_path, | |
| chat_client=chat_client, | |
| whisper_model_factory=whisper_model_factory, | |
| whisper_model=whisper_model, | |
| whisper_language=whisper_language, | |
| whisper_device=whisper_device, | |
| whisper_compute_type=whisper_compute_type, | |
| chatterbox_device=chatterbox_device, | |
| chatterbox_t3_model=chatterbox_t3_model, | |
| chatterbox_model_loader=chatterbox_model_loader, | |
| chatterbox_wav_writer=chatterbox_wav_writer, | |
| chatterbox_duration_reader=chatterbox_duration_reader, | |
| ) | |
| return pipeline.run( | |
| SpeechToSpeechRequest( | |
| request_id=request_id, | |
| audio_path=audio_path, | |
| metadata={"prompt_roman_urdu": prompt_roman_urdu}, | |
| ) | |
| ) | |
| def result_to_api_payload(result: SpeechToSpeechResult, *, repo_root: Path = ROOT) -> dict[str, Any]: | |
| tts_audio_path = result.tts.audio_path | |
| tts_audio_path_text = str(tts_audio_path) | |
| has_tts_audio_path = tts_audio_path_text not in {"", "."} | |
| resolved_tts_audio_path = resolve_path(repo_root, tts_audio_path) if has_tts_audio_path else None | |
| tts_audio_exists = bool(resolved_tts_audio_path and resolved_tts_audio_path.exists()) | |
| return { | |
| "request_id": result.request.request_id, | |
| "input_audio_path": str(result.request.audio_path), | |
| "asr_transcript": result.asr.text, | |
| "assistant_reply_urdu": result.reply.text_urdu, | |
| "devanagari_tts_text": result.bridge.text_devanagari, | |
| "tts_audio_path": tts_audio_path_text if has_tts_audio_path else "", | |
| "tts_audio_exists": tts_audio_exists, | |
| "tts_audio_url": ( | |
| f"/baseline/{result.request.request_id}/audio" | |
| if tts_audio_exists | |
| else "" | |
| ), | |
| "providers": { | |
| "asr": {"provider": result.asr.provider, "model": result.asr.model}, | |
| "reply": {"provider": result.reply.provider, "model": result.reply.model}, | |
| "bridge": {"provider": result.bridge.provider, "model": result.bridge.model}, | |
| "tts": { | |
| "provider": result.tts.provider, | |
| "model": result.tts.model, | |
| "voice": result.tts.voice, | |
| }, | |
| }, | |
| "duration_seconds": result.tts.duration_seconds, | |
| "trace": to_jsonable(result.trace), | |
| } | |
| def save_upload_stream( | |
| source: Any, | |
| *, | |
| upload_dir: Path, | |
| filename: str, | |
| request_id: str, | |
| ) -> Path: | |
| suffix = Path(filename or "").suffix or ".wav" | |
| safe_path = upload_dir / f"{request_id}{suffix}" | |
| upload_dir.mkdir(parents=True, exist_ok=True) | |
| with safe_path.open("wb") as handle: | |
| shutil.copyfileobj(source, handle) | |
| return safe_path | |
| def create_app( | |
| *, | |
| candidate_csv: Path = DEFAULT_BASELINE_CSV, | |
| repo_root: Path = ROOT, | |
| upload_dir: Path = DEFAULT_UPLOAD_DIR, | |
| tts_output_dir: Path = DEFAULT_TTS_OUTPUT_DIR, | |
| default_mode: str = "text_only", | |
| default_whisper_model: str = "large-v3", | |
| default_whisper_language: str = "ur", | |
| default_whisper_device: str = "cpu", | |
| default_whisper_compute_type: str = "int8", | |
| default_voice_prompt_audio_path: Path = DEFAULT_PRAXY_ANCHOR, | |
| default_chatterbox_device: str = "cuda", | |
| default_chatterbox_t3_model: str = "v3", | |
| live_text_runner: Callable[..., SpeechToSpeechResult] = run_live_text_only, | |
| live_tts_runner: Callable[..., SpeechToSpeechResult] = run_live_tts, | |
| ): | |
| try: | |
| from fastapi import FastAPI, File, HTTPException | |
| from fastapi.responses import FileResponse, HTMLResponse | |
| except ModuleNotFoundError as exc: # pragma: no cover - exercised only without API deps. | |
| raise RuntimeError( | |
| "FastAPI API dependencies are missing. Install with: " | |
| "python3 -m pip install -r requirements-api.txt" | |
| ) from exc | |
| app = FastAPI( | |
| title="Urdu S2S API", | |
| version="0.1.0", | |
| description="Deployable Urdu speech-to-speech service with live Praxy/Chatterbox audio output.", | |
| ) | |
| runtime_defaults = { | |
| "mode": default_mode, | |
| "whisper_model": default_whisper_model, | |
| "whisper_language": default_whisper_language, | |
| "whisper_device": default_whisper_device, | |
| "whisper_compute_type": default_whisper_compute_type, | |
| "voice_prompt_audio_path": str(default_voice_prompt_audio_path), | |
| "voice_prompt_audio_exists": default_voice_prompt_audio_path.exists(), | |
| "chatterbox_device": default_chatterbox_device, | |
| "chatterbox_t3_model": default_chatterbox_t3_model, | |
| } | |
| def demo() -> str: | |
| return build_demo_html(default_mode=default_mode) | |
| def health() -> dict[str, object]: | |
| return { | |
| "ok": True, | |
| "runtime": "urdu_s2s_mvp", | |
| "candidate_csv": str(candidate_csv), | |
| "candidate_csv_exists": candidate_csv.exists(), | |
| "runtime_defaults": runtime_defaults, | |
| } | |
| def baseline(benchmark_id: str) -> dict[str, Any]: | |
| try: | |
| result = run_baseline_lookup(candidate_csv, benchmark_id) | |
| except KeyError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| return result_to_api_payload(result, repo_root=repo_root) | |
| def baseline_audio(benchmark_id: str): | |
| try: | |
| result = run_baseline_lookup(candidate_csv, benchmark_id) | |
| except KeyError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| audio_path = resolve_path(repo_root, result.tts.audio_path) | |
| if not audio_path.exists(): | |
| raise HTTPException(status_code=404, detail=f"Audio file not found: {audio_path}") | |
| return FileResponse( | |
| audio_path, | |
| media_type="audio/wav", | |
| filename=audio_path.name, | |
| ) | |
| def live_tts_audio(request_id: str): | |
| audio_path = tts_output_dir / f"{request_id}.wav" | |
| if not audio_path.exists(): | |
| raise HTTPException(status_code=404, detail=f"Audio file not found: {audio_path}") | |
| return FileResponse( | |
| audio_path, | |
| media_type="audio/wav", | |
| filename=audio_path.name, | |
| ) | |
| async def s2s( | |
| audio: bytes = File(...), | |
| mode: str = default_mode, | |
| prompt_roman_urdu: str = "", | |
| request_id: str = "", | |
| whisper_model: str = default_whisper_model, | |
| whisper_language: str = default_whisper_language, | |
| whisper_device: str = default_whisper_device, | |
| whisper_compute_type: str = default_whisper_compute_type, | |
| voice_prompt_audio_path: str = str(default_voice_prompt_audio_path), | |
| chatterbox_device: str = default_chatterbox_device, | |
| chatterbox_t3_model: str = default_chatterbox_t3_model, | |
| ) -> dict[str, Any]: | |
| if mode not in {"text_only", "live_tts"}: | |
| raise HTTPException( | |
| status_code=400, | |
| detail="Supported modes: text_only, live_tts.", | |
| ) | |
| resolved_request_id = request_id.strip() or f"s2s_{uuid4().hex}" | |
| saved_audio_path = save_upload_stream( | |
| BytesIO(audio), | |
| upload_dir=upload_dir, | |
| filename="upload.wav", | |
| request_id=resolved_request_id, | |
| ) | |
| try: | |
| if mode == "text_only": | |
| result = live_text_runner( | |
| audio_path=saved_audio_path, | |
| request_id=resolved_request_id, | |
| prompt_roman_urdu=prompt_roman_urdu, | |
| whisper_model=whisper_model, | |
| whisper_language=whisper_language, | |
| whisper_device=whisper_device, | |
| whisper_compute_type=whisper_compute_type, | |
| ) | |
| else: | |
| output_audio_path = tts_output_dir / f"{resolved_request_id}.wav" | |
| result = live_tts_runner( | |
| audio_path=saved_audio_path, | |
| request_id=resolved_request_id, | |
| output_audio_path=output_audio_path, | |
| prompt_roman_urdu=prompt_roman_urdu, | |
| voice_prompt_audio_path=Path(voice_prompt_audio_path), | |
| whisper_model=whisper_model, | |
| whisper_language=whisper_language, | |
| whisper_device=whisper_device, | |
| whisper_compute_type=whisper_compute_type, | |
| chatterbox_device=chatterbox_device, | |
| chatterbox_t3_model=chatterbox_t3_model, | |
| ) | |
| except Exception as exc: # noqa: BLE001 - API should expose provider failures as 502. | |
| raise HTTPException(status_code=502, detail=str(exc)) from exc | |
| payload = result_to_api_payload(result, repo_root=repo_root) | |
| if mode == "live_tts" and payload["tts_audio_exists"]: | |
| payload["tts_audio_url"] = f"/s2s/{result.request.request_id}/audio" | |
| payload["mode"] = mode | |
| payload["uploaded_audio_path"] = str(saved_audio_path) | |
| return payload | |
| return app | |
| try: | |
| app = create_app() | |
| except RuntimeError: | |
| app = None | |