Spaces:
Running
Running
| # 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=""" | |
| <h2 class="tab-header">🏆 Leaderboard</h2> | |
| <p style="padding-left: 8px;"> | |
| 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. | |
| </p> | |
| """, | |
| 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=""" | |
| <h2 style="padding-top: 12px;" class="tab-header">📊 Head-to-Head Matchups</h2> | |
| <p style="padding-left: 8px; width: 80%;"> | |
| 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. | |
| </p> | |
| """, | |
| 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=""" | |
| <h2>Citation</h2> | |
| <p style="padding: 0 8px;"> | |
| When referencing this leaderboard or its dataset in academic publications, please cite: | |
| </p> | |
| """, | |
| 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=""" | |
| <h2>Terms of Use</h2> | |
| <p style="padding: 0 8px;"> | |
| Users are required to agree to the following terms before using the service: | |
| </p> | |
| <p style="padding: 0 8px;"> | |
| 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 | |
| <a href="https://discord.com/invite/humeai" target="_blank" class="provider-link"> | |
| Discord community | |
| </a>. | |
| </p> | |
| """, | |
| padding=False, | |
| ) | |
| gr.HTML( | |
| value=""" | |
| <h2>Acknowledgements</h2> | |
| <p style="padding: 0 8px;"> | |
| We thank all participants who contributed their votes to help build this leaderboard. | |
| </p> | |
| """, | |
| 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 | |