-
-
Save davidglavas/60d102bb236cda4f2ff129324352dc86 to your computer and use it in GitHub Desktop.
Classification section for simple audio classifier blogpost
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import glob | |
import os | |
from matplotlib import pyplot as plt | |
import numpy as np | |
import pandas as pd | |
import seaborn as sns | |
from sklearn import metrics | |
import tensorflow as tf | |
from tensorflow.python.data import Dataset | |
featureVectorSize = 140 | |
tf.logging.set_verbosity(tf.logging.ERROR) | |
pd.options.display.max_rows = 10 | |
pd.options.display.float_format = '{:.1f}'.format | |
def construct_feature_columns(): | |
"""Construct the TensorFlow Feature Columns. | |
Returns: | |
A set of feature columns | |
""" | |
return set([tf.feature_column.numeric_column('audioFeatures', shape=featureVectorSize)]) | |
def create_training_input_fn(features, labels, batch_size, num_epochs=None, shuffle=True): | |
"""A custom input_fn for sending our feature vectors to the estimator for training. | |
Args: | |
features: The training features. | |
labels: The training labels. | |
batch_size: Batch size to use during training. | |
Returns: | |
A function that returns batches of training features and labels during training. | |
""" | |
def _input_fn(num_epochs=num_epochs, shuffle=True): | |
idx = np.random.permutation(features.index) | |
raw_features = {"audioFeatures": features.reindex(idx)} | |
raw_labels = np.array(labels[idx]) | |
ds = Dataset.from_tensor_slices((raw_features, raw_labels)) | |
ds = ds.batch(batch_size).repeat(num_epochs) | |
if shuffle: | |
ds = ds.shuffle(10000) | |
# Returns the next batch of data. | |
feature_batch, label_batch = ds.make_one_shot_iterator().get_next() | |
return feature_batch, label_batch | |
return _input_fn | |
def create_predict_input_fn(features, labels, batch_size): | |
"""A custom input_fn for sending our feature vectors to the estimator for predictions. | |
Args: | |
features: The features to base predictions on. | |
labels: The labels of the prediction examples. | |
Returns: | |
A function that returns features and labels for predictions. | |
""" | |
def _input_fn(): | |
raw_features = {"audioFeatures": features.values} | |
raw_labels = np.array(labels) | |
ds = Dataset.from_tensor_slices((raw_features, raw_labels)) | |
ds = ds.batch(batch_size) | |
# Returns the next batch of data. | |
feature_batch, label_batch = ds.make_one_shot_iterator().get_next() | |
return feature_batch, label_batch | |
return _input_fn | |
def train_nn_classification_model( | |
learning_rate, | |
regularization_strength, | |
steps, | |
batch_size, | |
hidden_units, | |
training_examples, | |
training_labels, | |
validation_examples, | |
validation_labels, | |
model_Name='no_Name'): | |
"""Trains a neural network classification model. | |
In addition to training, this function also prints training progress information, | |
a plot of the training and validation loss over time, as well as a confusion | |
matrix. | |
Args: | |
learning_rate: An `int`, the learning rate to use. | |
regularization_strength: A float, the regularization strength. | |
steps: A non-zero `int`, the total number of training steps. A training step | |
consists of a forward and backward pass using a single batch. | |
batch_size: A non-zero `int`, the batch size. | |
hidden_units: A `list` of int values, specifying the number of units in each layer. | |
training_examples: A `DataFrame` containing the training features. | |
training_labels: A `DataFrame` containing the training labels. | |
validation_examples: A `DataFrame` containing the validation features. | |
validation_labels: A `DataFrame` containing the validation labels. | |
model_Name: A `string` containing the model's name which is used when storing the loss curve and confusion | |
matrix plots. | |
Returns: | |
The trained `DNNClassifier` object. | |
""" | |
periods = 10 | |
steps_per_period = steps / periods | |
# Create the input functions. | |
predict_training_input_fn = create_predict_input_fn( | |
training_examples, training_labels, batch_size) | |
predict_validation_input_fn = create_predict_input_fn( | |
validation_examples, validation_labels, batch_size) | |
training_input_fn = create_training_input_fn( | |
training_examples, training_labels, batch_size) | |
# Create feature columns. | |
feature_columns = construct_feature_columns() | |
# Create a DNNClassifier object. | |
my_optimizer = tf.train.ProximalAdagradOptimizer( | |
learning_rate=learning_rate, | |
l2_regularization_strength=regularization_strength # can be swapped for l1 regularization | |
) | |
classifier = tf.estimator.DNNClassifier( | |
feature_columns=feature_columns, | |
n_classes=10, | |
hidden_units=hidden_units, | |
optimizer=my_optimizer, | |
config=tf.contrib.learn.RunConfig(keep_checkpoint_max=1) | |
) | |
# Train the model, but do so inside a loop so that we can periodically assess loss metrics. | |
print("Training model...") | |
print("LogLoss error (on validation data):") | |
training_errors = [] | |
validation_errors = [] | |
for period in range(0, periods): | |
# Train the model, starting from the prior state. | |
classifier.train( | |
input_fn=training_input_fn, | |
steps=steps_per_period | |
) | |
# Use the current model to make predictions on both, the training and validation set. | |
training_predictions = list(classifier.predict(input_fn=predict_training_input_fn)) | |
training_pred_class_id = np.array([item['class_ids'][0] for item in training_predictions]) | |
training_pred_one_hot = tf.keras.utils.to_categorical(training_pred_class_id, 10) | |
validation_predictions = list(classifier.predict(input_fn=predict_validation_input_fn)) | |
validation_pred_class_id = np.array([item['class_ids'][0] for item in validation_predictions]) | |
validation_pred_one_hot = tf.keras.utils.to_categorical(validation_pred_class_id, 10) | |
# Use predictions to compute training and validation errors. | |
training_log_loss = metrics.log_loss(training_labels, training_pred_one_hot) | |
validation_log_loss = metrics.log_loss(validation_labels, validation_pred_one_hot) | |
# Print validation error of current model. | |
print(" period %02d : %0.2f" % (period, validation_log_loss)) | |
# Store loss metrics so we can plot them later. | |
training_errors.append(training_log_loss) | |
validation_errors.append(validation_log_loss) | |
print("Model training finished.") | |
# Remove event files to save disk space. | |
_ = map(os.remove, glob.glob(os.path.join(classifier.model_dir, 'events.out.tfevents*'))) | |
# Compute predictions of final model. | |
final_predictions = classifier.predict(input_fn=predict_validation_input_fn) | |
final_predictions = np.array([item['class_ids'][0] for item in final_predictions]) | |
# Evaluate predictions of final model. | |
accuracy = metrics.accuracy_score(validation_labels, final_predictions) | |
print("Final accuracy (on validation data): %0.2f" % accuracy) | |
# Output a graph of loss metrics over periods. | |
plt.ylabel("LogLoss") | |
plt.xlabel("Periods") | |
plt.title("LogLoss vs. Periods") | |
plt.plot(training_errors, label="training") | |
plt.plot(validation_errors, label="validation") | |
plt.legend() | |
# plt.show() # blocks execution | |
plt.savefig('Results\\' + model_Name + '_loss_curve.png', bbox_inches='tight') | |
plt.gcf().clear() | |
# Create a confusion matrix. | |
cm = metrics.confusion_matrix(validation_labels, final_predictions) | |
# Normalize the confusion matrix by the number of samples in each class (rows). | |
cm_normalized = cm.astype("float") / cm.sum(axis=1)[:, np.newaxis] | |
ax = sns.heatmap(cm_normalized, cmap="bone_r") | |
ax.set_aspect(1) | |
plt.title("Confusion matrix") | |
plt.ylabel("True label") | |
plt.xlabel("Predicted label") | |
# plt.show() # blocks execution | |
plt.savefig('Results\\' + model_Name + '_confusion_matrix.png', bbox_inches='tight') | |
plt.gcf().clear() | |
return classifier | |
def mean_normalize(featureMatrix): | |
"""Normalizes each feature (column in the matrix) by making them have zero mean and unit variance. | |
Args: | |
featureMatrix: Each row is a feature vector corresponding to one audio recording. Each column | |
represents values of different feature vectors for one feature. | |
Returns: | |
Original matrix but values are modified such that each feature has zero mean and unit variance. | |
""" | |
mean = np.mean(featureMatrix, axis=0) # compute mean of each column (feature) | |
std = np.std(featureMatrix, axis=0, ddof=1) # compute sample std of each column (feature) | |
featureMatrix -= mean # subtract each column's mean from every value in the corresponding column | |
featureMatrix /= std # divide values in each column with the corresponding sample std for that column | |
return featureMatrix | |
def k_fold_cross_validation(training_set_names, validation_set_names): | |
""" | |
Performs a k-fold cross validation. Trains k different models and lets you know how they perform by using the | |
corresponding validation set. | |
:param training_set_names: List of training sets stored as tuples. Each tuple is a pair of strings, first | |
element is the name of the training examples, second element is the name of the corresponding training labels. | |
:param validation_set_names: List of validation sets stored as tuples. Each tuple is a pair of strings, first | |
element is the name of the validation examples, second element is the name of the corresponding validation labels. | |
""" | |
# group each training set with its corresponding validation set | |
folds = zip(training_set_names, validation_set_names) | |
for (training_name, validation_name) in folds: | |
training_examples, training_labels = load_features(training_name) | |
validation_examples, validation_labels = load_features(validation_name) | |
print("#####################################################################################") | |
print("Model is trained with ", training_name[0], "and validated with", validation_name[0]) | |
train_nn_classification_model( | |
learning_rate=0.003, | |
regularization_strength=0.1, | |
steps=5000, | |
batch_size=32, | |
hidden_units=[120], | |
training_examples=training_examples, | |
training_labels=training_labels, | |
validation_examples=validation_examples, | |
validation_labels=validation_labels, | |
model_Name=training_name[0]) | |
def load_features(dataset_name): | |
""" | |
Unpickles the given examples and labels. Mean normalizes the examples. | |
:param dataset_name: Pair of names referring to an example and corresponding label set. | |
:return: Actual dataset as a pair, first element are the mean normalized examples (pandas DataFrame), second | |
element are the corresponding labels (pandas Series). | |
""" | |
examples_path = 'Extracted_Features\\' + dataset_name[0] | |
# unpickles and mean normalizes examples | |
examples = mean_normalize(pd.read_pickle(examples_path)) | |
# unpickles labels | |
labels_path = 'Extracted_Features\\' + dataset_name[1] | |
labels = pd.read_pickle(labels_path) | |
return examples, labels | |
def perform_cv(): | |
# order in training_set_names matches the order in validation_set_names | |
training_set_names = [('notFold1_features.pkl', 'notFold1_labels.pkl'), | |
('notFold2_features.pkl', 'notFold2_labels.pkl'), | |
('notFold3_features.pkl', 'notFold3_labels.pkl'), | |
('notFold4_features.pkl', 'notFold4_labels.pkl'), | |
('notFold5_features.pkl', 'notFold5_labels.pkl'), | |
('notFold6_features.pkl', 'notFold6_labels.pkl'), | |
('notFold7_features.pkl', 'notFold7_labels.pkl'), | |
('notFold8_features.pkl', 'notFold8_labels.pkl'), | |
('notFold9_features.pkl', 'notFold9_labels.pkl'), | |
('notFold10_features.pkl', 'notFold10_labels.pkl')] | |
validation_set_names = [('fold1_features.pkl', 'fold1_labels.pkl'), ('fold2_features.pkl', 'fold2_labels.pkl'), | |
('fold3_features.pkl', 'fold3_labels.pkl'), ('fold4_features.pkl', 'fold4_labels.pkl'), | |
('fold5_features.pkl', 'fold5_labels.pkl'), ('fold6_features.pkl', 'fold6_labels.pkl'), | |
('fold7_features.pkl', 'fold7_labels.pkl'), ('fold8_features.pkl', 'fold8_labels.pkl'), | |
('fold9_features.pkl', 'fold9_labels.pkl'), ('fold10_features.pkl', 'fold10_labels.pkl')] | |
k_fold_cross_validation(training_set_names, validation_set_names) | |
# performs the 10-fold cross-validation, make sure the extracted features are stored in the corresponding directory. | |
# perform_cv() | |
# for hyperparameter searching | |
def test_run(): | |
# unpickle and prepare training data | |
training_examples = mean_normalize(pd.read_pickle('Extracted_Features\\notFold1_features.pkl')) | |
training_labels = pd.read_pickle('Extracted_Features\\notFold1_labels.pkl') | |
# unpickle and prepare validation data | |
validation_examples = mean_normalize(pd.read_pickle('Extracted_Features\\fold1_features.pkl')) | |
validation_labels = pd.read_pickle('Extracted_Features\\fold1_labels.pkl') | |
for learning_rate in [0.001, 0.003, 0.01, 0.03, 0.1, 0.3]: | |
for regularization_strength in [0.0, 0.003, 0.03, 0.3]: | |
print("##########################################################################") | |
print("Learning rate:", learning_rate) | |
print("Regularization:", regularization_strength) | |
train_nn_classification_model( | |
learning_rate=learning_rate, | |
regularization_strength=regularization_strength, | |
steps=10000, | |
batch_size=32, | |
hidden_units=[120], | |
training_examples=training_examples, | |
training_labels=training_labels, | |
validation_examples=validation_examples, | |
validation_labels=validation_labels, | |
model_Name='lr ' + str(learning_rate) + ", reg " + str(regularization_strength)) | |
# test_run() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment