diff --git a/minimizer.py b/minimizer.py new file mode 100644 index 00000000..827d7368 --- /dev/null +++ b/minimizer.py @@ -0,0 +1,119 @@ +import argparse +import os +import pickle +import sys +import time +import glob +from colorama import Fore, Back, Style +from pathlib import Path +from os.path import join +from multiprocessing import Pool, cpu_count,set_start_method +from itertools import repeat + +from verifiers import * + + +def minimize_model(failed_model_file,output_dir): + with open(failed_model_file, 'rb') as fpcl: + """ The error data is a dict with the following keys: + solver: the used solver + verifier: the verifier that got used + mutations_per_model: the amount of mutations that were used + seed: the used seed + error: a dict containing: + type: the type of error that occured + model: the newly mutated model that failed/crashed + originalmodel: the name of the model file that was used + mutators: a list with executed mutations + constraints: the constraints that made the model fail/crash + """ + error_data = pickle.loads(fpcl.read()) + original_error = error_data["error"] + original_cons = error_data["error"]["constraints"] + amount_of_original_cons = len(error_data["error"]["constraints"]) + + new_cons = [] + + for con in toplevel_list(original_cons): + test_cons = original_error["constraints"] + test_cons.remove(con) + new_error_dict = copy.deepcopy(original_error) + + vrf_cls = lookup_verifier(error_data['verifier']) + verifier = vrf_cls(solver=error_data['solver'], + mutations_per_model=error_data["mutations_per_model"], + exclude_dict = {}, + time_limit = time.time()*3600, + seed = error_data["seed"] + ) + new_error = verifier.rerun(new_error_dict) + + if new_error is not None: + # if we still get the error than the constraint is responsible we keep it + new_cons.append(con) + + #copy new constraints to model + error_data["error"]["constraints"] = new_cons + if "model" in error_data["error"]: + new_model = error_data["error"]["model"] + new_model.constraints = new_cons + error_data["error"]["model"] = new_model + #Fore.LIGHTBLUE_EX + + + with open(join(output_dir, "minimized_"+os.path.basename(failed_model_file)), "wb") as ff: + pickle.dump(error_data, file=ff) + with open(join(output_dir, "minimized_"+Path(os.path.basename(failed_model_file)).stem+".txt"), "w") as ff: + ff.write(create_error_output_text(error_data)) + + return f"minimized {failed_model_file} with {amount_of_original_cons} constraints to model with {len(new_cons)} constraints" + + + + + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description = "A Python script for minimizing the constraint for a failed model. It will only keep the constraints that caused the error") + def check_positive(value): + """ + Small helper function used in the argparser for checking if the input values are positive or not + """ + ivalue = int(value) + if ivalue <= 0: + raise argparse.ArgumentTypeError("%s is an invalid positive int value" % value) + return ivalue + parser.add_argument("-m", "--failed_model_file", help = "The path to a single pickle file or the path to a directory containing multiple pickle files", required=True, type=str) + parser.add_argument("-o", "--output-dir", help = "The directory to store the output (will be created if it does not exist).", required=False, type=str, default="minimizer_output") + parser.add_argument("-p","--amount-of-processes", help = "The amount of processes that will be used to run the tests", required=False, default=cpu_count()-1 ,type=check_positive) # the -1 is for the main process + args = parser.parse_args() + + current_working_directory = os.getcwd() + output_dir = os.path.join(current_working_directory, args.output_dir) + os.makedirs(output_dir, exist_ok=True) + failed_model_file = os.path.join(current_working_directory, args.failed_model_file) + + if os.path.isfile(failed_model_file): + result = minimize_model(failed_model_file,output_dir) + print(Fore.LIGHTBLUE_EX +"\n"+result) + print(Style.RESET_ALL+f"stored minimized model in {output_dir}") + + elif os.path.isdir(failed_model_file): + set_start_method("spawn") + print("detected directory") + files = glob.glob(failed_model_file+"/*.pickle") + with Pool(args.amount_of_processes) as pool: + try: + print("rerunning failed models in directory",flush=True) + results = pool.starmap(minimize_model, zip(files,repeat(output_dir))) + print("\n") + [print(Fore.LIGHTBLUE_EX + result) for result in results] + # for result in results: + # print(Fore.LIGHTBLUE_EX +result) + print(Style.RESET_ALL+"\nsucessfully minimized all the models",flush=True ) + except KeyboardInterrupt: + pass + finally: + print("quiting the application",flush=True ) + print(Style.RESET_ALL+f"stored minimized models in {output_dir}") + else: + print(Fore.YELLOW +"failed model file not found") + print(Style.RESET_ALL) \ No newline at end of file