"""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"""
Urdu S2S
Urdu Speech-to-Speech
Upload a WAV prompt. The live GPU mode returns an Urdu assistant reply synthesized with the Praxy voice anchor.
Ready.
"""
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,
}
@app.get("/", response_class=HTMLResponse)
def demo() -> str:
return build_demo_html(default_mode=default_mode)
@app.get("/health")
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,
}
@app.get("/baseline/{benchmark_id}")
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)
@app.get("/baseline/{benchmark_id}/audio")
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,
)
@app.get("/s2s/{request_id}/audio")
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,
)
@app.post("/s2s")
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