from dash import Dash, html, dcc, Input, Output, State, no_update, ctx import pandas as pd import dash_mantine_components as dmc import time from config import DATASET_ID from data_utils import create_fresh_duckdb_with_views, get_last_updated from graphs.leaderboard import ( leaderboard_callback_logic, ) from helpers import build_slider_marks, get_thumb_label_single, get_thumb_labels from layout_components import ( build_main_layout, build_range_slider, build_single_slider, DATA_URL, ) # Initialize the app app = Dash(suppress_callback_exceptions=True) # suppress callback exceptions for multi-page layout server = app.server # Load dataset and determine time range print(f"Attempting to connect to dataset from Hugging Face Hub: {DATASET_ID}") try: overall_start_time = time.time() # Create fresh connection, views, and read start/end time conn = create_fresh_duckdb_with_views() try: time_range = conn.execute("SELECT MIN(time) AS min_time, MAX(time) AS max_time FROM all_downloads;").fetchdf() start_dt = pd.to_datetime(time_range["min_time"].iloc[0]) end_dt = pd.to_datetime(time_range["max_time"].iloc[0]) finally: conn.close() msg = f"Successfully connected to datasets in {time.time() - overall_start_time:.2f}s." print(msg) except Exception as e: err_msg = f"Failed to load dataset(s). Error: {e}" print(err_msg) raise # Create a dcc slider for time range selection by year (readable marks) start_ts = int(start_dt.timestamp()) end_ts = int(end_dt.timestamp()) marks = build_slider_marks(start_dt, end_dt) time_slider = build_range_slider( start_ts, end_ts, [start_ts, end_ts], marks, thumb_children=get_thumb_labels([start_ts, end_ts]), ) time_slider_alltime = build_single_slider( start_ts, end_ts, end_ts, marks, thumb_children=get_thumb_label_single(end_ts), ) last_updated_label = get_last_updated() # Define the app layout app.layout = dmc.MantineProvider( theme={ "colorScheme": "light", "primaryColor": "blue", "fontFamily": "Inter, sans-serif", }, children=[ html.A(id="paper-redirect", style={"display": "none"}), html.Div(id="data-link-open-noop", style={"display": "none"}), dcc.Store(id="selected-view", data="all_downloads"), dcc.Store(id="model-attribution-type", data="uploader"), dcc.Store(id="time-slider-value", data=[start_ts, end_ts]), dcc.Store(id="time-slider-alltime-value", data=end_ts), build_main_layout(last_updated_label, time_slider), ], ) # ---------- # Callbacks # ---------- # Update model attribution type based on user selection @app.callback( Output("model-attribution-type", "data"), Input("model-attribution-segmented", "value"), ) def update_model_attribution_type(selected_value): return selected_value # Toggle between range slider and all-time slider @app.callback( Output("slider-container", "children"), Output("slider-description", "children"), Input("time-range-toggle", "checked"), State("time-slider-value", "data"), State("time-slider-alltime-value", "data"), ) def toggle_time_slider(is_alltime, range_value, alltime_value): if is_alltime: description = "Select a specific date to view all-time cumulative downloads up to that point." slider_value = alltime_value if alltime_value is not None else end_ts slider = build_single_slider( start_ts, end_ts, slider_value, marks, thumb_children=get_thumb_label_single(slider_value), ) return [slider], description else: description = "Adjust the slider to filter leaderboard results by the difference in downloads within the time range." slider_val = range_value if range_value is not None else [start_ts, end_ts] slider = build_range_slider( start_ts, end_ts, slider_val, marks, thumb_children=get_thumb_labels(slider_val), ) return [slider], description # Sync slider values to dcc.Store components @app.callback( Output("time-slider-value", "data"), Input("time-slider", "value"), ) def sync_time_slider_value(value): return value # Sync all-time slider value to dcc.Store component @app.callback( Output("time-slider-alltime-value", "data"), Input("time-slider-alltime", "value"), ) def sync_time_slider_alltime_value(value): return value # Update Top Countries leaderboard @app.callback( Output("top_countries-table", "children"), Output("top_countries-toggle", "children"), Input("top_countries-toggle", "n_clicks"), Input("time-slider-value", "data"), Input("time-slider-alltime-value", "data"), Input("time-range-toggle", "checked"), Input("selected-view", "data"), Input("model-attribution-type", "data"), State("top_countries-toggle", "children"), ) def update_top_countries( n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label ): # Use the appropriate slider value based on the toggle active_slider_value = slider_alltime_value if is_alltime else slider_value return leaderboard_callback_logic( n_clicks, active_slider_value, current_label, group_col="org_country_single", filename="top_countries", default_label="▼ Show Top 50", chip_color="#F0F9FF", view=selected_view, derived_author_toggle=(attribution_type == "original_creator"), is_alltime=is_alltime, ) # Update Top Developers leaderboard @app.callback( Output("top_developers-table", "children"), Output("top_developers-toggle", "children"), Input("top_developers-toggle", "n_clicks"), Input("time-slider-value", "data"), Input("time-slider-alltime-value", "data"), Input("time-range-toggle", "checked"), Input("selected-view", "data"), Input("model-attribution-type", "data"), State("top_developers-toggle", "children"), ) def update_top_developers( n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label ): # Use derived_author if attribution_type == "original_creator", else author group_col = "derived_author" if attribution_type == "original_creator" else "author" # Use the appropriate slider value based on the toggle active_slider_value = slider_alltime_value if is_alltime else slider_value return leaderboard_callback_logic( n_clicks, active_slider_value, current_label, group_col=group_col, filename="top_developers", default_label="▼ Show Top 50", chip_color="#F0F9FF", view=selected_view, derived_author_toggle=(attribution_type == "original_creator"), is_alltime=is_alltime, ) # Update Top Models leaderboard @app.callback( Output("top_models-table", "children"), Output("top_models-toggle", "children"), Input("top_models-toggle", "n_clicks"), Input("time-slider-value", "data"), Input("time-slider-alltime-value", "data"), Input("time-range-toggle", "checked"), Input("selected-view", "data"), Input("model-attribution-type", "data"), State("top_models-toggle", "children"), ) def update_top_models( n_clicks, slider_value, slider_alltime_value, is_alltime, selected_view, attribution_type, current_label ): # Use the appropriate slider value based on the toggle active_slider_value = slider_alltime_value if is_alltime else slider_value return leaderboard_callback_logic( n_clicks, active_slider_value, current_label, group_col="model", filename="top_models", default_label="▼ Show More", chip_color="#F0F9FF", view=selected_view, derived_author_toggle=(attribution_type == "original_creator"), is_alltime=is_alltime, ) # Update thumb labels for range slider @app.callback( Output("time-slider", "thumbChildren"), Input("time-slider", "value"), ) def update_thumb_labels(values): return get_thumb_labels(values) # Update thumb label for all-time slider @app.callback( Output("time-slider-alltime", "thumbChildren"), Input("time-slider-alltime", "value"), ) def update_thumb_label_alltime(value): return get_thumb_label_single(value) # Update selected view based on segmented control @app.callback( Output("selected-view", "data"), Input("segmented", "value"), ) def update_selected_view(seg_value): if seg_value == "filtered-downloads": return "one_year_rolling" return "all_downloads" @app.callback( Output("paper-disclaimer-modal", "opened"), Output("paper-redirect", "href"), Input("data-link-trigger-header", "n_clicks"), Input("paper-disclaimer-close", "n_clicks"), Input("paper-disclaimer-confirm", "n_clicks"), State("paper-disclaimer-modal", "opened"), prevent_initial_call=True, ) def handle_paper_disclaimer( data_clicks, close_clicks, confirm_clicks, modal_opened, ): triggered_id = ctx.triggered_id if triggered_id == "data-link-trigger-header": return True, no_update if triggered_id == "paper-disclaimer-close": return False, no_update if triggered_id == "paper-disclaimer-confirm": return False, DATA_URL return modal_opened, no_update app.clientside_callback( """ function(url) { if (url) { window.open(url, "_blank", "noopener,noreferrer"); } return ""; } """, Output("data-link-open-noop", "children"), Input("paper-redirect", "href"), prevent_initial_call=True, ) # Run the app if __name__ == "__main__": app.run(debug=True)