Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| from gradio.components import Dataframe | |
| from gradio_leaderboard import Leaderboard, ColumnFilter, SelectColumns | |
| import pandas as pd | |
| from apscheduler.schedulers.background import BackgroundScheduler | |
| import os | |
| from gradio.themes import Soft | |
| from src.about import ( | |
| CITATION_BUTTON_LABEL, | |
| CITATION_BUTTON_TEXT, | |
| EVALUATION_QUEUE_TEXT, | |
| INTRODUCTION_TEXT, | |
| LLM_BENCHMARKS_TEXT, | |
| TITLE, | |
| Tasks | |
| ) | |
| from src.display.css_html_js import custom_css | |
| from src.display.utils import ( | |
| BENCHMARK_COLS, | |
| COLS, | |
| EVAL_COLS, | |
| EVAL_TYPES, | |
| AutoEvalColumn, | |
| auto_eval_column_attrs | |
| ) | |
| from src.envs import API, EVAL_REQUESTS_PATH, EVAL_RESULTS_PATH, REPO_ID, LOCAL_MODE | |
| from src.populate import get_evaluation_queue_df, get_leaderboard_df | |
| from src.submission.submit import add_new_eval | |
| def restart_space(): | |
| """Restart the Hugging Face space""" | |
| if LOCAL_MODE: | |
| print("Running in local mode, skipping space restart") | |
| return | |
| try: | |
| API.restart_space(repo_id=REPO_ID) | |
| except Exception as e: | |
| print(f"Failed to restart space: {e}") | |
| print("Continuing without restart") | |
| ### Space initialisation | |
| def initialize_data_directories(): | |
| """Initialize directories for assessment data""" | |
| # Create local directories if they don't exist | |
| os.makedirs(EVAL_REQUESTS_PATH, exist_ok=True) | |
| os.makedirs(EVAL_RESULTS_PATH, exist_ok=True) | |
| # Initialize data | |
| initialize_data_directories() | |
| # Load data for leaderboard | |
| LEADERBOARD_DF = get_leaderboard_df(EVAL_RESULTS_PATH, EVAL_REQUESTS_PATH, COLS, BENCHMARK_COLS) | |
| # Extract unique languages for filtering | |
| def get_unique_languages(df): | |
| """Extract all unique individual languages from the Language column""" | |
| if df.empty or auto_eval_column_attrs.language.name not in df.columns: | |
| return [] | |
| all_languages = set() | |
| for value in df[auto_eval_column_attrs.language.name].unique(): | |
| if isinstance(value, str): | |
| if "/" in value: | |
| languages = [lang.strip() for lang in value.split("/")] | |
| all_languages.update(languages) | |
| else: | |
| all_languages.add(value.strip()) | |
| return sorted(list(all_languages)) | |
| # Create a mapping for language filtering | |
| UNIQUE_LANGUAGES = get_unique_languages(LEADERBOARD_DF) | |
| # Create a special column for individual language filtering | |
| if not LEADERBOARD_DF.empty: | |
| # Create a column that contains all individual languages as a list | |
| LEADERBOARD_DF["_languages_list"] = LEADERBOARD_DF[auto_eval_column_attrs.language.name].apply( | |
| lambda x: [lang.strip() for lang in str(x).split("/")] if pd.notna(x) else [] | |
| ) | |
| # Create a text version of Active Maintenance for checkboxgroup filtering | |
| LEADERBOARD_DF["_maintenance_filter"] = LEADERBOARD_DF[auto_eval_column_attrs.availability.name].apply( | |
| lambda x: "Active" if x else "Inactive" | |
| ) | |
| # Load queue data | |
| ( | |
| finished_eval_queue_df, | |
| running_eval_queue_df, | |
| pending_eval_queue_df, | |
| rejected_eval_queue_df, | |
| ) = get_evaluation_queue_df(EVAL_REQUESTS_PATH, EVAL_COLS) | |
| def init_leaderboard(dataframe): | |
| """Initialize the leaderboard component""" | |
| if dataframe is None or dataframe.empty: | |
| # Create an empty dataframe with the expected columns | |
| all_columns = COLS + [task.value.col_name for task in Tasks] | |
| empty_df = pd.DataFrame(columns=pd.Index(all_columns)) | |
| print("Warning: Leaderboard DataFrame is empty. Using empty dataframe.") | |
| dataframe = empty_df | |
| # Create filter columns list with proper typing | |
| filter_columns = [] | |
| # 1. Library types | |
| filter_columns.append(ColumnFilter(auto_eval_column_attrs.library_type.name, type="checkboxgroup", label="Library types")) | |
| # 2. Programming Language (checkboxgroup - OR filtering) | |
| filter_columns.append(ColumnFilter(auto_eval_column_attrs.language.name, type="checkboxgroup", label="Programming Language")) | |
| # 3. GitHub Stars | |
| filter_columns.append(ColumnFilter( | |
| auto_eval_column_attrs.stars.name, | |
| type="slider", | |
| min=0, | |
| max=50000, | |
| label="GitHub Stars", | |
| )) | |
| # 4. Maintenance Status (checkboxgroup - separate from languages) | |
| filter_columns.append(ColumnFilter("_maintenance_filter", type="checkboxgroup", label="Maintenance Status")) | |
| # Hide columns | |
| hidden_columns = [getattr(auto_eval_column_attrs, field).name for field in AutoEvalColumn.model_fields if getattr(auto_eval_column_attrs, field).hidden] | |
| hidden_columns.extend(["_languages_list", "_maintenance_filter", "_original_language"]) # Hide helper columns | |
| return Leaderboard( | |
| value=dataframe, | |
| datatype="markdown", | |
| select_columns=SelectColumns( | |
| default_selection=[getattr(auto_eval_column_attrs, field).name for field in AutoEvalColumn.model_fields if getattr(auto_eval_column_attrs, field).displayed_by_default], | |
| cant_deselect=[getattr(auto_eval_column_attrs, field).name for field in AutoEvalColumn.model_fields if getattr(auto_eval_column_attrs, field).never_hidden], | |
| label="Select Columns to Display:", | |
| ), | |
| search_columns=[auto_eval_column_attrs.library.name, auto_eval_column_attrs.license_name.name], | |
| hide_columns=hidden_columns, | |
| filter_columns=filter_columns, # type: ignore | |
| bool_checkboxgroup_label="Filter libraries", | |
| interactive=False, | |
| ) | |
| demo = gr.Blocks(css=custom_css, theme=Soft()) | |
| # demo = gr.Blocks(css=custom_css, theme=Soft(font=["sans-serif"], font_mono=["monospace"])) | |
| with demo: | |
| gr.HTML(TITLE) | |
| gr.Markdown(INTRODUCTION_TEXT, elem_classes="markdown-text") | |
| with gr.Tabs(elem_classes="tab-buttons") as tabs: | |
| with gr.TabItem("π Vulnerability Leaderboard", elem_id="vulnerability-leaderboard-tab", id=0): | |
| leaderboard = init_leaderboard(LEADERBOARD_DF) | |
| with gr.TabItem("π About", elem_id="about-tab", id=2): | |
| gr.Markdown(LLM_BENCHMARKS_TEXT, elem_classes="markdown-text") | |
| with gr.TabItem("π Submit Library", elem_id="submit-library-tab", id=3): | |
| with gr.Column(): | |
| with gr.Row(): | |
| gr.Markdown(EVALUATION_QUEUE_TEXT, elem_classes="markdown-text") | |
| with gr.Column(): | |
| with gr.Accordion( | |
| f"β Completed Assessments ({len(finished_eval_queue_df)})", | |
| open=False, | |
| ): | |
| with gr.Row(): | |
| finished_eval_table = Dataframe( | |
| value=finished_eval_queue_df, | |
| headers=EVAL_COLS, | |
| datatype=EVAL_TYPES, | |
| row_count=5, | |
| ) | |
| with gr.Accordion( | |
| f"π In Progress Assessments ({len(running_eval_queue_df)})", | |
| open=False, | |
| ): | |
| with gr.Row(): | |
| running_eval_table = Dataframe( | |
| value=running_eval_queue_df, | |
| headers=EVAL_COLS, | |
| datatype=EVAL_TYPES, | |
| row_count=5, | |
| ) | |
| with gr.Accordion( | |
| f"β³ Pending Assessment Queue ({len(pending_eval_queue_df)})", | |
| open=False, | |
| ): | |
| with gr.Row(): | |
| pending_eval_table = Dataframe( | |
| value=pending_eval_queue_df, | |
| headers=EVAL_COLS, | |
| datatype=EVAL_TYPES, | |
| row_count=5, | |
| ) | |
| with gr.Row(): | |
| gr.Markdown("# βοΈβ¨ Submit a library for vulnerability assessment", elem_classes="markdown-text") | |
| with gr.Row(): | |
| with gr.Column(): | |
| library_name_textbox = gr.Textbox(label="Library name") | |
| submit_button = gr.Button("Submit for Assessment") | |
| submission_result = gr.Markdown() | |
| submit_button.click( | |
| add_new_eval, | |
| [ | |
| library_name_textbox, | |
| ], | |
| submission_result, | |
| ) | |
| with gr.Row(): | |
| with gr.Accordion("π Citation", open=True): | |
| citation_button = gr.Code( | |
| value=CITATION_BUTTON_TEXT, | |
| label=CITATION_BUTTON_LABEL, | |
| lines=14, | |
| elem_id="citation-button", | |
| language="yaml", | |
| ) | |
| # Only schedule space restarts if not in local mode | |
| if not LOCAL_MODE: | |
| scheduler = BackgroundScheduler() | |
| scheduler.add_job(restart_space, "interval", seconds=1800) | |
| scheduler.start() | |
| # Launch the app | |
| demo.queue(default_concurrency_limit=40).launch(show_error=True) |