# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python [conda env:tensorflow2]
#     language: python
#     name: conda-env-tensorflow2-py
# ---

# +
import os
import tqdm
import random
import itertools
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report, confusion_matrix

import tensorflow as tf
from tensorflow.keras.utils import *
from tensorflow.keras.layers import *
from tensorflow.keras.models import *
from tensorflow.keras.optimizers import *
from tensorflow.keras.callbacks import *

import librosa


# -

def plot_confusion_matrix(cm, classes, title='Confusion matrix', cmap=plt.cm.Blues):

    cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]

    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title, fontsize=25)
    #plt.colorbar()
    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=90, fontsize=15)
    plt.yticks(tick_marks, classes, fontsize=15)

    fmt = '.2f'
    thresh = cm.max() / 2.
    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        plt.text(j, i, format(cm[i, j], fmt),
                 horizontalalignment="center",
                 color="white" if cm[i, j] > thresh else "black", fontsize = 14)

    plt.ylabel('True label', fontsize=20)
    plt.xlabel('Predicted label', fontsize=20)


# +
### READ AND CONCAT DATA ###

label = pd.read_csv('profile.txt', sep='\t', header=None)
label.columns = ['Cooler','Valve','Pump','Accumulator','Flag']

df = pd.read_csv('EPS1.txt', sep='\t', header=None)

print(df.shape)
df.head()

# +
### PLOT ORIGINAL MOTOR POWER ###

df.iloc[2,:].plot(figsize=(8,5), title='Original Motor Power')
plt.ylabel('Motor Power (in W)'); plt.xlabel('Time')

# +
### APPLY CLIPPING AND FIRST ORDER DIFFERECE ###

df = df.values
df = np.clip(np.diff(df, axis=1), -5,5)
df.shape

# +
### PLOT STANDARDIZED DATA ###

plt.figure(figsize=(9,6))
plt.plot(df[2])
plt.title('Standardized Motor Power')
plt.ylabel('First Difference'); plt.xlabel('Time')
np.set_printoptions(False)

# +
### LABEL DISTRIBUTION ###

label = label.Flag
label.value_counts()

# +
### LABEL ENCODING ###

diz_label = {0:'stable', 1:'not stable'}
y = to_categorical(label)

# +
### APPLY MFCC TRANSFORMATION FOR EACH SIGNAL ###

df_mfcc = []

for i,sample in enumerate(tqdm.tqdm(df)):
    
    sample_mfcc = librosa.feature.mfcc(np.asfortranarray(sample),sr=40000)
    df_mfcc.append(sample_mfcc)
    
df_spectre = np.asarray(df_mfcc)
df_spectre = df_spectre.transpose(0,2,1)
df_spectre.shape

# +
### PLOT MFCC FOR A SINGLE SIGNAL ###

plt.figure(figsize=(9,6))

plt.plot(df_spectre[2])
plt.legend(['mfcc_'+str(i) for i in range(df_spectre.shape[-1])], 
          loc='center left', bbox_to_anchor=(1, 0.5))
plt.title('MFCC features')
plt.ylabel('Amplitudes'); plt.xlabel('Time')

np.set_printoptions(False)

# +
### TRAIN TEST SPLIT ###

X_train, X_test, y_train, y_test = train_test_split(df_spectre, y, random_state = 42, test_size=0.2)


# +
### DEFINE RESNET CNN ###

def set_seed(seed):
    
    tf.random.set_seed(seed)
    os.environ['PYTHONHASHSEED'] = str(seed)
    np.random.seed(seed)
    random.seed(seed)
    

def get_model(data):
    
    set_seed(33)
        
    def residual_block(init, hidden_dim):
        
        init = Conv1D(hidden_dim, 3, activation='relu', padding="same")(init)
        
        x = Conv1D(hidden_dim, 3, activation='relu', padding="same")(init)
        x = Conv1D(hidden_dim, 3, activation='relu', padding="same")(x)
        x = Conv1D(hidden_dim, 3, activation='relu', padding="same")(x)
        skip = Add()([x, init])
        
        return skip
    
    inp = Input(shape=(data.shape[1], data.shape[2]))
    
    x = residual_block(inp, 256)
    x = residual_block(x, 128)
    x = residual_block(x, 32)
    x = GlobalMaxPool1D()(x)
    
    out = Dense(len(diz_label), activation='softmax')(x)
    
    model = Model(inputs=inp, outputs=out)
    model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
    
    return model


# +
es = EarlyStopping(monitor='val_accuracy', mode='auto', restore_best_weights=True, verbose=1, patience=15)

model = get_model(X_train)
model.fit(X_train, y_train, epochs=100, batch_size=256, verbose=2, validation_split=0.1, callbacks=[es])

# +
### GET PREDICTED CLASS ON TEST ###

pred_test = np.argmax(model.predict(X_test), axis=1)
# -

print(classification_report([diz_label[np.argmax(label)] for label in y_test], 
                            [diz_label[label] for label in pred_test]))

# +
cnf_matrix = confusion_matrix([diz_label[np.argmax(label)] for label in y_test], 
                              [diz_label[label] for label in pred_test])

plt.figure(figsize=(7,7))
plot_confusion_matrix(cnf_matrix, classes=list(diz_label.values()))
plt.show()

# +
### DUMMY CLASSIFIER ACCURACY (always stable) ###

sum([diz_label[np.argmax(label)] == 'stable' for label in y_test]) / len(pred_test)

StarcoderdataJupyterScriptsDedupFiltered