# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

import os
import librosa
import random
import numpy as np
from sklearn.model_selection import train_test_split
# %matplotlib inline
import matplotlib.pyplot as plt
import python_speech_features
import tensorflow as tf
import tensorflow.keras as keras
from tensorflow.keras import layers, models

dataset_path = './dataset'
targets = [name for name in os.listdir(dataset_path) if os.path.isdir(os.path.join(dataset_path, name))]
print(targets)

num_samples = 0
for target in targets:
    print(len(os.listdir(os.path.join(dataset_path, target))))
    num_samples += len(os.listdir(os.path.join(dataset_path, target)))
print('Total samples:', num_samples)

audio_files = []
y = []
for index, target in enumerate(targets):
    print(os.path.join(dataset_path, target))
    audio_files.append(os.listdir(os.path.join(dataset_path, target)))
    y.append(np.ones(len(audio_files[index])) * index)
audio_files = [item for sublist in audio_files for item in sublist]
y = [item for sublist in y for item in sublist]
print(len(audio_files))
print(len(y))

audio_files_train, audio_files_val_test, y_orig_train, y_orig_val_train = train_test_split(audio_files, y, test_size = 0.2)
audio_files_val, audio_files_test, y_orig_val, y_orig_test = train_test_split(audio_files_val_test, y_orig_val_train, test_size = 0.5)
print(len(audio_files_train))
print(len(audio_files_val))
print(len(audio_files_test))
print(len(y_orig_train))
print(len(y_orig_val))
val_set_size = len(y_orig_val)
print(len(y_orig_test))
test_set_size = len(y_orig_test)


def calc_mfcc(path):
    signal, fs = librosa.load(path, sr=16000)
    mfccs = python_speech_features.base.mfcc(signal, 
                                             samplerate=fs,
                                             winlen = 0.025,
                                             winstep = 0.01,
                                             numcep = 20,
                                             nfilt = 20,
                                             nfft = 2048,
                                             preemph = 0,
                                             ceplifter = 0,
                                             appendEnergy = False,
                                             winfunc = np.hanning)
    return mfccs.transpose()


def extract_features(in_audio, in_y):
    prob_cnt = 0
    out_x = []
    out_y = []
        
    for index, audio in enumerate(in_audio):
    
        path = os.path.join(dataset_path, targets[int(in_y[index])], audio)
        
        if not path.endswith('.wav'):
            continue

        mfccs = calc_mfcc(path)

        if mfccs.shape[1] == 99:
            out_x.append(mfccs)
            out_y.append(in_y[index])
        else:
            print('Dropped:', index, mfccs.shape)
            prob_cnt += 1
            
    return out_x, out_y, prob_cnt


x_train, y_train, prob = extract_features(audio_files_train, y_orig_train)
print('Removed percentage:', prob / len(y_orig_train))
x_val, y_val, prob = extract_features(audio_files_val, y_orig_val)
print('Removed percentage:', prob / len(y_orig_val))
x_test, y_test, prob = extract_features(audio_files_test, y_orig_test)
print('Removed percentage:', prob / len(y_orig_test))

target_npz_name = "targets.npz"
np.savez(target_npz_name, 
         x_train=x_train, 
         y_train=y_train, 
         x_val=x_val, 
         y_val=y_val, 
         x_test=x_test, 
         y_test=y_test)
feature_sets = np.load(target_npz_name)
feature_sets.files

# # Model Training

target_npz_name = "targets.npz"
feature_sets = np.load(target_npz_name)
feature_sets.files
x_train = feature_sets['x_train']
y_train = feature_sets['y_train']
x_val = feature_sets['x_val']
y_val = feature_sets['y_val']
x_test = feature_sets['x_test']
y_test = feature_sets['y_test']
print(x_train.shape)
print(x_val.shape)
print(x_test.shape)

x_train = x_train.reshape(x_train.shape[0], 
                          x_train.shape[1], 
                          x_train.shape[2], 
                          1)
x_val = x_val.reshape(x_val.shape[0], 
                      x_val.shape[1], 
                      x_val.shape[2], 
                      1)
x_test = x_test.reshape(x_test.shape[0], 
                        x_test.shape[1], 
                        x_test.shape[2], 
                        1)
sample_shape = x_test.shape[1:]
print(x_train.shape)
print(sample_shape)

keras.backend.clear_session()
model = models.Sequential()
model.add(layers.Conv2D(32, (2, 2), activation='relu', input_shape=sample_shape))
model.add(layers.MaxPooling2D(pool_size=(2, 2)))
model.add(layers.Conv2D(32, (2, 2), activation='relu'))
model.add(layers.MaxPooling2D(pool_size=(2, 2)))
model.add(layers.Dropout(0.5))
model.add(layers.Conv2D(64, (2, 2), activation='relu'))
model.add(layers.MaxPooling2D(pool_size=(2, 2)))
model.add(layers.Flatten())
model.add(layers.Dense(128, activation='relu'))
model.add(layers.Dropout(0.5))
model.add(layers.Dense(len(targets), activation='softmax'))

model.summary()

model.compile(
    optimizer=tf.keras.optimizers.Adam(),
    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
    metrics=['accuracy'],
)

history = model.fit(x_train, 
                    y_train, 
                    epochs=100, 
                    batch_size=100, 
                    validation_data=(x_val, y_val))

metrics = history.history
plt.plot(history.epoch, metrics['loss'], metrics['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.show()

models.save_model(model, "cmd_recog_model.h5")

# # Model Testing

model = keras.models.load_model('cmd_recog_model.h5')

for i in range(100, 110):
    print('Answer:', y_test[i], ' Prediction:', model.predict(np.expand_dims(x_test[i], 0)))

model.evaluate(x=x_test, y=y_test)

# +
y_pred = np.argmax(model.predict(x_test), axis=1)
y_true = y_test

test_acc = sum(y_pred == y_true) / len(y_true)
print(f'Test set accuracy: {test_acc:.0%}')
# -

from sklearn.metrics import confusion_matrix
targets = ['on', 'two', 'one', 'three', 'off']
y = np.argmax(model.predict(x_val), axis=1)
cm = confusion_matrix(y_val, y,  normalize="all")
# %matplotlib inline
plt.close()
plt.figure(figsize = (8,8))
plt.imshow(cm, cmap=plt.cm.Blues)
plt.xlabel("Predicted labels")
plt.ylabel("True labels")
plt.xticks(np.arange(0, 5, 1), targets, rotation=90)
plt.yticks(np.arange(0, 5, 1), targets)
plt.tick_params(labelsize=12)
plt.title('Confusion matrix ')
plt.colorbar()
plt.show()



StarcoderdataJupyterScriptsDedupFiltered