import gradio as gr import pandas as pd import os os.environ['GRADIO_TEMP_DIR'] = "cache" from PIL import Image # 数据和图像的路径 # file_idx = 1 # 获取当前文件的绝对路径 current_dir = os.path.dirname(os.path.abspath(__file__)) image_dir = os.path.join(current_dir, f"mixed_images_1000") # image|description|ins_idx|instruction|type|answer|ans_t|score data_path = os.path.join(current_dir, "data_1000", "score_answer_full_one_stop_{}.csv") progress_df = None df = None progress_file = None # 读取图像的函数 def load_image(image_name): global image_dir image_path = os.path.join(image_dir, image_name) return Image.open(image_path) def get_next_instance(username): # 检查用户进度 user_progress = progress_df[progress_df["username"] == username] # 获取下一个未评估的图像 completed_images = user_progress["image"].unique() if not user_progress.empty else [] remaining_images = df[~df["image"].isin(completed_images)]["image"].unique() if len(remaining_images) == 0: return None, None, "所有数据已评估完毕！" current_image = remaining_images[0] filtered_df = df[df['image'] == current_image] # 构建QA对字典 # "username", "image", "description", "ins_idx", "instruction", "type", "answer", "score_precise", "score_normal", "score_creative" qa_pairs = [] for idx in range(18): instance = filtered_df[filtered_df['ins_idx'] == idx] question = instance['instruction'].values[0] precise_answer = instance[instance['ans_t'] == 'precise']['answer'].values[0] normal_answer = instance[instance['ans_t'] == 'normal']['answer'].values[0] creative_answer = instance[instance['ans_t'] == 'creative']['answer'].values[0] qa_pairs.append({ "image": current_image, "instruction": question, "ins_idx": idx, "type": instance['type'].values[0], "precise_answer": precise_answer, "normal_answer": normal_answer, "creative_answer": creative_answer, "score_precise": None, # 让用户填写 "score_normal": None, # 让用户填写 "score_creative": None # 让用户填写 }) return load_image(current_image), qa_pairs def display_content(name): # global current_image_index # current_image_index = 0 # 重置为第一个图像 name, file_idx = name.split("_") assert file_idx is not None, "请填写文件编号！" # 初始化进度文件，如果不存在则创建 global progress_df, progress_file, df, data_path data_path = data_path.format(file_idx) df = pd.read_csv(data_path, sep="|") progress_file = os.path.join(current_dir, "val_score", f"ser_progress_{file_idx}.csv") if not os.path.exists(progress_file): # image|description|ins_idx|instruction|type|answer|ans_t|score progress_df = pd.DataFrame(columns=["username", "image", "description", "ins_idx", "instruction", "type", "answer", "score_precise", "score_normal", "score_creative", "score_ins"]) progress_df.to_csv(progress_file, index=False) else: progress_df = pd.read_csv(progress_file) image, qa_pairs = get_next_instance(name) # content = image_question_list[current_image_index] # 用户名 + 进度条 user_progress = progress_df[progress_df["username"] == name] user_completed_images = user_progress["image"].unique() if not user_progress.empty else [] completed_images = progress_df["image"].unique() if not progress_df.empty else [] remaining_images_num = len(df['image'].unique()) - len(completed_images) custom_text = f"欢迎, {name}！请选择以下问题的答案：（进度：完成{len(user_completed_images)}//剩余{remaining_images_num}//总{len(df['image'].unique())}）" # 初始化所有问题的标签和选项 question_labels = [f"{i}. {qa_pairs[i]['type']}{qa_pairs[i]['instruction']}" for i in range(18)] question_updates = [] for i in range(18): if i == 14 or i == 17: question_updates.append(gr.update(label=f"precise: {qa_pairs[i]['precise_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"normal: {qa_pairs[i]['normal_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"creative: {qa_pairs[i]['creative_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label="===========❤该问题是否保留？===========", choices=["2:保留但需重写", "3:保留"], value=None)) else: question_updates.append(gr.update(label=f"precise: {qa_pairs[i]['precise_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"normal: {qa_pairs[i]['normal_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"creative: {qa_pairs[i]['creative_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label="===========❤该问题是否保留？===========", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) return [custom_text, gr.update(visible=False), gr.update(visible=False), image, gr.update(visible=True)] + question_labels + question_updates def next_image(name, *selected_options): name, file_idx = name.split("_") # 获取当前图像和相关QA对 _, qa_pairs = get_next_instance(name) # "username", "image", "description", "ins_idx", "instruction", "type", # "answers", "score_precise", "score_normal", "score_creative" def convert_score(ori_text): if ori_text == "1:丢弃": return 1 elif ori_text == "2:保留但需重写": return 2 elif ori_text == "3:保留": return 3 else: return 0 # 保存用户评分数据 global progress_df, progress_file new_rows = [] for i in range(18): # 保存评分结果 new_rows.append({ "username": name, "image": qa_pairs[i]["image"], "description": "待完成", "ins_idx": qa_pairs[i]["ins_idx"], "instruction": qa_pairs[i]["instruction"], "type": qa_pairs[i]["type"], "answers": "待完成", "score_precise": convert_score(selected_options[i*4]), "score_normal": convert_score(selected_options[i*4+1]), "score_creative": convert_score(selected_options[i*4+2]), "score_ins": convert_score(selected_options[i*4+3]) }) progress_df = pd.concat([progress_df, pd.DataFrame(new_rows)], ignore_index=True) # 如果处理完所有图像，保存结果 print("正在保存数据...") progress_df.to_csv(progress_file, index=False) print(f"数据已保存到 {progress_file}") # 切换到下一张图像 image, qa_pairs = get_next_instance(name) if image is None: return [None, "所有数据已评估完毕！"] + [None] * 90 # 用户名 + 进度条 user_progress = progress_df[progress_df["username"] == name] user_completed_images = user_progress["image"].unique() if not user_progress.empty else [] completed_images = progress_df["image"].unique() if not progress_df.empty else [] remaining_images_num = len(df['image'].unique()) - len(completed_images) custom_text = f"欢迎, {name}！请选择以下问题的答案：（进度：完成{len(user_completed_images)}//剩余{remaining_images_num}//总{len(df['image'].unique())}） 评分数据：{progress_file}）" question_labels = [f"{i}. {qa_pairs[i]['type']}{qa_pairs[i]['instruction']}" for i in range(18)] question_updates = [] for i in range(18): if i == 14 or i == 17: question_updates.append(gr.update(label=f"precise: {qa_pairs[i]['precise_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"normal: {qa_pairs[i]['normal_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"creative: {qa_pairs[i]['creative_answer']}", choices=["2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label="===========❤该问题是否保留？===========", choices=["2:保留但需重写", "3:保留"], value=None)) else: question_updates.append(gr.update(label=f"precise: {qa_pairs[i]['precise_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"normal: {qa_pairs[i]['normal_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label=f"creative: {qa_pairs[i]['creative_answer']}", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) question_updates.append(gr.update(label="===========❤该问题是否保留？===========", choices=["1:丢弃", "2:保留但需重写", "3:保留"], value=None)) return [image, custom_text] + question_labels + question_updates def validate_input(*selected_options): # 展示选择的答案 # [a]: (0)-1, (1)-1, (2)-2,...., (6)-1 # [b]: (7)-1, (8)-1, (9)-2,...., (13)-1 # [c]: (14)-1 # [d]: (15)-1, (16)-1, (17)-1 ans_a = [f"({i})-{selected_options[i*4+3]}" for i in range(7)] ans_b = [f"({i})-{selected_options[i*4+3]}" for i in range(7, 14)] ans_c = [f"({i})-{selected_options[i*4+3]}" for i in range(14, 15)] ans_d = [f"({i})-{selected_options[i*4+3]}" for i in range(15, 18)] msg_ans = f""" 答案： [a]: {' '.join(ans_a)} [b]: {' '.join(ans_b)} [c]: {' '.join(ans_c)} [d]: {' '.join(ans_d)}""" # 验证用户输入的答案是否完整 for i in range(18): # if None in selected_options[i*4:i*4+4]: # return False, f"请填写所有问题的答案！第{i}个问题未填写！" if selected_options[i*4+3] is None: return False, f"请填写所有问题的答案！第{i}个问题未填写！\n{msg_ans}" a_num, b_num = 0, 0 for i in range(18): if i < 7 and selected_options[i*4+3] != "1:丢弃": a_num += 1 if 7 <= i < 14 and selected_options[i*4+3] != "1:丢弃": b_num += 1 if a_num != 2 or b_num != 2: return False, f"请保留两个[a]类问题和两个[b]类问题！([a]:{a_num}, [b]:{b_num})\n{msg_ans}" if (selected_options[15*4+3] == "1:丢弃" and selected_options[16*4+3] == "1:丢弃") or \ (selected_options[15*4+3] != "1:丢弃" and selected_options[16*4+3] != "1:丢弃"): return False, f"15,16问题中保留一个！\n{msg_ans}" return True, msg_ans # 交互函数，用于更新按钮状态和错误消息 def update_button_and_error_message(*selected_options): is_valid, error_message = validate_input(*selected_options) # return gr.update(visible=not is_valid), error_message return gr.update(interactive=is_valid), gr.update(value=error_message, visible=not is_valid) # 创建Gradio界面 with gr.Blocks() as demo: output_text = gr.Markdown() name_input = gr.Textbox(label="请输入你的名字,严格按照格式'姓名_文件编号'填写,如‘ysm_1’,目前只评估0和9", placeholder="姓名_文件编号,如‘ysm_1’") submit_button = gr.Button("确定") output_image = gr.Image() # 动态生成18个问题，每个问题包含4个子问题 option_radios = [] question_labels = [] for i in range(18): question_label = gr.Markdown(f"**问题**") question_labels.append(question_label) with gr.Column() as col: for j in range(4): radio = gr.Radio(choices=[], label=f"答案评分") option_radios.append(radio) next_button = gr.Button("下一张图像", visible=False) error_message = gr.Markdown(visible=False) # 提交按钮状态和错误消息绑定到验证函数 option_radios_values = [radio for radio in option_radios] for radio in option_radios_values: radio.change(update_button_and_error_message, inputs=option_radios_values, outputs=[next_button, error_message]) # 显示内容初始化，包括文本、图像和18个问题 submit_button.click(display_content, inputs=name_input, outputs=[output_text, name_input, submit_button, output_image, next_button] + question_labels + option_radios) # 下一张图像按钮，切换图像和问题 next_button.click(next_image, inputs=[name_input] + option_radios, outputs=[output_image, output_text] + question_labels + option_radios) demo.launch(share=True, debug=True)