Skip to content

Instantly share code, notes, and snippets.

@davidglavas
Created May 21, 2018 12:56
Show Gist options
  • Save davidglavas/60d102bb236cda4f2ff129324352dc86 to your computer and use it in GitHub Desktop.
Save davidglavas/60d102bb236cda4f2ff129324352dc86 to your computer and use it in GitHub Desktop.
Classification section for simple audio classifier blogpost
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