"""The Runs page for the Trackio UI."""
import re
import gradio as gr
import pandas as pd
try:
import trackio.utils as utils
from trackio.sqlite_storage import SQLiteStorage
from trackio.ui import fns
except ImportError:
import utils
from sqlite_storage import SQLiteStorage
from ui import fns
def get_runs_data(project):
"""Get the runs data as a pandas DataFrame."""
configs = SQLiteStorage.get_all_run_configs(project)
if not configs:
return pd.DataFrame()
df = pd.DataFrame.from_dict(configs, orient="index")
df = df.fillna("")
df.index.name = "Name"
df.reset_index(inplace=True)
df.rename(columns=fns.CONFIG_COLUMN_MAPPINGS, inplace=True)
if "Created" in df.columns:
df["Created"] = df["Created"].apply(utils.format_timestamp)
if "Username" in df.columns:
df["Username"] = df["Username"].apply(
lambda x: f"{x}"
if x and x != "None"
else x
)
if "Name" in df.columns:
df["Name"] = df["Name"].apply(
lambda x: f"{x}"
if x and x != "None"
else x
)
df.insert(0, " ", False)
columns = list(df.columns)
# Ensure columns appear immediately after Name in this order: Group, Username, Created
if "Name" in columns:
order = [col for col in ["Group", "Username", "Created"] if col in columns]
if order:
for col in order:
columns.remove(col)
name_idx = columns.index("Name")
insert_pos = name_idx + 1
for col in order:
columns.insert(insert_pos, col)
insert_pos += 1
df = df[columns]
return df
def get_runs_table(project):
df = get_runs_data(project)
if df.empty:
return gr.DataFrame(pd.DataFrame(), visible=False)
datatype = ["bool"] + ["markdown"] * (len(df.columns) - 1)
return gr.DataFrame(
df,
visible=True,
pinned_columns=2,
datatype=datatype,
wrap=True,
column_widths=["40px", "150px"],
interactive=True,
static_columns=list(range(1, len(df.columns))),
row_count=(len(df), "fixed"),
col_count=(len(df.columns), "fixed"),
)
def check_write_access_runs(request: gr.Request, write_token: str) -> bool:
"""
Check if the user has write access to the Trackio dashboard based on token validation.
The token is retrieved from the cookie in the request headers or, as fallback, from the
`write_token` query parameter.
"""
cookies = request.headers.get("cookie", "")
if cookies:
for cookie in cookies.split(";"):
parts = cookie.strip().split("=")
if len(parts) == 2 and parts[0] == "trackio_write_token":
return parts[1] == write_token
if hasattr(request, "query_params") and request.query_params:
token = request.query_params.get("write_token")
return token == write_token
return False
def set_deletion_allowed(request: gr.Request, oauth_token: gr.OAuthToken | None):
"""Update the delete button value and interactivity based on the runs data and user write access."""
if oauth_token:
try:
fns.check_oauth_token_has_write_access(oauth_token.token)
except PermissionError:
return (
gr.Button("⚠️ Need write access to delete runs", interactive=False),
gr.Dataframe(interactive=False),
False,
)
elif not check_write_access_runs(request, run_page.write_token):
return (
gr.Button("⚠️ Need write access to delete runs", interactive=False),
gr.Dataframe(interactive=False),
False,
)
return (
gr.Button("Select runs to delete", interactive=False),
gr.Dataframe(interactive=True),
True,
)
def update_delete_button(deletion_allowed, runs_data):
"""Update the delete button value and interactivity based on the selected runs."""
if not deletion_allowed:
return gr.Button(interactive=False)
num_selected = 0
if runs_data is not None and len(runs_data) > 0:
first_column_values = runs_data.iloc[:, 0].tolist()
num_selected = sum(1 for x in first_column_values if x)
if num_selected:
return gr.Button(f"Delete {num_selected} selected run(s)", interactive=True)
else:
return gr.Button("Select runs to delete", interactive=False)
def delete_selected_runs(deletion_allowed, runs_data, project, request: gr.Request):
"""Delete the selected runs and refresh the table."""
if not deletion_allowed:
return runs_data
first_column_values = runs_data.iloc[:, 0].tolist()
for i, selected in enumerate(first_column_values):
if selected:
run_name_raw = runs_data.iloc[i, 1]
match = re.search(r">([^<]+)<", run_name_raw)
run_name = match.group(1) if match else run_name_raw
SQLiteStorage.delete_run(project, run_name)
updated_data = get_runs_data(project)
return updated_data
with gr.Blocks() as run_page:
with gr.Sidebar() as sidebar:
logo_urls = utils.get_logo_urls()
logo = gr.Markdown(
f"""
"""
)
project_dd = gr.Dropdown(label="Project", allow_custom_value=True)
navbar = gr.Navbar(value=[("Metrics", ""), ("Runs", "/runs")], main_page_name=False)
timer = gr.Timer(value=1)
allow_deleting_runs = gr.State(False)
with gr.Row():
with gr.Column():
if utils.get_space():
gr.LoginButton("Login to delete runs", size="md")
with gr.Column():
with gr.Row():
delete_run_btn = gr.Button(
"⚠️ Need write access to delete runs",
interactive=False,
variant="stop",
size="md",
)
confirm_btn = gr.Button(
"Confirm delete", variant="stop", size="md", visible=False
)
cancel_btn = gr.Button("Cancel", size="md", visible=False)
runs_table = gr.DataFrame()
gr.on(
[run_page.load],
fn=fns.get_projects,
outputs=project_dd,
show_progress="hidden",
queue=False,
api_name=False,
)
gr.on(
[timer.tick],
fn=lambda: gr.Dropdown(info=fns.get_project_info()),
outputs=[project_dd],
show_progress="hidden",
api_name=False,
)
gr.on(
[project_dd.change],
fn=get_runs_table,
inputs=[project_dd],
outputs=[runs_table],
show_progress="hidden",
api_name=False,
queue=False,
).then(
fns.update_navbar_value,
inputs=[project_dd],
outputs=[navbar],
show_progress="hidden",
api_name=False,
queue=False,
)
gr.on(
[run_page.load],
fn=set_deletion_allowed,
inputs=[],
outputs=[delete_run_btn, runs_table, allow_deleting_runs],
show_progress="hidden",
api_name=False,
queue=False,
)
gr.on(
[runs_table.change],
fn=update_delete_button,
inputs=[allow_deleting_runs, runs_table],
outputs=[delete_run_btn],
show_progress="hidden",
api_name=False,
queue=False,
)
gr.on(
[delete_run_btn.click],
fn=lambda: [
gr.Button(visible=False),
gr.Button(visible=True),
gr.Button(visible=True),
],
inputs=None,
outputs=[delete_run_btn, confirm_btn, cancel_btn],
show_progress="hidden",
api_name=False,
queue=False,
)
gr.on(
[confirm_btn.click, cancel_btn.click],
fn=lambda: [
gr.Button(visible=True),
gr.Button(visible=False),
gr.Button(visible=False),
],
inputs=None,
outputs=[delete_run_btn, confirm_btn, cancel_btn],
show_progress="hidden",
api_name=False,
queue=False,
)
gr.on(
[confirm_btn.click],
fn=delete_selected_runs,
inputs=[allow_deleting_runs, runs_table, project_dd],
outputs=[runs_table],
show_progress="hidden",
api_name=False,
queue=False,
)