test_model_baseline.py

#!/usr/bin/env python
import numpy as np
import paths as pt
from typing import List
from tools import file_writer, preprocessor, data_loader
from tools.classifiers import KnnClassifier, SvmClassifier, LrClassifier
from tools.classifiers import XgbClassifier, RfClassifier, MlpClassifier
from pathlib import Path
import csv
from utility.settings import load_settings
from utility.metrics import compute_mean, compute_std

def get_version_subtitle(version):
    if version == "NoAts":
        return "without Ats and/or Ex columns"
    elif version == "Embedded":
        return "with embeddings"
    else:
        return "with counts"

def prepare_data(X, y, settings):
    features_to_normalize = settings['features_to_normalize']
    features_to_scale = settings['features_to_scale']
    X = preprocessor.normalize_data(X, features_to_normalize)
    X = preprocessor.scale_data(X, features_to_scale)
    X = np.array(X)
    y = np.array(y)
    return X, y

def load_data_embedded(case, settings):
    if case == "Complete":
        dl = data_loader.CompleteDataLoader("complete_emb.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Compliance":
        dl = data_loader.ComplianceDataLoader("compliance_emb.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Alarm":
        dl = data_loader.AlarmDataLoader("alarm_emb.csv", settings).load_data()
        X, y = dl.get_data()
    else:
        dl = data_loader.FallDataLoader("fall_emb.csv", settings).load_data()
        X, y = dl.get_data()
    return X, y

def load_data_count(case, settings):
    if case == "Complete":
        dl = data_loader.CompleteDataLoader("complete_count.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Compliance":
        dl = data_loader.ComplianceDataLoader("compliance_count.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Alarm":
        dl = data_loader.AlarmDataLoader("alarm_count.csv", settings).load_data()
        X, y = dl.get_data()
    else:
        dl = data_loader.FallDataLoader("fall_count.csv", settings).load_data()
        X, y = dl.get_data()
    return X, y

def load_data_ohe(case, settings):
    if case == "Complete":
        dl = data_loader.CompleteDataLoader("complete_ohe.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Compliance":
        dl = data_loader.ComplianceDataLoader("compliance_ohe.csv", settings).load_data()
        X, y = dl.get_data()
    elif case == "Alarm":
        dl = data_loader.AlarmDataLoader("alarm_ohe.csv", settings).load_data()
        X, y = dl.get_data()
    else:
        dl = data_loader.FallDataLoader("fall_ohe.csv", settings).load_data()
        X, y = dl.get_data()
    return X, y

def main():
    num_iter = 5
    clf_names = ['KNN', 'SVM', 'LR', 'XGB', 'RF', 'MLP']
    num_clfs = len(clf_names)
    metrics = ['accuracy', 'precision', 'recall', 'roc_auc', 'average_precision', 'f1']
    #cases = ["Complete", "Compliance", "Fall", "Risk"]
    cases = ['Alarm']
    for case in cases:
        settings = load_settings(f'{case.lower()}_emb.yaml')
        output_filename = f"{case} model baseline.csv"
        header = ['clf', 'version', 'accuracy_mean', 'accuracy_std',
                  'precision_mean', 'precision_std', 'recall_mean',
                  'recall_std', 'roc_auc_mean', 'roc_auc_std',
                  'pr_auc_mean', 'pr_auc_std', 'f1_mean', 'f1_std']
        with open(Path.joinpath(pt.REPORTS_DIR, output_filename), 'w',
                  encoding='UTF8', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(header)
        versions = ['NoAts', 'Embedded', 'Counts', 'OneHot']
        for version in versions:
            if version == 'NoAts':
                ats_cols = [f"{i}Ats" for i in range(1, 11)]
                X, y = load_data_embedded(case, settings)
                X = X.drop(ats_cols, axis=1)
            elif version == "Embedded":
                X, y = load_data_embedded(case, settings)
            elif version == "Counts":
                X, y = load_data_count(case, settings)
            else:
                X, y = load_data_ohe(case, settings)
                
            X, y = prepare_data(X, y, settings)
            results = train_clf(X, y, version, output_filename, metrics, num_iter)
            subtitle = get_version_subtitle(version)
            make_plots(results, metrics, num_iter, num_clfs,
                       clf_names, case, version, subtitle)
                
def train_clf(X, y, version, output_filename, metrics, num_iter):
    iteration_results = list()
    for k in range(num_iter):
        results = {'KNN': KnnClassifier(X, y).evaluate(metrics, k),
                   'SVM': SvmClassifier(X, y).evaluate(metrics, k),
                   'LR': LrClassifier(X, y).evaluate(metrics, k),
                   'XGB': XgbClassifier(X, y).evaluate(metrics, k),
                   'RF': RfClassifier(X, y).evaluate(metrics, k),
                   'MLP': MlpClassifier(X, y).evaluate(metrics, k)}
        iteration_results.append(results)
        
    for iter_result in iteration_results:
        for clf_name, result in iter_result.items():
            with open(Path.joinpath(pt.REPORTS_DIR, output_filename), 'a',
                      encoding='UTF8', newline='') as f:
                writer = csv.writer(f)
                data = [clf_name, version]
                for metric in metrics:
                    mean = compute_mean(result[metric])
                    std = compute_std(result[metric])
                    data.extend((mean, std))
                writer.writerow(data)
                
    return iteration_results

def make_plots(results: np.ndarray, metrics: List[str], num_iter: int,
               num_clfs: int, clf_names, case: str, version: str, case_subtitle):
    for metric in metrics:
        total_means, total_stds = list(), list()
        for iter_result in results:
            means, stds = np.zeros(num_clfs), np.zeros(num_clfs)
            for i, (_, result) in enumerate(iter_result.items()):
                means[i] = compute_mean(result[metric])
                stds[i] = compute_std(result[metric])
            total_means.append(means)
            total_stds.append(stds)
        total_means = np.stack(total_means, axis=-1)
        total_stds = np.stack(total_stds, axis=-1)
        file_name = f"{case} version {version} - {metric}.pdf"
        file_writer.write_cv_plot(total_means, total_stds, metric,
                                  num_iter, clf_names, pt.REPORTS_PLOTS_DIR,
                                  file_name, case_subtitle)
       
if __name__ == '__main__':
    main()