prediction_postprocessing_scripts/handpick_best.py

# python3.9 ./analysis/scripts/summarize_metrics.py -s ./analysis/output/summaries -i abed_results -o only_best_plus_default
import os
import json
import argparse
import shutil

class MethodMeasurement:
    def __init__(self, f1_default=None, precision_default=None, recall_default=None, 
                 f1_oracle=None, precision_oracle=None, recall_oracle=None, 
                 f1_best=None, precision_best=None, recall_best=None, precision_f1_best=None, recall_f1_best=None):
        self.f1_default = f1_default
        self.precision_default = precision_default
        self.recall_default = recall_default
        self.f1_oracle = f1_oracle
        self.precision_oracle = precision_oracle
        self.recall_oracle = recall_oracle
        self.f1_best = f1_best
        self.precision_f1_best = precision_f1_best
        self.recall_f1_best = recall_f1_best
        self.precision_best = precision_best
        self.recall_best = recall_best

    def __setattr__(self, name, value):
        super().__setattr__(name, value)

    def __getattr__(self, name):
        return None


def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-s",
        "--summary-dir",
        help="Directory with summary files of all datasets/methods",
        required=True,
    )
    parser.add_argument(
        "-f",
        "--failure-threshold",
        help="The threshold of failed dataset runs per hyper parameter configuration (in decimal)",
        default=0.05,
        required=False,
    )
    parser.add_argument(
        "-i",
        "--input-directory",
        help="Directory with all raw results files of all datasets/methods",
        required=False,
    )
    parser.add_argument(
        "-o",
        "--output-directory",
        help="Directory in which to include the handpicked results files",
        required=False,
    )
    return parser.parse_args()

 
args = parse_args()
summaries_folder_path = args.summary_dir
failure_threshold_decimal = args.failure_threshold
input_directory = args.input_directory
output_directory = args.output_directory

# It will contain all the summaries from the summary directory. A summary is specific to one dataset. It has the results of running that dataset on all hyper parameters in all methods.
datasets_metrics = []
best_paths = []
default_paths = []

for filename in os.listdir(summaries_folder_path):
    if filename.endswith('.json'):
        file_path = os.path.join(summaries_folder_path, filename)
        with open(file_path, 'r') as file:
            data = json.load(file)
            datasets_metrics.append(data)

nb_datasets_threshold = len(datasets_metrics) * (1.0 - failure_threshold_decimal)
methods = set()

for dataset_metrics in datasets_metrics:
    if "results" in dataset_metrics:
        methods.update(dataset_metrics["results"].keys())
default_methods = {method for method in methods if method.startswith("default_")}
best_methods = {method for method in methods if method.startswith("best_")}
stripped_methods = {method.replace("best_", "").replace("default_", "") for method in methods if method.startswith("best_") or method.startswith("default_")}
# This dictionary will contain the conclusive results for each method (Default, Best, Oracle)
MethodsMeasurements = {method: MethodMeasurement() for method in stripped_methods}


def process_default(method):
    default_f1, default_precision, default_recall = -1, -1, -1
    stripped_method = method.replace("default_", "")
    nb_success = 0
    for dataset_metrics in datasets_metrics:
        if dataset_metrics["results"][method][0]["status"] == "SUCCESS":
            nb_success += 1
            metrics = dataset_metrics["results"][method][0]["scores"]
            default_f1 = max(0, default_f1) + metrics["f1"]
            default_precision = max(0, default_precision) + metrics["precision"]
            default_recall = max(0, default_recall) + metrics["recall"]
    if nb_success > nb_datasets_threshold:
        if default_f1 > -1:
            MethodsMeasurements[stripped_method].f1_default = default_f1 / nb_success
        else:
            MethodsMeasurements[stripped_method].f1_default = None
        if default_precision > -1:
            MethodsMeasurements[stripped_method].precision_default = default_precision / nb_success
        else:
            MethodsMeasurements[stripped_method].precision_default = None
        if default_recall > -1:
            MethodsMeasurements[stripped_method].recall_default = default_recall / nb_success
        else:
            MethodsMeasurements[stripped_method].recall_default = None
    for dataset_metrics in datasets_metrics:
        signature_id = dataset_metrics["dataset"]
        default_conf_file_name = dataset_metrics["results"][method][0]["task_file"]
        default_paths.append(signature_id + "/" + method + "/" + default_conf_file_name)

def process_best(method):
    hyperparams = dict()
    stripped_method = method.replace("best_", "")
    for dataset_metrics in datasets_metrics:
        for unit_method in dataset_metrics["results"][method]:
            conf = unit_method["args"]
            conf_str = json.dumps(conf, sort_keys=True)
            if unit_method["status"] == "SUCCESS":
                metrics = unit_method["scores"]
                f1 = metrics["f1"]
                precision = metrics["precision"]
                recall = metrics["recall"]
                if conf_str in hyperparams:
                    hyperparams[conf_str]["f1"].append(f1)
                    hyperparams[conf_str]["precision"].append(precision)
                    hyperparams[conf_str]["recall"].append(recall)
                else:
                    metrics_dict = {
                        "f1": [f1],
                        "precision": [precision],
                        "recall": [recall]
                    }
                    hyperparams[conf_str] = metrics_dict

    dict_f1 = {key: sum(value['f1']) / len(value['f1']) for key, value in hyperparams.items() if len(value['f1']) > nb_datasets_threshold}
    dict_precision = {key: sum(value['precision']) / len(value['precision']) for key, value in hyperparams.items() if len(value['precision']) > nb_datasets_threshold}
    dict_recall = {key: sum(value['recall']) / len(value['recall']) for key, value in hyperparams.items() if len(value['recall']) > nb_datasets_threshold}

    try:
        max_f1 = dict_f1[max(dict_f1, key=dict_f1.get)]
    except Exception as e:
        max_f1 = None
    try:
        precision_max_f1 = dict_precision[max(dict_f1, key=dict_f1.get)]
    except Exception as e:
        precision_max_f1 = None
    try:
        recall_max_f1 = dict_recall[max(dict_f1, key=dict_f1.get)]
    except Exception as e:
        recall_max_f1 = None
    try:
        max_precision = dict_precision[max(dict_precision, key=dict_precision.get)]
    except Exception as e:
        max_precision = None
    try:
        max_recall = dict_recall[max(dict_recall, key=dict_recall.get)]
    except Exception as e:
        max_recall = None
    MethodsMeasurements[stripped_method].f1_best = max_f1
    MethodsMeasurements[stripped_method].precision_best = max_precision
    MethodsMeasurements[stripped_method].recall_best = max_recall
    MethodsMeasurements[stripped_method].precision_f1_best = precision_max_f1
    MethodsMeasurements[stripped_method].recall_f1_best = recall_max_f1
    if max_f1:
        best_f1_conf = max(dict_f1, key=dict_f1.get)
        for dataset_metrics in datasets_metrics:
            signature_id = dataset_metrics["dataset"]
            best_conf_file_names = [conf["task_file"] for conf in dataset_metrics["results"][method] if json.dumps(conf["args"], sort_keys=True)  == best_f1_conf]
            for file_name in best_conf_file_names:
                best_paths.append(signature_id + "/" + method + "/" + file_name)
        
    

def process_oracle(method):
    metrics_dict = {'f1': [], 'recall': [], 'precision': []}
    stripped_method = method.replace("best_", "")
    for dataset_metrics in datasets_metrics:
        oracle_f1, oracle_precision, oracle_recall = -1, -1, -1
        for unit_method in dataset_metrics["results"][method]: 
            if unit_method["status"] == "SUCCESS":
                metrics = unit_method["scores"]
                f1 = metrics["f1"]
                precision = metrics["precision"]
                recall = metrics["recall"]
                if f1 > oracle_f1:
                    oracle_f1 = f1
                if precision > oracle_precision:
                    oracle_precision = precision
                if recall > oracle_recall:
                    oracle_recall = recall
        if oracle_f1 > -1:
            metrics_dict["f1"].append(oracle_f1)
        if oracle_precision > -1:
            metrics_dict["precision"].append(oracle_precision)
        if oracle_recall > -1:
            metrics_dict["recall"].append(oracle_recall)
    if len(metrics_dict["f1"]) > 0:
        MethodsMeasurements[stripped_method].f1_oracle = sum(metrics_dict["f1"]) / len(metrics_dict["f1"])
    if len(metrics_dict["precision"]) > 0:
        MethodsMeasurements[stripped_method].precision_oracle = sum(metrics_dict["precision"]) / len(metrics_dict["precision"])
    if len(metrics_dict["recall"]) > 0:
        MethodsMeasurements[stripped_method].recall_oracle = sum(metrics_dict["recall"]) / len(metrics_dict["recall"])


def copy_file(input_path: str, output_path: str):
    try:
        os.makedirs(os.path.dirname(output_path), exist_ok=True)  # Ensure directories exist
        shutil.copy2(input_path, output_path)
    except FileNotFoundError:
        print(f"Error: Input file '{input_path}' not found.")
    except PermissionError:
        print("Error: Permission denied.")
    except Exception as e:
        print(f"Error copying file: {e}")

for method in default_methods:
    process_default(method)
for method in best_methods:
    process_oracle(method)
    process_best(method)

data = {key: vars(value) for key, value in MethodsMeasurements.items()}

# Sort methods alphabetically by their names
sorted_methods = sorted(data.items())

latex_table = """
\\begin{table*}[t!]
\\centering
\\resizebox{\\textwidth}{!}{%
\\begin{tabular}{|l|c|c|c||c|c|c|c|c||c|c|c|}
\\hline
Method & \\multicolumn{3}{c||}{Default} & \\multicolumn{5}{c||}{Best} & \\multicolumn{3}{c|}{Oracle} \\\\
\\cline{2-13}
 & F1 & Precision & Recall & F1 & Precision & Recall & Precision (F1 max) & Recall (F1 max) & F1 & Precision & Recall \\\\
\\hline
"""

def format_none(value):
    if value is None:
        return ""
    return f"{value:.3f}"

for method, metrics in sorted_methods:
    latex_table += f"{method} & "
    latex_table += " & ".join([
        format_none(metrics.get('f1_default')),
        format_none(metrics.get('precision_default')),
        format_none(metrics.get('recall_default')),
        format_none(metrics.get('f1_best')),
        format_none(metrics.get('precision_best')),
        format_none(metrics.get('recall_best')),
        format_none(metrics.get('precision_f1_best')),
        format_none(metrics.get('recall_f1_best')),
        format_none(metrics.get('f1_oracle')),
        format_none(metrics.get('precision_oracle')),
        format_none(metrics.get('recall_oracle'))
    ])
    latex_table += " \\\\\n"

latex_table += "\\hline\n\\end{tabular}%%\n}\n\\caption{Performance Metrics for Methods}\n\\end{table*}"

for path in best_paths:
    input_path = input_directory + "/" + path
    output_path = output_directory + "/" + path
    copy_file(input_path, output_path)

for path in default_paths:
    input_path = input_directory + "/" + path
    output_path = output_directory + "/" + path
    copy_file(input_path, output_path)


print(latex_table)