import gradio as gr
import modal
import base64
import random
import json
import uuid
from pathlib import Path
from typing import Optional
from examples import SAMPLE_SENTENCES
APP_NAME = "arabic-tts-arena"
LEADERBOARD_FILE = Path(__file__).parent / "leaderboard.json"
MAX_SYNTHESIS_RETRIES = 2 # per-model retry cap before giving up
MIN_BATTLES = 45 # minimum battles for a model to appear on the leaderboard (to avoid unjust rankings for new models with few votes)
_AVAILABLE_MODELS_CACHE: dict[str, dict[str, str]] | None = None
def _fetch_model_registry() -> dict[str, dict[str, str]]:
"""Fetch the model registry from the Modal backend.
Returns dict like:
{"chatterbox": {"class_name": "ChatterboxModel", "display_name": "Chatterbox"}, ...}
"""
service = modal.Cls.from_name(APP_NAME, "ArenaService")
registry = service().get_model_registry.remote()
if registry:
print(f"✅ Fetched {len(registry)} models from Modal backend")
return registry
raise RuntimeError("Failed to fetch model registry from Modal backend")
def _get_available_models() -> dict[str, dict[str, str]]:
"""Lazy-load the model registry on first use (avoids crashing at import time)."""
global _AVAILABLE_MODELS_CACHE
if _AVAILABLE_MODELS_CACHE is None:
print("⏳ Fetching model registry from Modal backend...")
_AVAILABLE_MODELS_CACHE = _fetch_model_registry()
print(f"✅ Available models: {', '.join(_AVAILABLE_MODELS_CACHE.keys())}")
return _AVAILABLE_MODELS_CACHE
def get_model_cls(model_id: str):
"""Get a Modal class by model_id using the registered class name."""
available = _get_available_models()
if model_id not in available:
raise ValueError(f"Model not available: {model_id}")
class_name = available[model_id]["class_name"]
return modal.Cls.from_name(APP_NAME, class_name)
def get_display_name(model_id: str) -> str:
"""Get the human-readable display name for a model."""
available = _get_available_models()
if model_id in available:
return available[model_id].get("display_name", model_id)
return model_id
def get_arena_service():
"""Get ArenaService class for voting operations."""
return modal.Cls.from_name(APP_NAME, "ArenaService")
HEADER_MD = """
Arabic TTS Arena
Compare Arabic text‑to‑speech models side by side.
Listen, vote, and help build the community leaderboard.
Blog post
·
GitHub
"""
HOW_IT_WORKS_MD = """
How it works:
Enter Arabic text → Listen to two anonymous models → Vote for the better one
"""
# Leaderboard header removed — metadata is now rendered inline by refresh_leaderboard()
ABOUT_MD = """
📢 Latest Updates
—🤝 Have a model that should be here? Open a PR — we'd love to welcome it in.
Why We Built This
Arabic is spoken by over 400 million people. It's the language of poetry, prayer, storytelling, and everyday life. Yet when it comes to text-to-speech, Arabic has been an afterthought — tested in labs, benchmarked on charts, but rarely listened to by the people it's meant to serve.
We wanted to change that. Not with another paper or another metric — but by putting the microphone in your hands. You listen. You choose. Your ear is the benchmark.
How the Arena Works
- You type (or pick) an Arabic sentence
- Two anonymous models read it aloud
- You vote for the one that sounds more natural, more human
- Rankings update — and the best voices rise to the top
No model names are shown until after you vote, so every judgement is pure. Over time, thousands of these small choices build a leaderboard that reflects what people actually prefer — not what a loss function thinks is best.
Your Moves
| Choice | What it means |
|---|
| A is Better | Voice A sounded more natural to you |
| B is Better | Voice B sounded more natural to you |
| Both Good | Honestly, both sounded great |
| Both Bad | Neither felt right |
Quick Keys
| Key | Action |
|---|
| A | Vote for A |
| B | Vote for B |
| N | Next round |
Built with ❤️ for the Arabic-speaking world by Navid
"""
def decode_audio_to_file(audio_base64: str) -> Optional[str]:
"""Decode base64 WAV audio and write to a temp file.
Returns the file path (Gradio gr.Audio accepts file paths).
"""
import tempfile
try:
wav_bytes = base64.b64decode(audio_base64)
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp.write(wav_bytes)
tmp.flush()
tmp.close()
return tmp.name
except Exception:
return None
def synthesize_audio(text: str, model_id: str) -> dict:
"""Call the ArenaService to synthesize (or return cached) audio.
The backend checks its audio cache first and only calls the GPU model
on a cache miss, saving compute and reducing latency.
Retries up to MAX_SYNTHESIS_RETRIES times on failure before giving up.
"""
last_error = None
for attempt in range(1, MAX_SYNTHESIS_RETRIES + 1):
try:
service = get_arena_service()
result = service().synthesize_or_cache.remote(text, model_id)
if result.get("success"):
return result
last_error = result.get("error", "Unknown synthesis error")
print(f"⚠️ {model_id} attempt {attempt} failed: {last_error}")
except Exception as e:
last_error = str(e)
print(f"⚠️ {model_id} attempt {attempt} exception: {last_error}")
# All retries exhausted
return {
"success": False,
"error": f"{model_id} failed after {MAX_SYNTHESIS_RETRIES} attempts: {last_error}",
"model_id": model_id,
}
def _get_model_ratings() -> dict[str, dict]:
"""Read per-model elo and ci from the local leaderboard file."""
try:
if LEADERBOARD_FILE.exists():
with open(LEADERBOARD_FILE, "r") as f:
data = json.load(f)
return {
m["model_id"]: {
"elo": m.get("elo", 1000),
"ci": m.get("ci", 0),
"battles": m.get("battles", 0),
}
for m in data.get("models", [])
}
except Exception:
pass
return {}
def get_random_model_pair() -> tuple[str, str]:
"""Select two models using adaptive pairing for maximum information gain.
Combines two signals to score every possible pair:
1. **CI overlap** — pairs whose confidence intervals overlap are the
most uncertain (we don't know which is better), so a vote between
them is maximally informative. Measured as the fraction of overlap
relative to the smaller CI. Pairs with no CI data yet get the
maximum overlap score (1.0) so new models are explored.
2. **Under-sampling** — pairs where either model has few battles get
a boost via inverse-sqrt weighting, same as before.
The two signals are multiplied together and used as sampling weights
over all possible pairs, so the selection is stochastic (not greedy)
and every pair retains a non-zero chance of appearing.
"""
import math
from itertools import combinations
models = list(_get_available_models().keys())
if len(models) < 2:
raise ValueError("Not enough models available for comparison")
ratings = _get_model_ratings()
# --- score every candidate pair ---
pairs: list[tuple[str, str]] = list(combinations(models, 2))
pair_weights: list[float] = []
for a, b in pairs:
ra = ratings.get(a, {})
rb = ratings.get(b, {})
elo_a = ra.get("elo", 1000)
elo_b = rb.get("elo", 1000)
ci_a = ra.get("ci", 0)
ci_b = rb.get("ci", 0)
battles_a = ra.get("battles", 0)
battles_b = rb.get("battles", 0)
# -- Signal 1: CI overlap score (0–1) --
# If either model has no CI yet, treat as maximally uncertain → 1.0
if ci_a <= 0 or ci_b <= 0:
overlap_score = 1.0
else:
# Interval: [elo - ci, elo + ci]
lo_a, hi_a = elo_a - ci_a, elo_a + ci_a
lo_b, hi_b = elo_b - ci_b, elo_b + ci_b
overlap = max(0.0, min(hi_a, hi_b) - max(lo_a, lo_b))
span = min(ci_a, ci_b) * 2 # width of the smaller CI
overlap_score = min(overlap / span, 1.0) if span > 0 else 1.0
# Ensure a minimum floor so distant pairs still occasionally appear
overlap_score = max(overlap_score, 0.05)
# -- Signal 2: under-sampling boost --
exploration = (
1.0 / math.sqrt(battles_a + 1)
+ 1.0 / math.sqrt(battles_b + 1)
) / 2.0
pair_weights.append(overlap_score * exploration)
# --- sample one pair stochastically ---
(first, second), = random.choices(pairs, weights=pair_weights, k=1)
# Randomise A/B assignment so there's no positional bias
if random.random() < 0.5:
return (first, second)
return (second, first)
def get_random_sentence():
"""Return a random Arabic sample sentence."""
return random.choice(SAMPLE_SENTENCES)
def _empty_comparison():
"""Return values that reset the UI to the pre-synthesis state."""
return (
None,
None, # audio_a, audio_b
None,
None, # model_a_id, model_b_id
None,
None, # audio_a_base64, audio_b_base64
None,
None, # latency_a, latency_b
gr.update(visible=False), # audio_row
gr.update(visible=False), # vote_row
gr.update(visible=False), # result_display
gr.update(value="🔊 Synthesize", interactive=True), # synth_btn
gr.update(value="", visible=False), # status_display
"🔒 Hidden", # model_a_label
"🔒 Hidden", # model_b_label
gr.update(visible=False), # next_round_btn
gr.update(interactive=True), # vote_a_btn
gr.update(interactive=True), # vote_b_btn
gr.update(interactive=True), # vote_both_good_btn
gr.update(interactive=True), # vote_both_bad_btn
)
def _pick_replacement(exclude: set[str]) -> str | None:
"""Pick a model not in *exclude*, or None if none left."""
candidates = [m for m in _get_available_models() if m not in exclude]
return random.choice(candidates) if candidates else None
def _synth_one(text: str, model_id: str, used: set[str]) -> tuple[dict | None, str]:
"""Try to synthesize with *model_id*; on failure swap in a replacement once.
Returns (result_dict_or_None, final_model_id).
"""
result = synthesize_audio(text, model_id)
if result.get("success"):
return result, model_id
# First model failed after retries — try a replacement
replacement = _pick_replacement(used)
if replacement:
used.add(replacement)
result = synthesize_audio(text, replacement)
if result.get("success"):
return result, replacement
return None, model_id # give up
def generate_comparison(text: str):
"""Generate audio from two random TTS models for comparison.
Both models are synthesized in parallel using threads to halve wait time.
Uses a generator to yield status updates so the user sees progress.
"""
from concurrent.futures import ThreadPoolExecutor
if not text or not text.strip():
gr.Warning("Please enter some Arabic text first.")
yield _empty_comparison()
return
text = text.strip()
model_a_id, model_b_id = get_random_model_pair()
# — Show "synthesizing" status —
yield (
None,
None,
None,
None,
None,
None,
None,
None, # latency_a, latency_b
gr.update(visible=False),
gr.update(visible=False),
gr.update(visible=False),
gr.update(value="⏳ Synthesizing…", interactive=False),
gr.update(value="⏳ Generating audio from both models…", visible=True),
"🔒 Hidden",
"🔒 Hidden",
gr.update(visible=False),
gr.update(interactive=True),
gr.update(interactive=True),
gr.update(interactive=True),
gr.update(interactive=True),
)
# — Synthesize both models in parallel —
# Each thread gets its own used set for the replacement fallback logic.
def synth_a():
used = {model_a_id, model_b_id}
return _synth_one(text, model_a_id, used)
def synth_b():
used = {model_a_id, model_b_id}
return _synth_one(text, model_b_id, used)
try:
with ThreadPoolExecutor(max_workers=2) as pool:
future_a = pool.submit(synth_a)
future_b = pool.submit(synth_b)
result_a, model_a_id = future_a.result()
result_b, model_b_id = future_b.result()
except Exception as e:
gr.Warning(f"Connection error — is the backend deployed? ({e})")
yield _empty_comparison()
return
if result_a is None:
gr.Warning("Model A synthesis failed after retries. Please try again.")
yield _empty_comparison()
return
if result_b is None:
gr.Warning("Model B synthesis failed after retries. Please try again.")
yield _empty_comparison()
return
# — Decode audio to temp files for Gradio —
audio_a_path = decode_audio_to_file(result_a["audio_base64"])
audio_b_path = decode_audio_to_file(result_b["audio_base64"])
if not audio_a_path or not audio_b_path:
gr.Warning("Failed to decode audio from backend.")
yield _empty_comparison()
return
# Extract latency (None for cache hits / legacy responses)
latency_a = result_a.get("latency_seconds")
latency_b = result_b.get("latency_seconds")
yield (
audio_a_path,
audio_b_path,
model_a_id,
model_b_id,
result_a["audio_base64"],
result_b["audio_base64"],
latency_a,
latency_b,
gr.update(visible=True), # audio_row
gr.update(visible=True), # vote_row
gr.update(visible=False), # result_display
gr.update(value="🔊 Synthesize", interactive=True),
gr.update(value="", visible=False), # hide status
"🔒 Hidden", # model_a_label
"🔒 Hidden", # model_b_label
gr.update(visible=False), # next_round_btn
gr.update(interactive=True), # vote_a_btn
gr.update(interactive=True), # vote_b_btn
gr.update(interactive=True), # vote_both_good_btn
gr.update(interactive=True), # vote_both_bad_btn
)
def submit_vote(
vote: str,
text_prompt: str,
model_a_id: str,
model_b_id: str,
audio_a_b64: str,
audio_b_b64: str,
latency_a: float | None,
latency_b: float | None,
):
"""Submit a vote for the comparison."""
if not model_a_id or not model_b_id:
gr.Warning("Please synthesize audio first.")
return (
gr.update(visible=True), # vote_row stays
gr.update(visible=False), # result_display
gr.update(visible=False), # next_round_btn
"🔒 Hidden",
"🔒 Hidden",
gr.update(), # vote_a_btn unchanged
gr.update(), # vote_b_btn unchanged
gr.update(), # vote_both_good_btn unchanged
gr.update(), # vote_both_bad_btn unchanged
)
session_id = uuid.uuid4().hex
try:
service = get_arena_service()
result = service().record_vote.remote(
session_id=session_id,
text=text_prompt,
model_a=model_a_id,
model_b=model_b_id,
winner=vote,
audio_a_base64=audio_a_b64,
audio_b_base64=audio_b_b64,
latency_a=latency_a,
latency_b=latency_b,
)
except Exception as e:
gr.Warning(f"Vote failed: {e}")
return (
gr.update(visible=True),
gr.update(visible=False),
gr.update(visible=False),
"🔒 Hidden",
"🔒 Hidden",
gr.update(), # vote_a_btn unchanged
gr.update(), # vote_b_btn unchanged
gr.update(), # vote_both_good_btn unchanged
gr.update(), # vote_both_bad_btn unchanged
)
if not result.get("success"):
gr.Warning(f"Error: {result.get('error', 'Unknown')}")
return (
gr.update(visible=True),
gr.update(visible=False),
gr.update(visible=False),
"🔒 Hidden",
"🔒 Hidden",
gr.update(), # vote_a_btn unchanged
gr.update(), # vote_b_btn unchanged
gr.update(), # vote_both_good_btn unchanged
gr.update(), # vote_both_bad_btn unchanged
)
vote_emoji = {
"model_a": "🅰️ Model A",
"model_b": "🅱️ Model B",
"both_good": "👍 Both Good",
"both_bad": "👎 Both Bad",
}
name_a = get_display_name(model_a_id)
name_b = get_display_name(model_b_id)
result_md = f"""
✅ Vote Recorded!
You chose: {vote_emoji.get(vote, vote)}
🅰️ {name_a}
vs
🅱️ {name_b}
Thanks for voting! The leaderboard updates daily.
"""
return (
gr.update(visible=False), # hide vote_row
gr.update(value=result_md, visible=True), # show result
gr.update(visible=True), # show next_round_btn
f"**{name_a}**", # reveal model A
f"**{name_b}**", # reveal model B
gr.update(interactive=False), # disable vote_a_btn
gr.update(interactive=False), # disable vote_b_btn
gr.update(interactive=False), # disable vote_both_good_btn
gr.update(interactive=False), # disable vote_both_bad_btn
)
def refresh_leaderboard():
"""Read and display leaderboard from local JSON file."""
try:
if not LEADERBOARD_FILE.exists():
return _empty_leaderboard_md()
with open(LEADERBOARD_FILE, "r") as f:
data = json.load(f)
models = data.get("models", [])
last_updated = data.get("last_updated", "")
if not models:
return _empty_leaderboard_md()
# Hide models with fewer than 30 battles
models = [m for m in models if m.get("battles", 0) >= MIN_BATTLES]
if not models:
return _empty_leaderboard_md()
# Re-assign ranks after filtering
for i, m in enumerate(models, start=1):
m["rank"] = i
# Format timestamp
try:
from datetime import datetime
dt = datetime.fromisoformat(last_updated.replace("Z", "+00:00"))
updated_str = dt.strftime("%b %d, %Y")
except Exception:
updated_str = last_updated or "—"
# --- Styles ---
style_block = """
"""
# Metadata line
total_battles = sum(m.get("battles", 0) for m in models)
meta_html = (
f''
f'⚔️ {total_battles:,} battles'
f'·'
f'Updated {updated_str}'
f'
'
)
# Column labels
col_header = (
'"
)
# Build rows
tier_row = {1: "gold", 2: "silver", 3: "bronze"}
tier_rank = {1: "r-gold", 2: "r-silver", 3: "r-bronze"}
items_html = ""
for entry in models:
rank = entry["rank"]
name = entry["name"]
model_url = entry.get("model_url", "")
elo = entry["elo"]
ci = entry.get("ci", 0)
battles = entry.get("battles", 0)
avg_latency = entry.get("avg_latency")
gpu = entry.get("gpu", "")
row_cls = tier_row.get(rank, "")
rank_cls = tier_rank.get(rank, "")
is_api = not gpu
if model_url:
name_el = (
f'{name}'
)
else:
name_el = f'{name}'
votes_text = f"{battles:,}" if battles else "—"
ci_html = f'±{ci:.0f}' if ci else ""
latency_text = f"{avg_latency:.1f}s" if avg_latency is not None else "—"
if gpu:
gpu_attr = f' data-gpu="{gpu}"'
gpu_label = f'⚡ {gpu}'
elif is_api:
gpu_attr = ' data-gpu="API"'
gpu_label = '☁️ API'
else:
gpu_attr = ""
gpu_label = ""
items_html += (
f''
f'
{rank}
'
f"
{name_el}
"
f'
{elo:.0f}{ci_html}
'
f'
{latency_text}{gpu_label}
'
f'
{votes_text}
'
f"
'
f"{style_block}{meta_html}{col_header}"
f'
{items_html}
'
f"
'
"
No data yet!
"
"
Be the first to vote — head to the ⚔️ Battle tab.
"
"
Loading…
"
)
demo.load(fn=refresh_leaderboard, outputs=[leaderboard_display])
# About Tab
with gr.TabItem("📖 Story", id="story"):
gr.HTML(ABOUT_MD)
return demo
if __name__ == "__main__":
demo = create_demo()
demo.queue(default_concurrency_limit=4).launch(server_name="0.0.0.0", server_port=7860)