| __all__ = ['block', 'make_clickable_model', 'make_clickable_user', 'get_submissions'] | |
| import os | |
| import io | |
| import gradio as gr | |
| import pandas as pd | |
| import json | |
| import shutil | |
| import tempfile | |
| import datetime | |
| import zipfile | |
| import numpy as np | |
| from constants import * | |
| from huggingface_hub import Repository | |
| HF_TOKEN = os.environ.get("HF_TOKEN") | |
| global data_component, filter_component | |
| def upload_file(files): | |
| file_paths = [file.name for file in files] | |
| return file_paths | |
| # def add_new_eval( | |
| # input_file, | |
| # model_name_textbox: str, | |
| # revision_name_textbox: str, | |
| # model_link: str, | |
| # team_name: str, | |
| # contact_email: str, | |
| # access_type: str, | |
| # model_publish: str, | |
| # model_resolution: str, | |
| # model_fps: str, | |
| # model_frame: str, | |
| # model_video_length: str, | |
| # model_checkpoint: str, | |
| # model_commit_id: str, | |
| # model_video_format: str | |
| # ): | |
| # if input_file is None: | |
| # return "Error! Empty file!" | |
| # if model_link == '' or model_name_textbox == '' or contact_email == '': | |
| # return gr.update(visible=True), gr.update(visible=False), gr.update(visible=True) | |
| # # upload_data=json.loads(input_file) | |
| # upload_content = input_file | |
| # submission_repo = Repository(local_dir=SUBMISSION_NAME, clone_from=SUBMISSION_URL, use_auth_token=HF_TOKEN, repo_type="dataset") | |
| # submission_repo.git_pull() | |
| # filename = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") | |
| # now = datetime.datetime.now() | |
| # update_time = now.strftime("%Y-%m-%d") # Capture update time | |
| # with open(f'{SUBMISSION_NAME}/{filename}.zip','wb') as f: | |
| # f.write(input_file) | |
| # # shutil.copyfile(CSV_DIR, os.path.join(SUBMISSION_NAME, f"{input_file}")) | |
| # csv_data = pd.read_csv(CSV_DIR) | |
| # if revision_name_textbox == '': | |
| # col = csv_data.shape[0] | |
| # model_name = model_name_textbox.replace(',',' ') | |
| # else: | |
| # model_name = revision_name_textbox.replace(',',' ') | |
| # model_name_list = csv_data['Model Name (clickable)'] | |
| # name_list = [name.split(']')[0][1:] for name in model_name_list] | |
| # if revision_name_textbox not in name_list: | |
| # col = csv_data.shape[0] | |
| # else: | |
| # col = name_list.index(revision_name_textbox) | |
| # if model_link == '': | |
| # model_name = model_name # no url | |
| # else: | |
| # model_name = '[' + model_name + '](' + model_link + ')' | |
| # os.makedirs(filename, exist_ok=True) | |
| # with zipfile.ZipFile(io.BytesIO(input_file), 'r') as zip_ref: | |
| # zip_ref.extractall(filename) | |
| # upload_data = {} | |
| # for file in os.listdir(filename): | |
| # if file.startswith('.') or file.startswith('__'): | |
| # print(f"Skip the file: {file}") | |
| # continue | |
| # cur_file = os.path.join(filename, file) | |
| # if os.path.isdir(cur_file): | |
| # for subfile in os.listdir(cur_file): | |
| # if subfile.endswith(".json"): | |
| # with open(os.path.join(cur_file, subfile)) as ff: | |
| # cur_json = json.load(ff) | |
| # print(file, type(cur_json)) | |
| # if isinstance(cur_json, dict): | |
| # print(cur_json.keys()) | |
| # for key in cur_json: | |
| # upload_data[key.replace('_',' ')] = cur_json[key][0] | |
| # print(f"{key}:{cur_json[key][0]}") | |
| # elif cur_file.endswith('json'): | |
| # with open(cur_file) as ff: | |
| # cur_json = json.load(ff) | |
| # print(file, type(cur_json)) | |
| # if isinstance(cur_json, dict): | |
| # print(cur_json.keys()) | |
| # for key in cur_json: | |
| # upload_data[key.replace('_',' ')] = cur_json[key][0] | |
| # print(f"{key}:{cur_json[key][0]}") | |
| # # add new data | |
| # new_data = [model_name] | |
| # print('upload_data:', upload_data) | |
| # for key in TASK_INFO: | |
| # if key in upload_data: | |
| # new_data.append(upload_data[key]) | |
| # else: | |
| # new_data.append(0) | |
| # if team_name =='' or 'vbench' in team_name.lower(): | |
| # new_data.append("User Upload") | |
| # else: | |
| # new_data.append(team_name) | |
| # new_data.append(contact_email.replace(',',' and ')) # Add contact email [private] | |
| # new_data.append(update_time) # Add the update time | |
| # new_data.append(team_name) | |
| # new_data.append(access_type) | |
| # csv_data.loc[col] = new_data | |
| # csv_data = csv_data.to_csv(CSV_DIR, index=False) | |
| # with open(INFO_DIR,'a') as f: | |
| # f.write(f"{model_name}\t{update_time}\t{model_publish}\t{model_resolution}\t{model_fps}\t{model_frame}\t{model_video_length}\t{model_checkpoint}\t{model_commit_id}\t{model_video_format}\n") | |
| # submission_repo.push_to_hub() | |
| # print("success update", model_name) | |
| # return gr.update(visible=False), gr.update(visible=True), gr.update(visible=False) | |
| # def add_new_eval_i2v( | |
| # input_file, | |
| # model_name_textbox: str, | |
| # revision_name_textbox: str, | |
| # model_link: str, | |
| # team_name: str, | |
| # contact_email: str, | |
| # access_type: str, | |
| # model_publish: str, | |
| # model_resolution: str, | |
| # model_fps: str, | |
| # model_frame: str, | |
| # model_video_length: str, | |
| # model_checkpoint: str, | |
| # model_commit_id: str, | |
| # model_video_format: str | |
| # ): | |
| # COLNAME2KEY={ | |
| # "Video-Text Camera Motion":"camera_motion", | |
| # "Video-Image Subject Consistency": "i2v_subject", | |
| # "Video-Image Background Consistency": "i2v_background", | |
| # "Subject Consistency": "subject_consistency", | |
| # "Background Consistency": "background_consistency", | |
| # "Motion Smoothness": "motion_smoothness", | |
| # "Dynamic Degree": "dynamic_degree", | |
| # "Aesthetic Quality": "aesthetic_quality", | |
| # "Imaging Quality": "imaging_quality", | |
| # "Temporal Flickering": "temporal_flickering" | |
| # } | |
| # if input_file is None: | |
| # return "Error! Empty file!" | |
| # if model_link == '' or model_name_textbox == '' or contact_email == '': | |
| # return gr.update(visible=True), gr.update(visible=False), gr.update(visible=True) | |
| # upload_content = input_file | |
| # submission_repo = Repository(local_dir=SUBMISSION_NAME, clone_from=SUBMISSION_URL, use_auth_token=HF_TOKEN, repo_type="dataset") | |
| # submission_repo.git_pull() | |
| # filename = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") | |
| # now = datetime.datetime.now() | |
| # update_time = now.strftime("%Y-%m-%d") # Capture update time | |
| # with open(f'{SUBMISSION_NAME}/{filename}.zip','wb') as f: | |
| # f.write(input_file) | |
| # # shutil.copyfile(CSV_DIR, os.path.join(SUBMISSION_NAME, f"{input_file}")) | |
| # csv_data = pd.read_csv(I2V_DIR) | |
| # if revision_name_textbox == '': | |
| # col = csv_data.shape[0] | |
| # model_name = model_name_textbox.replace(',',' ') | |
| # else: | |
| # model_name = revision_name_textbox.replace(',',' ') | |
| # model_name_list = csv_data['Model Name (clickable)'] | |
| # name_list = [name.split(']')[0][1:] for name in model_name_list] | |
| # if revision_name_textbox not in name_list: | |
| # col = csv_data.shape[0] | |
| # else: | |
| # col = name_list.index(revision_name_textbox) | |
| # if model_link == '': | |
| # model_name = model_name # no url | |
| # else: | |
| # model_name = '[' + model_name + '](' + model_link + ')' | |
| # os.makedirs(filename, exist_ok=True) | |
| # with zipfile.ZipFile(io.BytesIO(input_file), 'r') as zip_ref: | |
| # zip_ref.extractall(filename) | |
| # upload_data = {} | |
| # for file in os.listdir(filename): | |
| # if file.startswith('.') or file.startswith('__'): | |
| # print(f"Skip the file: {file}") | |
| # continue | |
| # cur_file = os.path.join(filename, file) | |
| # if os.path.isdir(cur_file): | |
| # for subfile in os.listdir(cur_file): | |
| # if subfile.endswith(".json"): | |
| # with open(os.path.join(cur_file, subfile)) as ff: | |
| # cur_json = json.load(ff) | |
| # print(file, type(cur_json)) | |
| # if isinstance(cur_json, dict): | |
| # print(cur_json.keys()) | |
| # for key in cur_json: | |
| # upload_data[key] = cur_json[key][0] | |
| # print(f"{key}:{cur_json[key][0]}") | |
| # elif cur_file.endswith('json'): | |
| # with open(cur_file) as ff: | |
| # cur_json = json.load(ff) | |
| # print(file, type(cur_json)) | |
| # if isinstance(cur_json, dict): | |
| # print(cur_json.keys()) | |
| # for key in cur_json: | |
| # upload_data[key] = cur_json[key][0] | |
| # print(f"{key}:{cur_json[key][0]}") | |
| # # add new data | |
| # new_data = [model_name] | |
| # print('upload_data:', upload_data) | |
| # I2V_HEAD= ["Video-Text Camera Motion", | |
| # "Video-Image Subject Consistency", | |
| # "Video-Image Background Consistency", | |
| # "Subject Consistency", | |
| # "Background Consistency", | |
| # "Temporal Flickering", | |
| # "Motion Smoothness", | |
| # "Dynamic Degree", | |
| # "Aesthetic Quality", | |
| # "Imaging Quality" ] | |
| # for key in I2V_HEAD : | |
| # sub_key = COLNAME2KEY[key] | |
| # if sub_key in upload_data: | |
| # new_data.append(upload_data[sub_key]) | |
| # else: | |
| # new_data.append(0) | |
| # if team_name =='' or 'vbench' in team_name.lower(): | |
| # new_data.append("User Upload") | |
| # else: | |
| # new_data.append(team_name) | |
| # new_data.append(contact_email.replace(',',' and ')) # Add contact email [private] | |
| # new_data.append(update_time) # Add the update time | |
| # new_data.append(team_name) | |
| # new_data.append(access_type) | |
| # csv_data.loc[col] = new_data | |
| # csv_data = csv_data.to_csv(I2V_DIR , index=False) | |
| # with open(INFO_DIR,'a') as f: | |
| # f.write(f"{model_name}\t{update_time}\t{model_publish}\t{model_resolution}\t{model_fps}\t{model_frame}\t{model_video_length}\t{model_checkpoint}\t{model_commit_id}\t{model_video_format}\n") | |
| # submission_repo.push_to_hub() | |
| # print("success update", model_name) | |
| # return gr.update(visible=False), gr.update(visible=True), gr.update(visible=False) | |
| def get_baseline_df(): | |
| submission_repo = Repository(local_dir=SUBMISSION_NAME, clone_from=SUBMISSION_URL, use_auth_token=HF_TOKEN, repo_type="dataset") | |
| submission_repo.git_pull() | |
| df = pd.read_csv(CSV_DIR) | |
| df = df.sort_values(by=DEFAULT_INFO[0], ascending=False) | |
| # Add this line to display the results of all model types by default | |
| df = df[df['Model Type'].isin(model_type_filter.value)] | |
| # Add this line to display the results of both abilities by default | |
| df = df[df['Ability'].isin(ability_filter.value)] | |
| return df | |
| def get_all_df(dir=CSV_DIR): | |
| submission_repo = Repository(local_dir=SUBMISSION_NAME, clone_from=SUBMISSION_URL, use_auth_token=HF_TOKEN, repo_type="dataset") | |
| submission_repo.git_pull() | |
| df = pd.read_csv(dir) | |
| return df | |
| block = gr.Blocks() | |
| with block: | |
| gr.Markdown( | |
| LEADERBORAD_INTRODUCTION | |
| ) | |
| with gr.Tabs(elem_classes="tab-buttons") as tabs: | |
| # Table 0 | |
| with gr.TabItem("🏅 WorldScore Benchmark", elem_id="worldscore-tab-table", id=1): | |
| with gr.Column(): | |
| model_type_filter = gr.CheckboxGroup( | |
| choices=MODEL_TYPE, | |
| value=DEFAULT_MODEL_TYPE, | |
| label="Model Type", | |
| interactive=True | |
| ) | |
| ability_filter = gr.CheckboxGroup( | |
| choices=ABILITY, | |
| value=DEFAULT_ABILITY, | |
| label="Ability", | |
| interactive=True | |
| ) | |
| data_component = gr.components.Dataframe( | |
| column_widths="auto", # Automatically adjusts column widths | |
| value=get_baseline_df(), | |
| headers=COLUMN_NAMES, | |
| type="pandas", | |
| datatype=DATA_TITILE_TYPE, | |
| interactive=False, | |
| visible=True, | |
| ) | |
| def on_filter_change(model_types, abilities): | |
| df = get_baseline_df() | |
| # Filter by selected model types | |
| df = df[df['Model Type'].isin(model_types)] | |
| # Filter by selected abilities | |
| df = df[df['Ability'].isin(abilities)] | |
| return gr.Dataframe( | |
| column_widths="auto", # Automatically adjusts column widths | |
| value=df, | |
| headers=COLUMN_NAMES, | |
| type="pandas", | |
| datatype=DATA_TITILE_TYPE, | |
| interactive=False, | |
| visible=True | |
| ) | |
| model_type_filter.change( | |
| fn=on_filter_change, | |
| inputs=[model_type_filter, ability_filter], | |
| outputs=data_component | |
| ) | |
| ability_filter.change( | |
| fn=on_filter_change, | |
| inputs=[model_type_filter, ability_filter], | |
| outputs=data_component | |
| ) | |
| # with gr.TabItem("🚀 [I2V]Submit here! ", elem_id="mvbench-i2v-tab-table", id=7): | |
| # gr.Markdown(LEADERBORAD_INTRODUCTION, elem_classes="markdown-text") | |
| # with gr.Row(): | |
| # gr.Markdown(SUBMIT_INTRODUCTION, elem_classes="markdown-text") | |
| # with gr.Row(): | |
| # gr.Markdown("# ✉️✨ Submit your i2v model evaluation json file here!", elem_classes="markdown-text") | |
| # with gr.Row(): | |
| # gr.Markdown("Here is a required field", elem_classes="markdown-text") | |
| # with gr.Row(): | |
| # with gr.Column(): | |
| # model_name_textbox_i2v = gr.Textbox( | |
| # label="Model name", placeholder="Required field" | |
| # ) | |
| # revision_name_textbox_i2v = gr.Textbox( | |
| # label="Revision Model Name(Optional)", placeholder="If you need to update the previous results, please fill in this line" | |
| # ) | |
| # access_type_i2v = gr.Dropdown(["Open Source", "Ready to Open Source", "API", "Close"], label="Please select the way user can access your model. You can update the content by revision_name, or contact the VBench Team.") | |
| # with gr.Column(): | |
| # model_link_i2v = gr.Textbox( | |
| # label="Project Page/Paper Link/Github/HuggingFace Repo", placeholder="Required field. If filling in the wrong information, your results may be removed." | |
| # ) | |
| # team_name_i2v = gr.Textbox( | |
| # label="Your Team Name(If left blank, it will be user upload)", placeholder="User Upload" | |
| # ) | |
| # contact_email_i2v = gr.Textbox( | |
| # label="E-Mail(Will not be displayed)", placeholder="Required field" | |
| # ) | |
| # with gr.Row(): | |
| # gr.Markdown("The following is optional and will be synced to [GitHub] (https://github.com/Vchitect/VBench/tree/master/sampled_videos#what-are-the-details-of-the-video-generation-models)", elem_classes="markdown-text") | |
| # with gr.Row(): | |
| # release_time_i2v = gr.Textbox(label="Time of Publish", placeholder="1970-01-01") | |
| # model_resolution_i2v = gr.Textbox(label="resolution", placeholder="Width x Height") | |
| # model_fps_i2v = gr.Textbox(label="model fps", placeholder="FPS(int)") | |
| # model_frame_i2v = gr.Textbox(label="model frame count", placeholder="INT") | |
| # model_video_length_i2v = gr.Textbox(label="model video length", placeholder="float(2.0)") | |
| # model_checkpoint_i2v = gr.Textbox(label="model checkpoint", placeholder="optional") | |
| # model_commit_id_i2v = gr.Textbox(label="github commit id", placeholder='main') | |
| # model_video_format_i2v = gr.Textbox(label="pipeline format", placeholder='mp4') | |
| # with gr.Column(): | |
| # input_file_i2v = gr.components.File(label = "Click to Upload a ZIP File", file_count="single", type='binary') | |
| # submit_button_i2v = gr.Button("Submit Eval") | |
| # submit_succ_button_i2v = gr.Markdown("Submit Success! Please press refresh and retfurn to LeaderBoard!", visible=False) | |
| # fail_textbox_i2v = gr.Markdown('<span style="color:red;">Please ensure that the `Model Name`, `Project Page`, and `Email` are filled in correctly.</span>', elem_classes="markdown-text",visible=False) | |
| # submission_result_i2v = gr.Markdown() | |
| # # submit_button_i2v.click( | |
| # # add_new_eval_i2v, | |
| # # inputs = [ | |
| # # input_file_i2v, | |
| # # model_name_textbox_i2v, | |
| # # revision_name_textbox_i2v, | |
| # # model_link_i2v, | |
| # # team_name_i2v, | |
| # # contact_email_i2v, | |
| # # release_time_i2v, | |
| # # access_type_i2v, | |
| # # model_resolution_i2v, | |
| # # model_fps_i2v, | |
| # # model_frame_i2v, | |
| # # model_video_length_i2v, | |
| # # model_checkpoint_i2v, | |
| # # model_commit_id_i2v, | |
| # # model_video_format_i2v | |
| # # ], | |
| # # outputs=[submit_button_i2v, submit_succ_button_i2v, fail_textbox_i2v] | |
| # # ) | |
| # def refresh_data(): | |
| # value1 = get_baseline_df() | |
| # return value1 | |
| # with gr.Row(): | |
| # data_run = gr.Button("Refresh") | |
| # data_run.click(on_filter_model_size_method_change, inputs=[checkbox_group], outputs=data_component) | |
| block.launch() | |