| import json | |
| import multiprocessing | |
| import os | |
| from evalplus._experimental.type_mut_for_eff import TypedMutEffGen | |
| from evalplus.data import HUMANEVAL_PLUS_INPUTS_PATH, get_human_eval_plus | |
| HUMANEVAL_PLUS_BIG_INPUTS_PATH = "/home/yuyao/eval-plus/HumanEvalPlusBigInputs" | |
| def main(): | |
| problems = get_human_eval_plus() | |
| for p in problems: | |
| print(f"{p['task_id']}...") | |
| filename = p["task_id"].replace("/", "_") | |
| big_input_path = os.path.join( | |
| HUMANEVAL_PLUS_BIG_INPUTS_PATH, f"{filename}.json" | |
| ) | |
| if os.path.exists(big_input_path): | |
| continue | |
| inputs = p["base_input"] | |
| signature = p["entry_point"] | |
| contract_code = p["prompt"] + p["contract"] + p["canonical_solution"] | |
| def input_generation(inputs, signature, contract_code): | |
| try: | |
| gen = TypedMutEffGen(inputs, signature, contract_code) | |
| new_inputs = gen.generate() | |
| results.append(new_inputs) | |
| except: | |
| with open("fail.txt", "a") as f: | |
| f.write(f"{signature} failed") | |
| results.append("fail") | |
| manager = multiprocessing.Manager() | |
| results = manager.list() | |
| proc = multiprocessing.Process( | |
| target=input_generation, args=(inputs, signature, contract_code) | |
| ) | |
| proc.start() | |
| proc.join(timeout=300) | |
| if proc.is_alive(): | |
| proc.terminate() | |
| proc.kill() | |
| continue | |
| if len(results) == 0 or type(results[0]) == str: | |
| continue | |
| new_inputs = results[0] | |
| new_input_dict = dict() | |
| new_input_dict["task_id"] = p["task_id"] | |
| new_input_dict["inputs"] = [] | |
| new_input_dict["sd"] = [] | |
| for item in new_inputs: | |
| new_input_dict["inputs"].append(item.inputs) | |
| new_input_dict["sd"].append(item.fluctuate_ratio) | |
| with open( | |
| os.path.join(HUMANEVAL_PLUS_BIG_INPUTS_PATH, f"{filename}.json"), "w" | |
| ) as f: | |
| json.dump(new_input_dict, f) | |
| if __name__ == "__main__": | |
| main() | |