328 lines
13 KiB
Python
328 lines
13 KiB
Python
import json
|
|
from python_speech_features.python_speech_features.base import mfcc
|
|
import numpy as np
|
|
from sklearn.model_selection import train_test_split
|
|
import tensorflow as tf
|
|
import tensorflow.keras as keras
|
|
from keras import backend as K
|
|
from pathlib import Path
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
import statistics
|
|
|
|
# Path to json file that stores MFCCs and subject labels for each processed sample
|
|
DATA_PATH_MFCC = str(Path.cwd()) + "/mfcc_data.json"
|
|
|
|
# Loads data from the json file and reshapes X_data(samples, 1, 208) and y_data(samples, 1)
|
|
# Input: JSON path
|
|
# Ouput: X(mfcc data), y(labels), session_lengths
|
|
def load_data_from_json(data_path, nr_classes):
|
|
|
|
with open(data_path, "r") as fp:
|
|
data = json.load(fp)
|
|
|
|
# convert lists to numpy arraysls
|
|
X = np.array(data['mfcc'])
|
|
#print(X.shape)
|
|
X = X.reshape(X.shape[0], 1, X.shape[1])
|
|
print(X.shape)
|
|
|
|
y = np.array(data["labels"])
|
|
#print(y.shape)
|
|
y = keras.utils.to_categorical(y, nr_classes)
|
|
print(y.shape)
|
|
|
|
session_lengths = np.array(data['session_lengths'])
|
|
#print(session_lengths.shape)
|
|
|
|
|
|
print("Data succesfully loaded!")
|
|
|
|
return X, y, session_lengths
|
|
|
|
# Plots the training history with two subplots. First training and test accuracy, and then
|
|
# loss with respect to epochs
|
|
# Input: History(from model.fit(...))
|
|
# Ouput: None -> plot
|
|
def plot_history(history):
|
|
"""Plots accuracy/loss for training/validation set as a function of the epochs
|
|
:param history: Training history of model
|
|
:return:
|
|
"""
|
|
|
|
fig, axs = plt.subplots(2)
|
|
|
|
# create accuracy sublpot
|
|
axs[0].plot(history.history["accuracy"], label="train accuracy")
|
|
axs[0].plot(history.history["val_accuracy"], label="test accuracy")
|
|
axs[0].set_ylabel("Accuracy")
|
|
axs[0].legend(loc="lower right")
|
|
axs[0].set_title("Accuracy eval")
|
|
|
|
# create error sublpot
|
|
axs[1].plot(history.history["loss"], label="train error")
|
|
axs[1].plot(history.history["val_loss"], label="test error")
|
|
axs[1].set_ylabel("Error")
|
|
axs[1].set_xlabel("Epoch")
|
|
axs[1].legend(loc="upper right")
|
|
axs[1].set_title("Error eval")
|
|
|
|
plt.show()
|
|
|
|
# Takes in data and labels, and splits it into train, validation and test sets by percentage
|
|
# Input: Data, labels, whether to shuffle, % validatiion, % test
|
|
# Ouput: X_train, X_validation, X_test, y_train, y_validation, y_test
|
|
def prepare_datasets_percentsplit(X, y, shuffle_vars, validation_size=0.2, test_size=0.25,):
|
|
|
|
# Create train, validation and test split
|
|
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, shuffle=shuffle_vars)
|
|
X_train, X_validation, y_train, y_validation = train_test_split(X_train, y_train, test_size=validation_size, shuffle=shuffle_vars)
|
|
|
|
return X_train, X_validation, X_test, y_train, y_validation, y_test
|
|
|
|
# Takes in data, labels, and session_lengths and splits it into train and test sets by session_index
|
|
# Input: Data, labels, session_lengths, test_session_index
|
|
# Ouput: X_train, X_test, y_train, y_test
|
|
def prepare_datasets_sessions(X, y, session_lengths, test_session_index=4, nr_subjects=5):
|
|
|
|
session_lengths = session_lengths.tolist()
|
|
|
|
subject_starting_index = 0
|
|
start_test_index = subject_starting_index + sum(session_lengths[0][:test_session_index-1])
|
|
end_test_index = start_test_index + session_lengths[0][test_session_index-1]
|
|
end_subject_index = subject_starting_index + sum(session_lengths[0])
|
|
|
|
# Testing to check correctly slicing
|
|
'''
|
|
print(session_lengths[0], 'Sum:', sum(session_lengths[0]))
|
|
print('Subject start:', subject_starting_index)
|
|
print('Test start:', start_test_index)
|
|
print('Test end:', end_test_index)
|
|
print('Subject end:', end_subject_index, '\n -------')
|
|
'''
|
|
if start_test_index == subject_starting_index:
|
|
X_test = X[start_test_index:end_test_index]
|
|
y_test = y[start_test_index:end_test_index]
|
|
X_train = X[end_test_index:end_subject_index]
|
|
y_train = y[end_test_index:end_subject_index]
|
|
|
|
elif end_test_index == end_subject_index:
|
|
#print(X[subject_starting_index:start_test_index].shape)
|
|
X_train = X[subject_starting_index:start_test_index]
|
|
y_train = y[subject_starting_index:start_test_index]
|
|
X_test = X[start_test_index:end_test_index]
|
|
#print(X[start_test_index:end_test_index].shape, '\n ---')
|
|
y_test = y[start_test_index:end_test_index]
|
|
|
|
else:
|
|
X_train = X[subject_starting_index:start_test_index]
|
|
y_train = y[subject_starting_index:start_test_index]
|
|
X_test = X[start_test_index:end_test_index]
|
|
y_test = y[start_test_index:end_test_index]
|
|
X_train = np.concatenate((X_train, X[end_test_index:end_subject_index]))
|
|
y_train = np.concatenate((y_train, y[end_test_index:end_subject_index]))
|
|
#print(X_train.shape, '\n -------')
|
|
subject_starting_index = max(end_subject_index, end_test_index)
|
|
|
|
for i in range(1, nr_subjects):
|
|
start_test_index = subject_starting_index + sum(session_lengths[i][:test_session_index-1])
|
|
end_test_index = start_test_index + session_lengths[i][test_session_index-1]
|
|
end_subject_index = subject_starting_index + sum(session_lengths[i])
|
|
|
|
# Testing to check correctly slicing
|
|
'''
|
|
print(session_lengths[i], 'Sum:', sum(session_lengths[i]))
|
|
print('Subject start:', subject_starting_index)
|
|
print('Test start:', start_test_index)
|
|
print('Test end:', end_test_index)
|
|
print('Subject end:', end_subject_index, '\n -------')
|
|
'''
|
|
if start_test_index == subject_starting_index:
|
|
X_test = np.concatenate((X_test, X[start_test_index:end_test_index]))
|
|
y_test = np.concatenate((y_test, y[start_test_index:end_test_index]))
|
|
X_train = np.concatenate((X_train, X[end_test_index:end_subject_index]))
|
|
y_train = np.concatenate((y_train, y[end_test_index:end_subject_index]))
|
|
|
|
elif end_test_index == end_subject_index:
|
|
#print(X[subject_starting_index:start_test_index].shape)
|
|
X_train = np.concatenate((X_train, X[subject_starting_index:start_test_index]))
|
|
y_train = np.concatenate((y_train, y[subject_starting_index:start_test_index]))
|
|
#print(X[start_test_index:end_test_index].shape, '\n ---')
|
|
X_test = np.concatenate((X_test, X[start_test_index:end_test_index]))
|
|
y_test = np.concatenate((y_test, y[start_test_index:end_test_index]))
|
|
else:
|
|
X_train = np.concatenate((X_train, X[subject_starting_index:start_test_index]))
|
|
y_train = np.concatenate((y_train, y[subject_starting_index:start_test_index]))
|
|
X_test = np.concatenate((X_test, X[start_test_index:end_test_index]))
|
|
y_test = np.concatenate((y_test, y[start_test_index:end_test_index]))
|
|
X_train = np.concatenate((X_train, X[end_test_index:end_subject_index]))
|
|
y_train = np.concatenate((y_train, y[end_test_index:end_subject_index]))
|
|
#print(X_train.shape, '\n -------')
|
|
subject_starting_index = max(end_subject_index, end_test_index)
|
|
|
|
return X_train, X_test, y_train, y_test
|
|
|
|
# Creates a RNN_LSTM neural network model
|
|
# Input: input shape, classes of classification
|
|
# Ouput: model:Keras.model
|
|
def RNN_LSTM(input_shape, nr_classes=5):
|
|
"""Generates RNN-LSTM model
|
|
:param input_shape (tuple): Shape of input set
|
|
:return model: RNN-LSTM model
|
|
"""
|
|
|
|
# build network topology
|
|
model = keras.Sequential()
|
|
|
|
# 2 LSTM layers
|
|
model.add(keras.layers.LSTM(64, input_shape=input_shape, return_sequences=True))
|
|
model.add(keras.layers.LSTM(64))
|
|
|
|
# dense layer
|
|
model.add(keras.layers.Dense(64, activation='relu'))
|
|
model.add(keras.layers.Dropout(0.3))
|
|
|
|
# output layer
|
|
model.add(keras.layers.Dense(nr_classes, activation='softmax'))
|
|
|
|
return model
|
|
|
|
# Trains the model
|
|
# Input: Keras.model, batch_size, nr epochs, training, and validation data
|
|
# Ouput: History
|
|
def train(model, X_train, y_train, verbose, batch_size=64, epochs=30, X_validation=None, y_validation=None):
|
|
|
|
optimiser = keras.optimizers.Adam(learning_rate=0.0001)
|
|
model.compile(optimizer=optimiser,
|
|
loss='categorical_crossentropy',
|
|
metrics=['accuracy'])
|
|
|
|
if X_validation != None:
|
|
history = model.fit(X_train,
|
|
y_train,
|
|
validation_data=(X_validation, y_validation),
|
|
batch_size=batch_size,
|
|
epochs=epochs,
|
|
verbose=verbose)
|
|
else:
|
|
history = model.fit(X_train,
|
|
y_train,
|
|
batch_size=batch_size,
|
|
epochs=epochs,
|
|
verbose=verbose)
|
|
return history
|
|
|
|
|
|
# Gives nr of datapoints for chosen session
|
|
# Input: session_lengths 2d-list, session_nr, nr of subjects
|
|
# Ouput: int(datapoints)
|
|
def get_nr_in_session(session_lengths:list, session_nr, nr_subjects=5):
|
|
summ = 0
|
|
for i in range(nr_subjects):
|
|
summ += session_lengths[i][session_nr-1]
|
|
return summ
|
|
|
|
# Prints session and training data
|
|
# Input: None
|
|
# Ouput: None -> print
|
|
def print_session_train_data(X_train, X_test, y_train, y_test, session_lengths, session_nr):
|
|
print(X_train.size)
|
|
print(X_train.shape)
|
|
print(X_test.shape)
|
|
print(y_train.shape)
|
|
print(y_test.shape)
|
|
print('Datapoints in session ' + str(session_nr) + ':', get_nr_in_session(session_lengths, session_nr))
|
|
print('Should be remaining:', 2806 - get_nr_in_session(session_lengths, session_nr))
|
|
|
|
# Reshapes training og test data into batches
|
|
# Input: training, test data (and validation), batch_size
|
|
# Ouput: training, test data (and validation)
|
|
def batch_formatting(X_train, X_test, y_train, y_test, batch_size=64, nr_classes=5, X_validation=None, y_validation=None):
|
|
|
|
train_splits = X_train.shape[0] // batch_size
|
|
train_rest = X_train.shape[0] % batch_size
|
|
test_splits = X_test.shape[0] // batch_size
|
|
test_rest = X_test.shape[0] % batch_size
|
|
|
|
X_train = X_train[:-train_rest]
|
|
y_train = y_train[:-train_rest]
|
|
X_test = X_test[:-test_rest]
|
|
y_test = y_test[:-test_rest]
|
|
|
|
X_train_batch = np.reshape(X_train, (batch_size, train_splits, 208))
|
|
y_train_batch = np.reshape(y_train, (batch_size, train_splits, nr_classes))
|
|
X_test_batch = np.reshape(X_test, (batch_size, test_splits, 208))
|
|
y_test_batch = np.reshape(y_test, (batch_size, test_splits, nr_classes))
|
|
|
|
if X_validation != None:
|
|
val_splits = X_validation.shape[0] // batch_size
|
|
val_rest = X_validation.shape[0] % batch_size
|
|
X_validation = X_validation[:-val_rest]
|
|
y_validation = y_validation[:-val_rest]
|
|
X_val_batch = np.reshape(X_validation, (batch_size, val_splits, 208))
|
|
y_val_batch = np.reshape(y_validation, (batch_size, val_splits))
|
|
return X_train_batch, X_test_batch, y_train_batch, y_test_batch, X_val_batch, y_val_batch
|
|
|
|
return X_train_batch, X_test_batch, y_train_batch, y_test_batch
|
|
|
|
def session_cross_validation(X, y, session_lengths, nr_sessions, batch_size=64, epochs=30):
|
|
session_training_results = []
|
|
for i in range(nr_sessions):
|
|
model = RNN_LSTM(input_shape=(1, 208))
|
|
X_train_session, X_test_session, y_train_session, y_test_session = prepare_datasets_sessions(X, y, session_lengths, i)
|
|
train(model, X_train_session, y_train_session, verbose=0, batch_size=batch_size, epochs=epochs)
|
|
test_loss, test_acc = model.evaluate(X_test_session, y_test_session, verbose=2)
|
|
session_training_results.append(test_acc)
|
|
del model
|
|
K.clear_session()
|
|
print('Session', i, 'as test data gives accuracy:', test_acc)
|
|
average_result = statistics.mean((session_training_results))
|
|
return average_result, session_training_results
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
# ----- Load data ------
|
|
# X.shape = (2806, 1, 208)
|
|
# y.shape = (2806, 5)
|
|
# session_lengths.shape = (5, 4)
|
|
X, y, session_lengths = load_data_from_json(DATA_PATH_MFCC, nr_classes=5)
|
|
|
|
# Parameters:
|
|
NR_SUBJECTS = 5
|
|
NR_SESSIONS = 4
|
|
BATCH_SIZE = 64
|
|
EPOCHS = 30
|
|
|
|
# ----- Get prepared data: train, validation, and test ------
|
|
'''
|
|
|
|
X_train, X_test, y_train, y_test = prepare_datasets_sessions(X, y, session_lengths, session_nr)
|
|
print(X_train.shape)
|
|
print(y_train.shape)
|
|
print(X_test.shape)
|
|
print(y_test.shape)
|
|
#print_session_train_data(X_train, X_test, y_train, y_test, session_lengths, session_nr)
|
|
'''
|
|
|
|
#'''
|
|
# ----- Make model ------
|
|
#model = RNN_LSTM(input_shape=(1, 208)) # (timestep, coefficients)
|
|
#model.summary()
|
|
|
|
# ----- Train network ------
|
|
#history = train(model, X_train, y_train, batch_size=batch_size, epochs=30)
|
|
average = session_cross_validation(X, y, session_lengths, NR_SESSIONS, BATCH_SIZE, EPOCHS)
|
|
print('\nCrossvalidated:', average)
|
|
|
|
# plot accuracy/error for training and validation
|
|
#plot_history(history)
|
|
|
|
# ----- Evaluate model on test set ------
|
|
#test_loss, test_acc = model.evaluate(X_test, y_test, verbose=1)
|
|
#print('\nTest accuracy:', test_acc)
|
|
#'''
|
|
|
|
|
|
|