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)