""" Description : analyze the output from autoline mode. Author : Ruidi Qiu (r.qiu@tum.de) Time : 2023/12/12 17:35:00 LastEdited : 2024/9/17 23:35:03 """ import loader_saver as ls import utils.utils as utils from data.probset import dictlist, HDLBitsProbset, muti_dictlist from LLM_call import PRICING_MODELS import os import math LOOSE_FACTOR = 0.8 DEFAULT_SAVING_DIR = "analysis" DEFAULT_LOG_NAME = "analyze_out.log" DEFAULT_LOG_PATH = os.path.join(DEFAULT_SAVING_DIR, DEFAULT_LOG_NAME) K_LIST = [1] # insert your Chatbench_RunInfo.json's path here to re-analyze the data, then run this file directly. the result will be saved in analysis/analyze_out.log CHATBENCH_RUNINFO_PATH = "saves_inEDA/DATE25/Main_Results/CorrectBench/disc_70wrong_25correct_20240831_181427/Chatbench_RunInfo.json" # this is for multiple directories' analysis, insert the directory path here and change the main function to regular_multiA_main, then run the file directly. the result will be saved in analysis/analyze_out.log MULTI_DIR = "saves_inEDA/DATE25/Main_Results/CorrectBench" def main(): # diy_main() regular_main() # regular_multiA_main() def diy_main(): # find the passed with correction Chatbench_RunInfo = ls.load_json_dict(CHATBENCH_RUNINFO_PATH) # analyze the data analyzer = Analyzer(Chatbench_RunInfo) analyzer.run() for i in analyzer.data: if i.get("TB_corrected", False) and analyzer.Eval2_pass(i): print(i["task_id"]) # save the result to a txt file # with open(DEFAULT_LOG_PATH, "w") as f: # f.write(analyzer.messages) # # also, write the current time # f.write("analysis time: %s\n" % (utils.get_time())) # with open(DEFAULT_LOG_PATH, "w") as f: # f.write("analysis time: %s\n" % (utils.get_time())) # # task_eval2passtimes_analyze() # full_analyzer = Analyzer(ls.load_json_dict(CHATBENCH_RUNINFO_PATH)) # full_analyzer.run() # with open(DEFAULT_LOG_PATH, "a") as f: # f.write(full_analyzer.messages) # analyze_subset(HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "CMB"}), CHATBENCH_RUNINFO_PATH, "full CMB") # analyze_subset(HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "SEQ"}), CHATBENCH_RUNINFO_PATH, "full SEQ") # analyze_subset("data/HDLBits/HDLBits_data_CMB15.jsonl", CHATBENCH_RUNINFO_PATH, "CMB15") # analyze_subset("data/HDLBits/HDLBits_data_SEQ15.jsonl", CHATBENCH_RUNINFO_PATH, "SEQ15") k_list = [1] # Eval_scenchecks = ["Eval0_scencheck", "Eval0_noscencheck", "Eval1_scencheck", "Eval1_noscencheck", "Eval2_scencheck", "Eval2_noscencheck"] # Eval_scenchecks = ["Eval2b", "Eval2"] multi_analyzer = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer.exclude_debug = True multi_analyzer.messages += "\n#################### TOTAL ####################\n" multi_analyzer.run() multi_analyzer.get_avg_tokens_one_task() multi_analyzer.get_avg_pass_by_disc_and_corr() multi_analyzer.save() CMB_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "CMB"}) SEQ_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "SEQ"}) CMB_tasks = CMB_set.task_id_list SEQ_tasks = SEQ_set.task_id_list multi_analyzer_CMB = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer_CMB.exclude_debug = True multi_analyzer_CMB.del_items(SEQ_tasks, del_by_list=True) # print(multi_analyzer_CMB.access("total_num")) multi_analyzer_CMB.messages += "\n#################### CMB ####################\n" multi_analyzer_CMB.run() multi_analyzer_CMB.get_avg_tokens_one_task() multi_analyzer_CMB.get_avg_pass_by_disc_and_corr() multi_analyzer_CMB.save(os.path.join(DEFAULT_SAVING_DIR, "CMB_" + DEFAULT_LOG_NAME)) multi_analyzer_SEQ = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer_SEQ.exclude_debug = True multi_analyzer_SEQ.del_items(CMB_tasks, del_by_list=True) multi_analyzer_SEQ.messages += "\n#################### SEQ ####################\n" multi_analyzer_SEQ.run() multi_analyzer_SEQ.get_avg_tokens_one_task() multi_analyzer_SEQ.get_avg_pass_by_disc_and_corr() multi_analyzer_SEQ.save(os.path.join(DEFAULT_SAVING_DIR, "SEQ_" + DEFAULT_LOG_NAME)) # multi_analyzer_SEQ.save(os.path.join(DEFAULT_SAVING_DIR, "SEQ_" + DEFAULT_LOG_NAME)) # show the pass num for seq15 of each try # multi_analyzer = MultiAnalyzer(MULTI_DIR) # multi_analyzer.exclude_debug = True # SEQ15_set = HDLBitsProbset("data/HDLBits/HDLBits_data_SEQ15.jsonl") # SEQ15_tasks = SEQ15_set.task_id_list # multi_analyzer.del_items(SEQ15_tasks, del_by_list=False) # multi_analyzer.renew_result_dict() # multi_analyzer.run() # print(multi_analyzer.access("total_num")) # print(multi_analyzer.access("fullpass_num_nodebug")) # print(sum(multi_analyzer.access("fullpass_num_nodebug"))) # for task in multi_analyzer.result_dict.data: # print(task["task_id"] + ": " + str(round(task["Eval2_pass_at_1"], 2))) # k_list = [1,5,10] # multi_analyzer = MultiAnalyzer(MULTI_DIR, k_list) # data_list = multi_analyzer.dictlists # CMB_num_list = [] # for data in data_list: # CMB_num = 0 # for i in data.data: # if i.get("circuit_type", "NO data") == "CMB": # CMB_num += 1 # CMB_num_list.append(CMB_num) # print(CMB_num_list) # k_list = [1,5,10] # multi_analyzer = MultiAnalyzer(MULTI_DIR, k_list) # # multi_analyzer.exclude_debug = True # multi_analyzer.messages += "\n#################### TOTAL ####################\n" # multi_analyzer.run() # multi_analyzer.save() # CMB_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "CMB"}) # SEQ_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "SEQ"}) # CMB_tasks = CMB_set.task_id_list # SEQ_tasks = SEQ_set.task_id_list # multi_analyzer_CMB = MultiAnalyzer(MULTI_DIR, k_list) # # multi_analyzer_CMB.exclude_debug = True # multi_analyzer_CMB.del_items(SEQ_tasks, del_by_list=True) # # print(multi_analyzer_CMB.access("total_num")) # multi_analyzer_CMB.messages += "\n#################### CMB ####################\n" # multi_analyzer_CMB.run() # multi_analyzer_CMB.save(os.path.join(DEFAULT_SAVING_DIR, "CMB_" + DEFAULT_LOG_NAME)) # multi_analyzer_SEQ = MultiAnalyzer(MULTI_DIR, k_list) # # multi_analyzer_SEQ.exclude_debug = True # multi_analyzer_SEQ.del_items(CMB_tasks, del_by_list=True) # multi_analyzer_SEQ.messages += "\n#################### SEQ ####################\n" # multi_analyzer_SEQ.run() # multi_analyzer_SEQ.save(os.path.join(DEFAULT_SAVING_DIR, "SEQ_" + DEFAULT_LOG_NAME)) # k_list = [1,3,5,10] # multi_analyzer = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer.exclude_debug = True # multi_analyzer.messages += "\n#################### TOTAL ####################\n" # multi_analyzer.run() # multi_analyzer.save() # CMB_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "CMB"}) # SEQ_set = HDLBitsProbset("data/HDLBits/HDLBits_circuit_type.jsonl", filter_content={"circuit_type": "SEQ"}) # CMB_tasks = CMB_set.task_id_list # SEQ_tasks = SEQ_set.task_id_list # multi_analyzer_CMB = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer_CMB.del_items(SEQ_tasks, del_by_list=True) # multi_analyzer_CMB.exclude_debug = True # multi_analyzer_CMB.messages += "\n#################### CMB ####################\n" # multi_analyzer_CMB.run() # multi_analyzer_CMB.save(os.path.join(DEFAULT_SAVING_DIR, "CMB_" + DEFAULT_LOG_NAME)) # multi_analyzer_SEQ = MultiAnalyzer(MULTI_DIR, k_list) # multi_analyzer_SEQ.del_items(CMB_tasks, del_by_list=True) # multi_analyzer_SEQ.exclude_debug = True # multi_analyzer_SEQ.messages += "\n#################### SEQ ####################\n" # multi_analyzer_SEQ.run() # multi_analyzer_SEQ.save(os.path.join(DEFAULT_SAVING_DIR, "SEQ_" + DEFAULT_LOG_NAME)) # Chatbench_RunInfo = ls.load_json_dict(CHATBENCH_RUNINFO_PATH) # analyzer = Analyzer(Chatbench_RunInfo) # # task 2024/04/25 17:09:39, extract SEQ15 info from SEQ # Chatbench_RunInfo = ls.load_json_dict(CHATBENCH_RUNINFO_PATH) # seq15 = HDLBitsProbset("data/HDLBits/HDLBits_data_SEQ15.jsonl") # seq15_taskids = seq15.task_id_list # analyzer = Analyzer(Chatbench_RunInfo) # analyzer.del_items(seq15_taskids, False) # # analyzer.filter({"debug_iter_iv": 0}) # analyzer.run() # with open(DEFAULT_LOG_PATH, "w") as f: # f.write(analyzer.messages) # # also, write the current time # f.write("analysis time: %s\n" % (utils.get_time())) # Chatbench_RunInfo = ls.load_json_dict(CHATBENCH_RUNINFO_PATH) # # analyze the data # analyzer = Analyzer(Chatbench_RunInfo) # analyzer.filter({"debug_iter_iv": 0}) # analyzer.run() # # save the result to a txt file # with open(DEFAULT_LOG_PATH, "w") as f: # f.write(analyzer.messages) # # also, write the current time # f.write("analysis time: %s\n" % (utils.get_time())) pass def regular_multiA_main(): multi_analyzer = MultiAnalyzer(MULTI_DIR) multi_analyzer.run() multi_analyzer.save() def regular_main(): Chatbench_RunInfo = ls.load_json_dict(CHATBENCH_RUNINFO_PATH) # analyze the data analyzer = Analyzer(Chatbench_RunInfo) analyzer.run() # save the result to a txt file with open(DEFAULT_LOG_PATH, "w") as f: f.write(analyzer.messages) # also, write the current time f.write("analysis time: %s\n" % (utils.get_time())) def analyze_subset(subset:str|HDLBitsProbset, runinfo_path, subset_name=""): """ this function is used to only analyze a subset of the runinfo data - subset (only the task_ids are needed): - str: path of the subset - HDLBitsProbset: the subset - runinfo_path: path of the Chatbench_RunInfo.json - subset_name: the name of the subset """ if isinstance(subset, str): # path, load the subset subset = HDLBitsProbset(subset) elif isinstance(subset, HDLBitsProbset): subset = subset else: raise TypeError("subset should be a path or a HDLBitsProbset") subset_tasks = subset.task_id_list analyzer = Analyzer(ls.load_json_dict(runinfo_path)) analyzer.del_items(subset_tasks, False) analyzer.out_txt += f"\n#################### {subset_name} ####################\n" analyzer.run() with open(DEFAULT_LOG_PATH, "a") as f: f.write(analyzer.messages) f.write("analysis time: %s\n" % (utils.get_time())) class Analyzer(HDLBitsProbset): def __init__(self, Chatbench_RunInfo, pricing_model="gpt-4o-2024-05-13"): super().__init__() self.data = Chatbench_RunInfo self.check_existance() self.pricing_model = pricing_model self.out_txt = "" self.loose_factor = LOOSE_FACTOR def run(self): self.out_txt += "\n########## Analyze of Chatbench_RunInfo ##########\n" self.out_txt += "\n#### pass numbers:\n" if self.Eval2b_exist: self.out_txt += "Eval2b: %d\n" % self.Eval2bpass_num self.out_txt += "Eval2 : %d\n" % self.fullpass_num self.out_txt += "Eval1 : %d\n" % self.Eval1pass_num self.out_txt += "Eval0 : %d\n" % self.Eval0pass_num self.out_txt += "total : %d " % self.total_num self.out_txt += "(Failed: %d)\n" % (self.total_num - self.Eval0pass_num) if self.reboot_times_exist: self.out_txt += "passed TB by autoline reboot action (from TB3_check): %d\n" % self.autoline_reboot_task_num if self.TB_corrected_exist: self.out_txt += "\npassed TB by functional corrector: %d\n" % self.corrected_num # === [新增] CGA Coverage 统计 === # 计算平均覆盖率 total_coverage = 0.0 max_coverage = 0.0 min_coverage = 100.0 has_cov_data = False for task in self.data: cov = task.get("coverage", 0.0) if cov > 0: has_cov_data = True total_coverage += cov if cov > max_coverage: max_coverage = cov if cov < min_coverage: min_coverage = cov avg_cov = total_coverage / self.total_num if self.total_num > 0 else 0.0 self.out_txt += "\n#### CGA Coverage Info:\n" self.out_txt += "Average Coverage : %.2f%%\n" % avg_cov if has_cov_data: self.out_txt += "Max Coverage : %.2f%%\n" % max_coverage self.out_txt += "Min Coverage : %.2f%%\n" % min_coverage else: self.out_txt += "(No coverage data found in JSON)\n" # =============================== # self.out_txt += self.get_avg_debug_iter_on_sim_pass_with_debug()[-1] # self.out_txt += self.get_debug_failed_num()[-1] # self.out_txt += self.get_debug_total_pass_num()[-1] # self.out_txt += self.get_debug_sim_pass_num()[-1] self.out_txt += "\n#### tokens and cost:\n" # self.out_txt += "average prompt tokens: %d\naverage completion tokens: %d\n" % (self.prompt_tokens_num/self.total_num, self.completion_tokens_num/self.total_num) self.out_txt += "average prompt tokens: %d\n" % (self.prompt_tokens_num / self.total_num) self.out_txt += "average completion tokens: %d\n" % (self.completion_tokens_num / self.total_num) self.out_txt += "total cost: %.4f\n" % self.cost self.out_txt += "average cost: %.4f\n" % self.avg_cost self.out_txt += "\n#### time:\n" self.out_txt += "average time: %.2fs\n" % self.avg_time self.out_txt += "\n#### debug info table:\n" self.out_txt += self.get_debug_infotable() self.out_txt += "\n#### Eval2 ratio:\n" self.out_txt += self.get_eval2_ratio_each_problem() # === [新增] 每个任务的覆盖率详情 === self.out_txt += "\n#### CGA Coverage Detail List:\n" self.out_txt += f"{'Task ID':<25} | {'Coverage':<10}\n" self.out_txt += "-" * 40 + "\n" for task in self.data: tid = task.get("task_id", "Unknown") cov = task.get("coverage", 0.0) self.out_txt += f"{tid:<25} | {cov:.2f}%\n" # ================================== if self.Eval2b_exist: self.out_txt += "\n#### Eval2b ratio:\n" self.out_txt += self.get_eval2b_ratio_each_problem() # self.get_iv_runing_time_info() self.out_txt += "\nloose Eval2 pass metric applied: %s\n\n" % self.loose_factor # 将生成的 out_txt 同步给 messages 属性，保证 save 时能写入文件 def find_fake_eval0pass(self): self.filter({"sim_pass": 1}) task_ids_fake_eval0pass = [] for i in self.data: if i.get("Eval1_pass","NO data") == "NO data": task_ids_fake_eval0pass.append(i["task_id"]) self.out_txt += "fake Eval0 pass: %d\n" % len(task_ids_fake_eval0pass) for i in task_ids_fake_eval0pass: self.out_txt += i + "\n" def check_existance(self): self.Eval2b_exist = False self.TB_corrected_exist = False self.reboot_times_exist = False for i in self.data: if "Eval2b_pass" in i.keys(): self.Eval2b_exist = True if "TB_corrected" in i.keys(): self.TB_corrected_exist = True if "reboot_times" in i.keys(): self.reboot_times_exist = True if self.Eval2b_exist and self.TB_corrected_exist and self.reboot_times_exist: break # task def draw_Eval2_histogram(self, figurename="eval2_histogram.png"): import matplotlib.pyplot as plt import matplotlib matplotlib.use('Agg') ratios = [] for i in self.data: if self.Eval0_pass(i) and self.Eval1_pass(i): # get the numerical ratio ratio_str = i.get("Eval2_ratio", None) if ratio_str is None: continue numerator, denominator = ratio_str.split("/") ratio = float(numerator) / float(denominator) ratios.append(ratio) # draw histogram of ratios, it will have 10 bars, 0~10%, 10~20%, ..., 90~100% plt.hist(ratios, bins=10, range=(0,1)) # save to analysis/eval2_histogram.png plt.savefig(os.path.join(DEFAULT_SAVING_DIR, figurename)) plt.close() @property def messages(self): return self.out_txt @property def total_num(self): if not hasattr(self, "_total_num"): self._total_num = len(self.data) return self._total_num @property def fullpass_num(self): if not hasattr(self, "_fullpass_num"): self._fullpass_num = 0 for i in self.data: if self.Eval0_pass(i) and i.get("Eval1_pass",0) and self.Eval2_pass(i): self._fullpass_num += 1 return self._fullpass_num @property def fullpass_num_nodebug(self): if not hasattr(self, "_fullpass_num"): self._fullpass_num = 0 for i in self.data: if self.Eval0_pass(i) and i.get("Eval1_pass",0) and self.Eval2_pass(i) and self.debug_iter(i) == 0: self._fullpass_num += 1 return self._fullpass_num @property def Eval2bpass_num(self): if not hasattr(self, "_Eval2bpass_num"): self._Eval2bpass_num = 0 if self.Eval2b_exist: for i in self.data: if self.Eval0_pass(i) and i.get("Eval1_pass",0) and i.get("Eval2b_pass",0): self._Eval2bpass_num += 1 return self._Eval2bpass_num @property def Eval0pass_num(self): if not hasattr(self, "_Eval0pass_num"): self._Eval0pass_num = 0 for i in self.data: if self.Eval0_pass(i): self._Eval0pass_num += 1 return self._Eval0pass_num @property def Eval1pass_num(self): if not hasattr(self, "_Eval1pass_num"): self._Eval1pass_num = 0 for i in self.data: if self.Eval0_pass(i) and i.get("Eval1_pass",0): self._Eval1pass_num += 1 return self._Eval1pass_num @property def corrected_num(self): if not hasattr(self, "_corrected_num"): self._corrected_num = 0 if self.TB_corrected_exist: for i in self.data: if i.get("TB_corrected",0) and self.Eval2_pass(i): self._corrected_num += 1 return self._corrected_num @property def autoline_reboot_task_num(self): if not hasattr(self, "_autoline_reboot_task_num"): self._autoline_reboot_task_num = 0 if self.reboot_times_exist: for i in self.data: if (i.get("reboot_times",0) > 0) and self.Eval2_pass(i): self._autoline_reboot_task_num += 1 return self._autoline_reboot_task_num @property def avg_time(self): if not hasattr(self, "_avg_time"): time_sum = 0 for i in self.data: time_sum += i.get("time",0) self._avg_time = time_sum / len(self.data) return self._avg_time @property def tokens_num(self): if not hasattr(self, "_tokens_num"): prompt_tokens_sum = 0 completion_tokens_sum = 0 for i in self.data: prompt_tokens_sum += i.get("prompt_tokens",0) completion_tokens_sum += i.get("completion_tokens",0) self._prompt_tokens_num = prompt_tokens_sum self._completion_tokens_num = completion_tokens_sum self._tokens_num = prompt_tokens_sum + completion_tokens_sum return self._tokens_num @property def prompt_tokens_num(self): if not hasattr(self, "_prompt_tokens_num"): self.tokens_num return self._prompt_tokens_num @property def completion_tokens_num(self): if not hasattr(self, "_completion_tokens_num"): self.tokens_num return self._completion_tokens_num @property def avg_tokens(self): if not hasattr(self, "_avg_tokens"): self._avg_tokens = self.tokens_num / self.total_num return self._avg_tokens @property def cost(self): if not hasattr(self, "_cost"): self._cost = self.get_total_cost() return self._cost @property def avg_cost(self): if not hasattr(self, "_avg_cost"): self._avg_cost = self.cost / self.total_num return self._avg_cost def get_total_cost(self): """ return the average cost of the data """ prompt_cost_perk, completion_cost_perk = PRICING_MODELS[self.pricing_model] prompt_cost = self.prompt_tokens_num * prompt_cost_perk / 1000 completion_cost = self.completion_tokens_num * completion_cost_perk / 1000 total_cost = prompt_cost + completion_cost return total_cost def get_eval2_ratio_each_problem(self): """ return the ratio of the second evaluation """ txt_out = "" for i in self.data: if self.Eval0_pass(i) and i.get("Eval1_pass",0): task_id = i["task_id"] eval2_ratio = i.get("Eval2_ratio", "No Eval2 ratio data") txt_out += "%s: %s\n" % (task_id, eval2_ratio) return txt_out def get_eval2b_ratio_each_problem(self): """ return the ratio of the second evaluation """ txt_out = "" for i in self.data: if self.Eval2b_exist: if self.Eval0_pass(i) and i.get("Eval1_pass",0): task_id = i["task_id"] eval2_ratio = i.get("Eval2b_ratio", "No Eval2b ratio data") txt_out += "%s: %s\n" % (task_id, eval2_ratio) else: txt_out = "No Eval2b data" return txt_out def get_debug_infotable(self): """ return the debug info table: | un-debugged | debugged | total | failed | - | 2 | 2 | Eval0 | 3 | 5 | 8 | Eval1 | 2 | 2 | 4 | Eval2 | 1 | 0 | 1 | if have Eval2b: Eval2b | 1 | 0 | 1 | """ txt_out = "" # debugged but failed failed_debugged_num = 0 failed_undebugged_num = 0 Eval0_debugged_num, Eval1_debugged_num, Eval2_debugged_num, Eval2b_debugged_num = 0, 0, 0, 0 Eval0_undebugged_num, Eval1_undebugged_num, Eval2_undebugged_num, Eval2b_undebugged_num = 0, 0, 0, 0 if self.reboot_times_exist or self.TB_corrected_exist: mode = "funcdebug" else: mode = "syndebug" for i in self.data: if mode == "syndebug": debugged = (self.debug_iter(i) != 0) elif mode == "funcdebug": debugged = (i.get("reboot_times", 0) > 0 or (i.get("TB_corrected", False))) else: raise ValueError("mode should be 'syndebug' or 'funcdebug'") failed_debugged_num += 1 if not self.Eval0_pass(i) and debugged else 0 failed_undebugged_num += 1 if not self.Eval0_pass(i) and (not debugged) else 0 Eval0_debugged_num += 1 if self.Eval0_pass(i) and debugged else 0 Eval0_undebugged_num += 1 if self.Eval0_pass(i) and (not debugged) else 0 Eval1_debugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and debugged else 0 Eval1_undebugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and (not debugged) else 0 Eval2_debugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and self.Eval2_pass(i) and debugged else 0 Eval2_undebugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and self.Eval2_pass(i) and (not debugged) else 0 if self.Eval2b_exist: Eval2b_debugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and self.Eval2b_pass(i) and debugged else 0 Eval2b_undebugged_num += 1 if self.Eval0_pass(i) and i.get("Eval1_pass", 0) and self.Eval2b_pass(i) and (not debugged) else 0 failed_num = failed_debugged_num + failed_undebugged_num Eval0_num = Eval0_debugged_num + Eval0_undebugged_num Eval1_num = Eval1_debugged_num + Eval1_undebugged_num Eval2_num = Eval2_debugged_num + Eval2_undebugged_num if self.Eval2b_exist: Eval2b_num = Eval2b_debugged_num + Eval2b_undebugged_num # make a table; each cell should have a width of 11 txt_out += ("SYNTACTIC" if mode == "syndebug" else "FUNCTIONAL") + " debug info table:\n" txt_out += "(debugged here means " + ("syntactic debugging" if mode == "syndebug" else "functional debugging") + ")\n" if mode == "syndebug": txt_out += " | un-synt-debugged | synt-debugged | total |\n" elif mode == "funcdebug": txt_out += " | un-func-debugged | func-debugged | total |\n" txt_out += "failed | %16d | %13d | %5d |\n" % (failed_undebugged_num, failed_debugged_num, failed_num) txt_out += "Eval0 | %16d | %13d | %5d |\n" % (Eval0_undebugged_num, Eval0_debugged_num, Eval0_num) txt_out += "Eval1 | %16d | %13d | %5d |\n" % (Eval1_undebugged_num, Eval1_debugged_num, Eval1_num) txt_out += "Eval2 | %16d | %13d | %5d |\n" % (Eval2_undebugged_num, Eval2_debugged_num, Eval2_num) if self.Eval2b_exist: txt_out += "Eval2b | %16d | %13d | %5d |\n" % (Eval2b_undebugged_num, Eval2b_debugged_num, Eval2b_num) return txt_out def get_iv_runing_time_info(self): max_time = 0.0 min_time = 0.0 total_time = 0.0 cnt = 0 for i in self.data: if self.Eval0_pass(i): time = float(i.get("iv_runing_time", 0.0)) if (time > max_time) or (max_time == 0.0): max_time = time if (time < min_time) or (min_time == 0.0): min_time = time total_time += time cnt += 1 avg_time = total_time / cnt if cnt != 0 else 0.0 if cnt != 0: self.out_txt += "\n#### iv_runing_time info:\n" self.out_txt += "avg_time: %.2fs\n" % avg_time self.out_txt += "max_time: %.2fs\n" % max_time self.out_txt += "min_time: %.2fs\n" % min_time def Eval0_pass(self, data): if "Eval0_pass" in data.keys(): return data["Eval0_pass"] # latest version elif "sim_pass" in data.keys(): return data["sim_pass"] # old version else: return False def Eval1_pass(self, data): return data.get("Eval1_pass", False) def Eval2_pass(self, data): """check if one data pass the Eval 2""" # we use this to compensate special cases: m2014_q3 if data["task_id"] == "m2014_q3": if data.get("Eval2_failed_mutant_idxes", []) == [3,4,7,8,9,10] or self.loose_Eval2_pass(data): return True else: return False # normal cases else: return self.loose_Eval2_pass(data) def Eval0_scencheck_pass(self, data): return self.Eval0_pass(data) and data.get("checklist_worked", False) def Eval1_scencheck_pass(self, data): return self.Eval1_pass(data) and data.get("checklist_worked", False) def Eval2_scencheck_pass(self, data): return self.Eval2_pass(data) and data.get("checklist_worked", False) def Eval0_noscencheck_pass(self, data): return self.Eval0_pass(data) and (not data.get("checklist_worked", False)) def Eval1_noscencheck_pass(self, data): return self.Eval1_pass(data) and (not data.get("checklist_worked", False)) def Eval2_noscencheck_pass(self, data): return self.Eval2_pass(data) and (not data.get("checklist_worked", False)) def Eval0_nodebug_pass(self, data): return (self.Eval0_pass(data)) and (self.debug_iter(data) == 0) def Eval1_nodebug_pass(self, data): return (self.Eval1_pass(data)) and (self.debug_iter(data) == 0) def Eval2_nodebug_pass(self, data): return (self.Eval2_pass(data)) and (self.debug_iter(data) == 0) def Eval2b_pass(self, data): """check if one data pass the Eval 2""" # we use this to compensate special cases: m2014_q3 if data["task_id"] == "m2014_q3": if self.Eval2_pass(data): return True else: return False # normal cases else: return self.loose_Eval2b_pass(data) def debug_iter(self, data): if data.get("debug_iter", None) is not None: return data["debug_iter"] else: return data.get("debug_iter_iv", 0) + data.get("debug_iter_py", 0) def loose_Eval2_pass(self, data): """pass for 9/10, 8/10 and 4/5""" if data.get("Eval2_pass", False): return True ratio_str = data.get("Eval2_ratio", None) if ratio_str is None: return False numerator, denominator = ratio_str.split("/") numerator, denominator = int(numerator), int(denominator) # if int(numerator) + 1 >= int(denominator): if float(numerator) / float(denominator) >= self.loose_factor: return True else: return False def loose_Eval2b_pass(self, data): """pass for 9/10, 8/10 and 4/5""" if data.get("Eval2b_pass", False): return True ratio_str = data.get("Eval2b_ratio", None) if ratio_str is None: return False numerator, denominator = ratio_str.split("/") numerator, denominator = int(numerator), int(denominator) # if int(numerator) + 1 >= int(denominator): if float(numerator) / float(denominator) >= self.loose_factor: return True else: return False class MultiAnalyzer(muti_dictlist): def __init__(self, group_dir:str=None, pass_at_k_kvalues = K_LIST): """ group_dir: includes many subdirs, each subdir contains a Chatbench_RunInfo.json """ super().__init__(id_key="task_id") self.runinfo_paths = [] self.result = {} # include the final results self.pass_at_k_kvalues = pass_at_k_kvalues self.exclude_debug = False # this is uesd in baseline when analyzing the data without debug self.messages = "" self.group_dir = group_dir for subdir in os.listdir(group_dir): path_runinfo = os.path.join(group_dir, subdir, "Chatbench_RunInfo.json") if os.path.exists(path_runinfo): self.runinfo_paths.append(path_runinfo) for path in self.runinfo_paths: self.dictlists.append(Analyzer(ls.load_json_dict(path))) self.dictlists: list[Analyzer] # check if the values in self.num(list) are the same if not self.all_equal("num"): print(self.num) raise ValueError("The total_num of the data are not the same") @property def analyzers(self): return self.dictlists def run(self, Evals=["Eval0", "Eval1", "Eval2"]): num_tasks = self.dictlists[0].total_num pass_at = self.pass_at_k_kvalues for Eval_idx in Evals: for pass_at_k in pass_at: self.Evalx_ratio_passatk(Eval_idx, pass_at_k) self.messages += "\n########## Analyze of Chatbench_RunInfos ##########\n" self.messages += "\n#### basic info:\n" self.messages += "total number of tasks: %d\n" % self.dictlists[0].total_num self.messages += "sample numbers: %d\n" % len(self.dictlists) self.messages += "\n#### pass@k ratios:\n" for key, value in self.result.items(): self.messages += "%s: %.2f%% (%.1f)\n" % (key, value*100, value*num_tasks) self.messages += "\nloose Eval2 pass metric applied: %s\n\n" % self.dictlists[0].loose_factor def save(self, path:str=None): if path is None: path = DEFAULT_LOG_PATH with open(path, "w") as f: f.write(self.messages) # also, write the current time f.write("analysis time: %s\n" % (utils.get_time())) def get_avg_tokens_one_task(self): self.prompt_tokens_num = 0 self.completion_tokens_num = 0 for analyzer in self.dictlists: self.prompt_tokens_num += analyzer.prompt_tokens_num self.completion_tokens_num += analyzer.completion_tokens_num self.avg_prompt_tokens = self.prompt_tokens_num / len(self.dictlists) / self.dictlists[0].total_num self.avg_completion_tokens = self.completion_tokens_num / len(self.dictlists) / self.dictlists[0].total_num self.messages += "average prompt tokens: %d\n" % self.avg_prompt_tokens self.messages += "average completion tokens: %d\n" % self.avg_completion_tokens def get_avg_pass_by_disc_and_corr(self): self.pass_by_corrected = 0.0 self.pass_by_disc = 0.0 for analyzer in self.dictlists: self.pass_by_corrected += analyzer.corrected_num self.pass_by_disc += analyzer.autoline_reboot_task_num self.pass_by_corrected /= len(self.dictlists) self.pass_by_disc /= len(self.dictlists) self.messages += "passed with functional corrector: %.1f\n" % self.pass_by_corrected self.messages += "passed with autoline reboot action: %.1f\n" % self.pass_by_disc def renew_result_dict(self): self.result_dict = HDLBitsProbset() self.result_dict.create_empty_set_via_taskids(self.dictlists[0].task_id_list) def Evalx_ratio_passatk(self, Eval_idx="Eval0", pass_at:int=1): """ return the ratio of the Eval0 pass under pass@k """ # assert Eval_idx in ["Eval0", "Eval1", "Eval2", "Eval2b"], "Eval_idx should be one of Eval0, Eval1, Eval2, Eval2b" if not hasattr(Analyzer, Eval_idx + "_pass"): raise ValueError("The function %s_pass is not defined in Analyzer" % Eval_idx) k = pass_at n = len(self.dictlists) Evalx_pass_at_k_total = 0 # compute the pass ratio under pass@k for each task for task_id in self.dictlists[0].task_id_list: if hasattr(self, "result_dict"): task_result = self.result_dict.access_data_via_taskid(task_id) pass_num = 0 for dictlist in self.dictlists: Evalx_pass_func = getattr(dictlist, "%s_pass"%Eval_idx) # if dictlist.access_data_via_taskid(task_id)["%s_pass"%Eval_idx]: if Evalx_pass_func(dictlist.access_data_via_taskid(task_id)): if not (self.exclude_debug and dictlist.debug_iter(dictlist.access_data_via_taskid(task_id))): # if exclude_debug is True and the task is debugged, we will not count it pass_num += 1 # data = self.result_set.access_data_via_taskid(task_id) pass_at_k = self.pass_at_k_under_n(n, k, pass_num) Evalx_pass_at_k_total += pass_at_k if hasattr(self, "result_dict"): task_result["%s_pass_num"%Eval_idx] = pass_num task_result["%s_pass_at_%d"%(Eval_idx, k)] = pass_at_k Evalx_pass_at_k_total /= self.dictlists[0].total_num self.result["%s_pass_at_%d" % (Eval_idx, k)] = Evalx_pass_at_k_total @staticmethod def pass_at_k_under_n(n:int, k:int, c:int): """ - n: total number of samples - k: number of samples we want to pick - c: number of samples passed - output: pass@k under n - return the pass ratio under pass@k for n times; we have n samples, pass_num samples passed. Now we want to calculate the possibility that we pick k samples and at least one of them passed """ return 1 - (math.comb(n-c, k) / math.comb(n, k)) def Eval2_histogram(): """draw Eval2 histogram""" k_list = [1,5,10] multi_analyzer = MultiAnalyzer(MULTI_DIR, k_list) ratios = [] for analyzer in multi_analyzer.dictlists: for i in analyzer.data: if analyzer.Eval0_pass(i) and analyzer.Eval1_pass(i): # get the numerical ratio ratio_str = i.get("Eval2_ratio", None) if ratio_str is None: continue numerator, denominator = ratio_str.split("/") ratio = float(numerator) / float(denominator) ratios.append(ratio) # draw histogram of ratios, it will have 10 bars, 0~10%, 10~20%, ..., 90~100% import matplotlib.pyplot as plt import matplotlib matplotlib.use('Agg') plt.hist(ratios, bins=10, range=(0,1)) # title: distribution of Eval2 (Our work) plt.title("distribution of Eval2 (Baseline)") # y label is the number of tasks, tick is 10 plt.ylabel("number of tasks") plt.yticks(range(0, 700, 50)) # small grid so we can see the number of tasks, major grid, both axis. the grid color should be very light plt.grid(True, which='both', axis='both', color='lightgray', linestyle='-', linewidth=0.5) # x label is the ratio, tick is 10% plt.xlabel("ratio") plt.xticks([0.1 * i for i in range(11)]) # save to analysis/eval2_histogram.png plt.savefig(os.path.join(DEFAULT_SAVING_DIR, "eval2_histogram_Baseline.png")) # save the ratios to out, it is grouped by 0.1. the first group will contain both 0 and 0.1 # the format is csv with open(os.path.join(DEFAULT_SAVING_DIR, "eval2_ratios_Baseline.txt"), "w") as f: for i in range(11): ratio = 0.1 * i ratio_num = len([j for j in ratios if math.floor(j*10) == i]) f.write("%.1f, %d\n" % (ratio, ratio_num)) # export the original bin data with open(os.path.join(DEFAULT_SAVING_DIR, "eval2_ratios_bin_Baseline.txt"), "w") as f: for i in ratios: f.write("%.2f\n" % i) plt.close() def task_eval2passtimes_analyze(): multi_analyzer = MultiAnalyzer(MULTI_DIR) pass_taskids_list = [] for analyzer in multi_analyzer.dictlists: # find the task_id that pass Eval2 pass_taskids = [] for data in analyzer.data: if analyzer.Eval2_pass(data): pass_taskids.append(data["task_id"]) pass_taskids_list.append(pass_taskids) # find the idx of task "lemming3" idxs = [] for idx, pass_taskids in enumerate(pass_taskids_list): if "countbcd" in pass_taskids: idxs.append(idx) print("countbcd passed at:") print(idxs) # calculate the pass times for each task_id # pass_times_dict = {} # for pass_taskids in pass_taskids_list: # for task_id in pass_taskids: # if task_id in pass_times_dict.keys(): # pass_times_dict[task_id] += 1 # else: # pass_times_dict[task_id] = 1 # circuit_type_data_path = "data/HDLBits/HDLBits_circuit_type.jsonl" # SEQ_task_ids = HDLBitsProbset(circuit_type_data_path, filter_content={"circuit_type": "SEQ"}).task_id_list # # remove tasks that are not seq and then print # seq_passed_tasks = [] # for task_id in pass_times_dict.keys(): # if task_id in SEQ_task_ids: # seq_passed_tasks.append(task_id) # # pick 5 most complex tasks according to complexity = 1*len(description) + 2*len(module_code) # HDLdata = HDLBitsProbset("data/HDLBits/HDLBits_data.jsonl", only_tasks=seq_passed_tasks) # HDLdata.data.sort(key=lambda x: 1*len(x["description"]) + 2*len(x["module_code"]), reverse=True) # # print the task_id and pass times of the 5 most complex tasks # for i in range(10): # task_id = HDLdata.data[i]["task_id"] # pass_times = pass_times_dict[task_id] # print(task_id + ": " + str(pass_times)) if __name__ == "__main__": main() # FULLEXP_no1_paths = [ # "saves/1211~1217/Chatbench_RunInfo.json" # ] # def get_TCpass_num(data): # """ # return the number of passed tasks # """ # pass_num = 0 # for i in data: # if i["TC_pass"]: # pass_num += 1 # return pass_num # def get_debugTCpass_num(data): # """ # return the number of passed tasks with debug iter != 0 # """ # pass_num = 0 # for i in data: # if i["TC_pass"] and self.debug_iter(i) != 0: # pass_num += 1 # return pass_num # def get_debugsimpass_num(data): # """ # return the number of passed tasks with debug iter != 0 # """ # pass_num = 0 # for i in data: # if self.Eval0_pass(i) and self.debug_iter(i) != 0: # pass_num += 1 # return pass_num # def get_average_debugiter_debugTCpass(data): # """ # return the average debug iter of the passed with debug data # """ # debug_iter_sum = 0 # debug_iter_num = 0 # for i in data: # if i["TC_pass"] and self.debug_iter(i) != 0: # debug_iter_sum += self.debug_iter(i) # debug_iter_num += 1 # return debug_iter_sum / debug_iter_num # def get_average_time(data): # """ # return the average time of the data # """ # time_sum = 0 # for i in data: # time_sum += i["time"] # return time_sum / len(data) # def get_num_of_onetime_simpass(data): # """ # return the number of tasks that passed in the first run # """ # pass_num = 0 # for i in data: # if self.Eval0_pass(i) and self.debug_iter(i) == 0: # pass_num += 1 # return pass_num # def analyze_EXP1_main(json_file_list): # output_data_list = [] # for json_file in json_file_list: # data = ls.load_json_dict(json_file) # output_data_list.extend(data) # analyze(output_data_list) # def correct_exp_no1_main(json_file_list): # output_data_list = correct_exp_no1(json_file_list) # analyze(output_data_list) # def analyze(output_data_list): # # show all of the above processed data # print("total number of tasks: %d" % (get_total_num(output_data_list))) # print("number of simpassed tasks: %d" % (get_simpass_num(output_data_list))) # print("number of allpassed tasks: %d" % (get_TCpass_num(output_data_list))) # print("TCpass percentage: %.2f%%" % (get_TCpass_num(output_data_list) / get_total_num(output_data_list) * 100)) # print("number of debug_and_TCpass: %d" % (get_debugTCpass_num(output_data_list))) # print("number of debug_and_simpass: %d" % (get_debugsimpass_num(output_data_list))) # print("average debug iter of debug_and_TCpass: %.2f" % (get_average_debugiter_debugTCpass(output_data_list))) # print("average time: %.2fs" % (get_average_time(output_data_list))) # # save them to a txt file analyze_out.txt # with open("analyze_out.txt", "w") as f: # f.write("total number of tasks: %d\n" % (get_total_num(output_data_list))) # f.write("number of simpassed tasks: %d\n" % (get_simpass_num(output_data_list))) # f.write("number of allpassed tasks: %d\n" % (get_TCpass_num(output_data_list))) # f.write("TCpass percentage: %.2f%%\n" % (get_TCpass_num(output_data_list) / get_total_num(output_data_list) * 100)) # f.write("number of debug_and_TCpass: %d\n" % (get_debugTCpass_num(output_data_list))) # f.write("number of debug_and_simpass: %d\n" % (get_debugsimpass_num(output_data_list))) # f.write("average debug iter of debug_and_TCpass: %.2f\n" % (get_average_debugiter_debugTCpass(output_data_list))) # f.write("average time: %.2fs\n" % (get_average_time(output_data_list))) # # also, write the current time # f.write("time: %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))) # def correct_exp_no1(json_file_list): # new_results = [] # for json_file in json_file_list: # data_dir = json_file[:json_file.rfind("/")+1] # results = ls.load_json_dict(json_file) # for prob_result in results: # prob_dir = data_dir + prob_result["task_id"] + "/" # if prob_result["debug_iter"] == 0: # prob_last_run_info_path = prob_dir + "TBgen_codes/" + "run_info.txt" # else: # prob_last_run_info_path = prob_dir + "debug_%s" % (prob_result["debug_iter"]) + "/" + "run_info.txt" # prob_last_run_info = ls.load_txt(prob_last_run_info_path) # # check if "All test cases passed" is in the last run info # if "test cases passed" in prob_last_run_info: # prob_result["TC_pass"] = True # else: # prob_result["TC_pass"] = False # new_results.extend(results) # ls.save_dict_json_form(new_results, "corrected_exp_no1.json") # return new_results # if __name__ == "__main__": # # correct_exp_no1_main(FULLEXP_no1_paths) # analyze_EXP1_main(FULLEXP_no1_paths) # # main()