process_statistics

process_statistics.py

def process_statistics(model_containers, config):
    for model_name, model_data in model_containers.items():
        run_statistics = {}
        for label, counts in model_data['stats'].items():
            recall_denom = counts['false_negatives'] + counts['true_positives']
            precision_denom = counts['false_positives'] + counts['true_positives']
            if (recall_denom > 0) and (precision_denom > 0):
                precision = counts['true_positives'] / precision_denom
                recall = counts['true_positives'] / recall_denom
                if precision + recall > 0:
                    model_data['stats'][label]['f1'] = 2 * ((precision * recall) / (precision + recall))
            if counts['total'] > 0:
                accuracy = (counts['true_positives'] / counts['total']) * 100
                model_data['stats'][label]['accuracy'] = accuracy

        model_data['stats'] = {
            k: v for k, v in
            sorted(model_data['stats'].items(), key=lambda item: item[1]['f1'], reverse=True)
        }
        f1_score = model_data['metrics']['f1'].compute(average=F1_AVERAGE)['f1']
        acc_score = model_data['metrics']['acc'].compute()['accuracy']

        run_statistics['overall_accuracy'] = acc_score
        run_statistics['overall_f1'] = f1_score
        run_statistics['class_statistics'] = model_data['stats']

        run_statistics_path = os.path.join(os.getcwd(), f"{model_name}_run_statistics.json")
        with open(run_statistics_path, "w", encoding="utf-8") as f:
            json.dump(run_statistics, f, ensure_ascii=False, indent=4)

        print(f"{model_name} test accuracy: {acc_score}")
        print(f"{model_name} test f1: {f1_score}")
        s3_run_stats = f"{config.s3_parent_dir}/run_{config.run_num}/inference/{model_name}_run_statistics.json"

        print(f"Uploading {model_name} run statistics to {s3_run_stats}")
        bucket.upload_file(run_statistics_path, s3_run_stats)