# Standard Library Imports
import hashlib
import json
import time
from typing import List, Optional, Tuple, Union
# Third-Party Library Imports
import gradio as gr
# Local Application Imports
from src.common import logger
from src.core import VotingService
class Leaderboard:
"""
Manages the state, data fetching, and UI construction for the Leaderboard tab.
Includes caching and throttling for leaderboard data updates.
"""
def __init__(self, voting_service: VotingService):
"""
Initializes the Leaderboard component.
Args:
voting_service: The service for voting/leaderboard DB operations.
"""
self.voting_service = voting_service
# leaderboard update state
self.leaderboard_data: List[List[str]] = [[]]
self.battle_counts_data: List[List[str]] = [[]]
self.win_rates_data: List[List[str]] = [[]]
self.leaderboard_cache_hash: Optional[str] = None
self.last_leaderboard_update_time: float = 0.0
self.min_refresh_interval: int = 30
async def _update_leaderboard_data(self, force: bool = False) -> bool:
"""
Fetches leaderboard data from the source if cache is stale or force=True.
Updates internal state variables (leaderboard_data, battle_counts_data,
win_rates_data, cache_hash, last_update_time) if new data is fetched.
Uses time-based throttling defined by `min_refresh_interval`.
Args:
force: If True, bypasses cache hash check and time throttling.
Returns:
True if the leaderboard data state was updated, False otherwise.
"""
current_time = time.time()
time_since_last_update = current_time - self.last_leaderboard_update_time
# Skip update if throttled and not forced
if not force and time_since_last_update < self.min_refresh_interval:
logger.debug(f"Skipping leaderboard update (throttled): last updated {time_since_last_update:.1f}s ago.")
return False
try:
# Fetch the latest data
(
latest_leaderboard_data,
latest_battle_counts_data,
latest_win_rates_data
) = await self.voting_service.get_formatted_leaderboard_data()
# Check if data is valid before proceeding
if not latest_leaderboard_data or not latest_leaderboard_data[0]:
logger.error("Invalid data received from get_leaderboard_data.")
return False
# Generate a hash of the primary leaderboard data to check for changes
# Use a stable serialization format (sort_keys=True)
data_str = json.dumps(latest_leaderboard_data, sort_keys=True)
new_data_hash = hashlib.md5(data_str.encode()).hexdigest()
# Skip if data hasn't changed and not forced
if not force and new_data_hash == self.leaderboard_cache_hash:
logger.debug("Leaderboard data unchanged since last fetch.")
return False
# Update the state and cache
self.leaderboard_data = latest_leaderboard_data
self.battle_counts_data = latest_battle_counts_data
self.win_rates_data = latest_win_rates_data
self.leaderboard_cache_hash = new_data_hash
self.last_leaderboard_update_time = current_time
logger.info("Leaderboard data updated successfully.")
return True
except Exception as e:
logger.error(f"Failed to update leaderboard data: {e!s}", exc_info=True)
return False
async def refresh_leaderboard(
self, force: bool = False
) -> Tuple[Union[dict, gr.skip], Union[dict, gr.skip], Union[dict, gr.skip]]:
"""
Refreshes leaderboard data state and returns Gradio updates for the tables.
Calls `_update_leaderboard_data` and returns updates only if data changed
or `force` is True. Returns gr.skip() otherwise.
Args:
force: If True, forces `_update_leaderboard_data` to bypass throttling/cache.
Returns:
A tuple of Gradio update dictionaries for the leaderboard, battle counts,
and win rates tables, or gr.skip() for each if no update is needed.
Raises:
gr.Error: If leaderboard data is empty/invalid after attempting an update.
(Changed from previous: now raises only if data is *still* bad)
"""
data_updated = await self._update_leaderboard_data(force=force)
if not self.leaderboard_data or not isinstance(self.leaderboard_data[0], list):
logger.error("Leaderboard data is empty or invalid after update attempt.")
raise gr.Error("Unable to retrieve leaderboard data. Please refresh the page or try again shortly.")
if data_updated or force:
logger.debug("Returning leaderboard table updates.")
return (
gr.update(value=self.leaderboard_data),
gr.update(value=self.battle_counts_data),
gr.update(value=self.win_rates_data)
)
logger.debug("Skipping leaderboard table updates (no data change).")
return gr.skip(), gr.skip(), gr.skip()
async def build_leaderboard_section(self) -> Tuple[gr.DataFrame, gr.DataFrame, gr.DataFrame]:
"""
Constructs the Gradio UI layout for the Leaderboard tab.
Defines the DataFrames, HTML descriptions, and refresh button logic.
Returns:
A tuple containing the Gradio DataFrame components for:
- Main Leaderboard table
- Battle Counts table
- Win Rates table
These components are needed by the main Frontend class to wire up events.
"""
logger.debug("Building Leaderboard UI section...")
# Pre-load leaderboard data before building UI that depends on it
await self._update_leaderboard_data(force=True)
# --- UI components ---
with gr.Row():
with gr.Column(scale=5):
gr.HTML(
value="""
This leaderboard presents community voting results for different TTS providers, showing which
ones users found more expressive and natural-sounding. The win rate reflects how often each
provider was selected as the preferred option in head-to-head comparisons. Click the refresh
button to see the most up-to-date voting results.
""",
padding=False,
)
refresh_button = gr.Button(
"↻ Refresh",
variant="primary",
elem_classes="refresh-btn",
scale=1,
)
with gr.Column(elem_id="leaderboard-table-container"):
leaderboard_table = gr.DataFrame(
headers=["Rank", "Provider", "Model", "Win Rate", "Votes"],
datatype=["html", "html", "html", "html", "html"],
column_widths=[80, 300, 180, 120, 116],
value=self.leaderboard_data,
min_width=680,
interactive=False,
render=True,
elem_id="leaderboard-table"
)
with gr.Column():
gr.HTML(
value="""
These tables show how each provider performs against others in direct comparisons.
The first table shows the total number of comparisons between each pair of providers.
The second table shows the win rate (percentage) of the row provider against the column provider.
""",
padding=False
)
with gr.Row(equal_height=True):
with gr.Column(min_width=420):
battle_counts_table = gr.DataFrame(
headers=["", "Hume AI", "OpenAI", "ElevenLabs"],
datatype=["html", "html", "html", "html"],
column_widths=[132, 132, 132, 132],
value=self.battle_counts_data,
interactive=False,
)
with gr.Column(min_width=420):
win_rates_table = gr.DataFrame(
headers=["", "Hume AI", "OpenAI", "ElevenLabs"],
datatype=["html", "html", "html", "html"],
column_widths=[132, 132, 132, 132],
value=self.win_rates_data,
interactive=False,
)
with gr.Accordion(label="Citation", open=False):
with gr.Column(variant="panel"):
with gr.Column(variant="panel"):
gr.HTML(
value="""
Citation
When referencing this leaderboard or its dataset in academic publications, please cite:
""",
padding=False,
)
gr.Markdown(
value="""
**BibTeX**
```BibTeX
@misc{expressive-tts-arena,
title = {Expressive TTS Arena: An Open Platform for Evaluating Text-to-Speech Expressiveness by Human Preference},
author = {Alan Cowen, Zachary Greathouse, Richard Marmorstein, Jeremy Hadfield},
year = {2025},
publisher = {Hugging Face},
howpublished = {\\url{https://huggingface.co/spaces/HumeAI/expressive-tts-arena}}
}
```
"""
)
gr.HTML(
value="""
Terms of Use
Users are required to agree to the following terms before using the service:
All generated audio clips are provided for research and evaluation purposes only.
The audio content may not be redistributed or used for commercial purposes without
explicit permission. Users should not upload any private or personally identifiable
information. Please report any bugs, issues, or concerns to our
Discord community
.
""",
padding=False,
)
gr.HTML(
value="""
Acknowledgements
We thank all participants who contributed their votes to help build this leaderboard.
""",
padding=False,
)
# Wrapper for the async refresh function
async def async_refresh_handler() -> Tuple[Union[dict, gr.skip], Union[dict, gr.skip], Union[dict, gr.skip]]:
"""Async helper to call refresh_leaderboard and handle its tuple return."""
logger.debug("Refresh button clicked, calling async_refresh_handler.")
return await self.refresh_leaderboard(force=True)
# Handler to re-enable the button after a short delay
def reenable_button() -> dict: # Returns a Gradio update dict
"""Waits briefly and returns an update to re-enable the refresh button."""
throttle_delay = 3 # seconds
time.sleep(throttle_delay) # Okay in Gradio event handlers (runs in thread)
return gr.update(interactive=True)
# Refresh button click event handler
refresh_button.click(
fn=lambda _=None: (gr.update(interactive=False)), # Disable button immediately
inputs=[],
outputs=[refresh_button],
).then(
fn=async_refresh_handler,
inputs=[],
outputs=[leaderboard_table, battle_counts_table, win_rates_table] # Update all three tables
).then(
fn=reenable_button, # Re-enable the button after a delay
inputs=[],
outputs=[refresh_button]
)
logger.debug("Leaderboard UI section built.")
# Return the component instances needed by the Frontend class
return leaderboard_table, battle_counts_table, win_rates_table