test_utilities.py

"""Tests for the experiment utilities module."""

from pathlib import Path
from shutil import rmtree

import numpy as np
import numpy.testing as npt
import pandas as pd
import pytest

from orca_python.testing import TEST_DATASETS_DIR
from orca_python.utilities import Utilities


@pytest.fixture
def util():
    general_conf = {}
    configurations = {}
    return Utilities(general_conf, configurations)


def create_csv(path, filename):
    """Create a csv file with sample data."""
    sample_data = "1,2,3,0\n4,5,6,1"
    (path / filename).write_text(sample_data)


def test_load_complete_dataset(tmp_path, util):
    """Loading dataset composed of 5 partitions, each one of them composed of
    a train and test file.

    """
    dataset_path = tmp_path / "complete"
    dataset_path.mkdir()

    for i in range(5):
        create_csv(dataset_path, f"train_complete.{i}")
        create_csv(dataset_path, f"test_complete.{i}")

    partition_list = util._load_dataset(dataset_path)

    # Check all partitions have been loaded
    npt.assert_equal(len(partition_list), (len(list(dataset_path.iterdir())) / 2))
    # Check if every partition has train and test inputs and outputs (4 diferent dictionaries)
    npt.assert_equal(
        all([len(partition[1]) == 4 for partition in partition_list]), True
    )


def test_load_partitionless_dataset(tmp_path, util):
    """Loading dataset composed of only two csv files (train and test files)"""
    dataset_path = tmp_path / "partitionless"
    dataset_path.mkdir()

    create_csv(dataset_path, "train_partitionless.csv")
    create_csv(dataset_path, "test_partitionless.csv")

    partition_list = util._load_dataset(dataset_path)

    npt.assert_equal(len(partition_list), 1)
    npt.assert_equal(
        all([len(partition[1]) == 4 for partition in partition_list]), True
    )


def test_load_nontestfile_dataset(tmp_path, util):
    """Loading dataset composed of five train files."""
    dataset_path = tmp_path / "nontestfile"
    dataset_path.mkdir()

    for i in range(5):
        create_csv(dataset_path, f"train_nontestfile.{i}")

    partition_list = util._load_dataset(dataset_path)

    npt.assert_equal(len(partition_list), len(list(dataset_path.iterdir())))
    npt.assert_equal(
        all([len(partition[1]) == 2 for partition in partition_list]), True
    )


def test_load_nontrainfile_dataset(tmp_path, util):
    """Loading dataset with 2 partitions, one of them lacking its train file.
    This should raise an exception.

    """
    dataset_path = tmp_path / "nontrainfile"
    dataset_path.mkdir()

    for i in range(2):
        create_csv(dataset_path, f"test_nontrainfile.{i}")

    with pytest.raises(RuntimeError):
        util._load_dataset(dataset_path)


def test_normalize_data(tmp_path, util):
    # Test preparation
    dataset_path = Path(TEST_DATASETS_DIR) / "balance-scale"

    train_file = np.loadtxt(dataset_path / "train_balance-scale.csv", delimiter=",")
    X_train = train_file[:, 0:(-1)]

    test_file = np.loadtxt(dataset_path / "test_balance-scale.csv", delimiter=",")
    X_test = test_file[:, 0:(-1)]

    # Test execution
    norm_X_train, _ = util._normalize_data(X_train, X_test)

    # Test verification
    result = (norm_X_train >= 0).all() and (norm_X_train <= 1).all()
    npt.assert_equal(result, True)


def test_standardize_data(util):
    # Test preparation
    dataset_path = Path(TEST_DATASETS_DIR) / "balance-scale"

    train_file = np.loadtxt(dataset_path / "train_balance-scale.csv", delimiter=",")
    X_train = train_file[:, 0:(-1)]

    test_file = np.loadtxt(dataset_path / "test_balance-scale.csv", delimiter=",")
    X_test = test_file[:, 0:(-1)]

    # Test execution
    std_X_train, _ = util._standardize_data(X_train, X_test)

    # Test verification
    npt.assert_almost_equal(np.mean(std_X_train), 0)
    npt.assert_almost_equal(np.std(std_X_train), 1)


@pytest.fixture
def main_folder():
    return Path(__file__).parent.parent.parent


@pytest.fixture
def dataset_folder(main_folder):
    return main_folder / "datasets" / "data"


@pytest.fixture
def general_conf(dataset_folder):
    return {
        "basedir": dataset_folder,
        "datasets": ["tae", "contact-lenses"],
        "input_preprocessing": "std",
        "hyperparam_cv_nfolds": 3,
        "jobs": 10,
        "output_folder": "my_runs/",
        "metrics": ["ccr", "mae", "amae", "mze"],
        "cv_metric": "mae",
    }


@pytest.fixture
def configurations():
    return {
        "SVM": {
            "classifier": "SVC",
            "parameters": {"C": [0.001, 0.1, 1, 10, 100], "gamma": [0.1, 1, 10]},
        },
        "SVMOP": {
            "classifier": "OrdinalDecomposition",
            "parameters": {
                "dtype": "ordered_partitions",
                "decision_method": "frank_hall",
                "base_classifier": "SVC",
                "parameters": {
                    "C": [0.01, 0.1, 1, 10],
                    "gamma": [0.01, 0.1, 1, 10],
                    "probability": ["True"],
                },
            },
        },
    }


def test_run_experiment(main_folder, general_conf, configurations):
    """To test the main method, a configuration will be run until the end.
    Next we will check that every expected result file has been created,
    having all of them the proper dimensions and types.

    """
    # Declaring Utilities object and running the experiment
    util = Utilities(general_conf, configurations, verbose=False)
    util.run_experiment()
    # Saving results information
    util.write_report()

    # Checking if all outputs have been generated and are correct
    outputs_folder = Path("my_runs")
    npt.assert_equal(outputs_folder.exists(), True)

    experiment_folder = sorted(outputs_folder.iterdir())
    experiment_folder = outputs_folder / experiment_folder[-1].name

    for dataset in util.general_conf["datasets"]:
        for conf_name, _ in util.configurations.items():

            # Check if the folder for that dataset-configurations exists
            conf_folder = experiment_folder / f"{dataset}-{conf_name}"
            npt.assert_equal(conf_folder.exists(), True)

            # Checking CSV containning all metrics for that configuration
            metrics_csv = pd.read_csv(conf_folder / f"{dataset}-{conf_name}.csv")
            metrics_csv = metrics_csv.iloc[:, -12:]

            npt.assert_equal(metrics_csv.shape, (30, 12))
            npt.assert_equal(all(str(c) == "float64" for c in metrics_csv.dtypes), True)

            # Checking that all models have been saved
            models_folder = conf_folder / "models"
            npt.assert_equal(models_folder.exists(), True)
            npt.assert_equal(len(list(models_folder.iterdir())), 30)

            # Checking that all predictions have been saved
            predictions_folder = conf_folder / "predictions"
            npt.assert_equal(predictions_folder.exists(), True)
            npt.assert_equal(len(list(predictions_folder.iterdir())), 60)

    # Checking if summaries are correct
    train_summary = pd.read_csv(experiment_folder / "train_summary.csv")
    npt.assert_equal(train_summary.shape, (4, 13))
    npt.assert_equal(
        all(str(c) == "float64" for c in train_summary.dtypes.iloc[1:]), True
    )

    test_summary = pd.read_csv(experiment_folder / "test_summary.csv")
    npt.assert_equal(test_summary.shape, (4, 13))
    npt.assert_equal(
        all(str(c) == "float64" for c in test_summary.dtypes.iloc[1:]), True
    )

    rmtree(outputs_folder)