odd_class_classification.py

# -*- coding: utf-8 -*-
"""
out-of-distribution experiment where a class (alignment) is removed from the
training set but remains in the testing set

assume a binary classification problem
"""

import matplotlib.pyplot as plt

from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, ConfusionMatrixDisplay

from data_module.alignment_data import AlignmentData
from data_module.feature_extractor import FeatureExtractor
from data_module.features import Mean


rm_filenames = ['misaligned_0.03_1500rpm_10min.txt',
                'misaligned_0.06_1500rpm_10min.txt',
                'misaligned_0.09_1500rpm_10min.txt',
                'misaligned_0.12_1500rpm_10min.txt',
                'misaligned_0.15_1500rpm_10min.txt']

sampling_method = {'name': 'segment', 'interval': 10000}

features = [Mean()]

class_map = {'0.0': '0', '0.03': '1', '0.06': '2', '0.09': '3', '0.12': '4', '0.15': '5'}


def main():
    
    # prep data
    data = AlignmentData()
    data.load_all_filenames()
    data.remove_filenames(rm_filenames)
    data.load_raw_data()
    
    # create dataset
    feature_extractor = FeatureExtractor(features)
    df = data.create_dataset(feature_extractor, sampling_params=sampling_method, class_map=class_map)
    print('Number of Observations: ', df.shape[0])
    
    # train and test sets 
    X_train, X_test, y_train, y_test, label_train, _ = train_test_split(df.iloc[:,:14], df['Label'], df['Alignment'], test_size=0.33)

    alignment_classes = df.loc[df['Label'] == 'misaligned', 'Alignment'].unique()
    for c in alignment_classes:   
        
        # drop from training set
        X_train_c, y_train_c = X_train.copy(), y_train.copy()
        drop_idx = label_train[label_train == c].index
        X_train_c.drop(index=drop_idx, inplace=True)
        y_train_c.drop(index=drop_idx, inplace=True)
        
        # train model
        rf = RandomForestClassifier().fit(X_train_c, y_train_c)
        
        # evaluate
        y_pred = rf.predict(X_test)
        accuracy = accuracy_score(y_test, y_pred)
        print('Accuracy ' + str(c) + ' removed: ', accuracy)
        
        ConfusionMatrixDisplay.from_predictions(y_test, y_pred)
        plt.title('Confusion Matrix ' + str(c) + ' Removed')

    
    
if __name__ == "__main__":
    main()