Simple speech recognition

Audio 데이터를 다뤄서 학습하는 방법을 배워보도록 합시다.

머신러닝 모델을 학습하는 방법은 아래와 같습니다.

• Examine and understand data
• Build an input pipeline
• Build the model
• Train the model
• Test the model
• Improve the model and repeat the process
• 모델 완성 후 평가 지표에 따라서 모델을 평가해 봅시다.

Dataset

• 구글 드라이브 링크를 이용해 Dataset을 다운받습니다.
  • 데이터셋 드라이브 링크

# 기본으로 설정되어있는 Dataset path
DATASET_PATH = "/content/drive/MyDrive/Datasets"

Project 설명

Task

• 1초 길이의 오디오 음성데이터를 이용해 단어를 분류하는 것이 목표입니다.
• 주어진 데이터를 이용해 딥러닝 트레이닝 과정을 구현해 보는것이 목표입니다.
• This code is borrowed from Kaggle/TensorFlow Speech Recognition Challenge.
• This is version 0.01 of the data set containing 64,727 audio files, released on August 3rd 2017.

Baseline

• ResNet 구조와 유사한 skip connection 구조를 구현해 보자.
• 오버피팅을 방지하기 위한 다양한 방법들을 사용해보자.
• Training
  • tf.data.dataset과 model.fit()을 사용

Import packages

• 우리가 사용할 packages 를 import 하는 부분 입니다.
• 필요에 따른 packages를 선언합니다.

import tensorflow as tf
from tensorflow.keras import layers
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split

import librosa
import librosa.display
import matplotlib.pyplot as plt

import os
from os.path import isdir, join

import random
import copy
import sys

tf.__version__

#2.15.0

Import modules

• Colab 적용을 위한 변수 지정 및 드라이브 마운트

use_colab = True
assert use_colab in [True, False]

from google.colab import drive
drive.mount('/content/drive')

# Mounted at /content/drive

Load dataset

• 사용할 데이터셋을 살펴봅시다.

if use_colab:
    DATASET_PATH = "/content/drive/MyDrive/dataset/wave_cls"
else:
    DATASET_PATH = "./Datasets"

if not os.path.isdir(DATASET_PATH):
    os.makedirs(DATASET_PATH)

speech_data = np.load(os.path.join(DATASET_PATH, "speech_wav_8000.npz"), allow_pickle=True) #npy, npz

• npz 형태의 파일은 npy의 압축형태이며, files 이름 내에 데이터를 저장하고 불러올 수 있습니다.
• files를 출력하면, 데이터가 어떤 key값으로 저장되어 있는지 확인할 수 있습니다!

print(speech_data.files)

#['wav_vals', 'label_vals']

• 각 데이터가 어떤 형태로 저장되어 있는지 확인해봅시다.

print(speech_data["wav_vals"].shape, speech_data["label_vals"].shape)

# (50620, 8000) (50620, 1)

• 숫자로 이뤄진 데이터가 진짜 오디오 데이터가 맞는지 확인해봅시다.

idx = 219 # 총 50,620개의 오디오 파일이 존재
test_audio = speech_data["wav_vals"][idx]
test_labels = speech_data["label_vals"][idx]

import IPython.display as ipd

sr = 8000 # 1초동안 재생되는 샘플의 갯수
data = test_audio
print(data[2990:3000])
plt.plot(data)

print(test_labels)
ipd.Audio(data, rate=sr)

Model dataset setting

• 변환된 데이터를 이용해서 학습에 활용할 데이터셋을 설정
• 전체 데이터셋을 train_test_split 함수를 이용해서 train set, test set으로 나눠보자
  • train_test_split function

sklearn.model_selection.train_test_split(train, label,
                                         test_size=None,
                                         shuffle=True)
                                        ```

sr = 8000
train_wav, test_wav, train_label, test_label = train_test_split(#,
                                                                #,
                                                                test_size=0.1,
                                                                shuffle=True)

# for convolution layers
train_wav = train_wav.reshape([-1, #, 1])
test_wav = test_wav.reshape([-1, #, 1])

print(train_wav.shape)
print(test_wav.shape)
print(train_label.shape)
print(test_label.shape)

Label 데이터를 구분해보자

set(speech_data["label_vals"].flatten())

{'down',
 'go',
 'left',
 'no',
 'off',
 'on',
 'right',
 'silence',
 'stop',
 'unknown',
 'up',
 'yes'}

# del raw dataset for memory
del speech_data

• 총 12개의 클래스르 분류하는 작업이된다.
• unknown과 silence에는 target list 이외의 단어가 들어간다. (혹은 노이즈)

#target_list
label_value = ['yes', 'no', 'up', 'down', 'left', 'right', 'on', 'off', 'stop', 'go', 'unknown', 'silence']

new_label_value = dict()
for i, l in enumerate(label_value):
    new_label_value[l] = i
label_value = new_label_value

label_value

{'down': 3,
 'go': 9,
 'left': 4,
 'no': 1,
 'off': 7,
 'on': 6,
 'right': 5,
 'silence': 11,
 'stop': 8,
 'unknown': 10,
 'up': 2,
 'yes': 0}

• Text 형태로 이뤄진 label 데이터를 학습에 사용하기 위해서 idx로 변환해 줍니다.
• 메모리를 절약하기 위해 이미 사용중인 변수를 변환
  • text 형태의 데이터를 int 형으로 변환
  • 한번 변환 후에는 text 형태가 아니기때문에 오류가 발생!

temp = []
for v in train_label:
    temp.append(label_value[v[0]]) # ["down"] => "down"
train_label = np.array(temp)

temp = []
for v in test_label:
    temp.append(label_value[v[0]])
test_label = np.array(temp)

del temp

print('Train_Wav Demension : ' + str(np.shape(train_wav)))
print('Train_Label Demension : ' + str(np.shape(train_label)))
print('Test_Wav Demension : ' + str(np.shape(test_wav)))
print('Test_Label Demension : ' + str(np.shape(test_label)))
print('Number Of Labels : ' + str(len(label_value)))

train_label[0:10]

Checkpoint setting

• 학습 전반에서 사용할 checkpoint dir을 설정한다.

# the save point
if use_colab:
    checkpoint_dir ='./drive/MyDrive/train_ckpt/wave/exp1'
    if not os.path.isdir(checkpoint_dir):
        os.makedirs(checkpoint_dir)
else:
    checkpoint_dir = 'wave/exp1'

Dataset 구성

• index로 구분된 label을 학습에 사용하기 위해선 두가지 방법이 있습니다.
  • Sparse loss function 사용
  • One hot vector로 변환
• One hot vector를 이용한 방법을 실습해보겠습니다.
• classes 구성을 위한 tf.one_hot
  • tf.one_hot function

tf.one_hot(
    indices, depth,
    on_value=None,
    off_value=None,
    axis=None,
    dtype=None,
    name=None
)

def one_hot_label(wav, label):
    label = tf.one_hot(label, depth=#)
    return wav, label

# [0, 1, 2]

# 0 => [1, 0, 0]
# 1 => [0, 1, 0]
# 2 => [0, 0, 1]

batch_size = 16

# for train
train_dataset = tf.data.Dataset.from_tensor_slices((#, #))
train_dataset = train_dataset.map(#)
train_dataset = train_dataset.shuffle(10000).repeat().batch(batch_size)
print(train_dataset)

# for test
test_dataset = tf.data.Dataset.from_tensor_slices((#, #))
test_dataset = test_dataset.map(#)
test_dataset = test_dataset.batch(batch_size)
print(test_dataset)

for t, l in train_dataset.take(1):
    print(l)

Dataset 구성 검증

<BatchDataset shapes: ((None, 8000, 1), (None, 12)), types: (tf.float32, tf.float32)>
<BatchDataset shapes: ((None, 8000, 1), (None, 12)), types: (tf.float32, tf.float32)>

Model 구현

• Wave 파일 데이터를 이용해 학습을 할 수 있는 모델을 구현합니다.
  • inputs = [batch_size, 8000, 1]
  • conv1 = [batch_size, 4000, 16]
  • conv2 = [batch_size, 2000, 32]
  • conv3 = [batch_size, 1000, 64]
  • desne = [batch_size, 64]
  • output = [batch_size, 12]

# TODO

# layers.Dense(12, activation='softmax') -> tf.keras.losses.CategoricalCrossentropy(from_logits=False)
# layers.Dense(12) -> tf.keras.losses.CategoricalCrossentropy(from_logits=True)

model.compile(loss=tf.keras.losses.CategoricalCrossentropy(from_logits=#), # Ture or False
             optimizer="adam",# 1e-3 # tf.keras.optimizers.adam(1e-4)
             metrics=['accuracy'])

# without training, just inference a model
predictions = model(train_wav[0:1], training=False)
print("Predictions: ", predictions.numpy())

model.summary()

Model training

• 모델 체크포인트로 저장공간을 확인 후 학습을 진행합니다.

cp_callback = tf.keras.callbacks.ModelCheckpoint(checkpoint_dir,
                                                 save_weights_only=True,
                                                 monitor='val_loss',
                                                 mode='auto',
                                                 save_best_only=True,
                                                 verbose=1)

• Learning rate를 조절하는 callback 함수도 있습니다.
• Cos 그래프 형태로 Learning rate를 줄여주는 함수를 확인해봅시다.

tf.keras.optimizers.schedules.CosineDecay(
    initial_learning_rate, decay_steps, alpha=0.0, name=None
)

• 코사인 디케이 함수가 어떻게 동작하는지 확인해봅시다

def decayed_learning_rate(step):
step = min(step, decay_steps)
cosine_decay = 0.5 * (1 + cos(pi * step / decay_steps))
decayed = (1 - alpha) * cosine_decay + alpha
return initial_learning_rate * decayed

decay_steps = 100
lr_decayed_fn = tf.keras.optimizers.schedules.CosineDecay(
    initial_learning_rate, decay_steps)

cos_decay = tf.keras.experimental.CosineDecay(#1e-3, # 초기 learning rate
                                              #decay_steps) # 에폭기준으로 언제 0이 될건지
lr_callback = tf.keras.callbacks.LearningRateScheduler(cos_decay, verbose=1)

# using `tf.data.Dataset`, model.fit_generator
history = model.fit(# dataset,
                    steps_per_epoch=#,
                    epochs=#,
                    callbacks=[#,#],
                    validation_data=#,
                    validation_steps=#)

# fit(
#     x=None, y=None, batch_size=None, epochs=1, verbose='auto',
#     callbacks=None, validation_split=0.0, validation_data=None, shuffle=True,
#     class_weight=None, sample_weight=None, initial_epoch=0, steps_per_epoch=None,
#     validation_steps=None, validation_batch_size=None, validation_freq=1,
#     max_queue_size=10, workers=1, use_multiprocessing=False
# )

# fit_generator(
#     generator, steps_per_epoch=None, epochs=1, verbose=1, callbacks=None,
#     validation_data=None, validation_steps=None, validation_freq=1,
#     class_weight=None, max_queue_size=10, workers=1, use_multiprocessing=False,
#     shuffle=True, initial_epoch=0
# )

학습 결과 확인

• model fit의 return 값인 history에서 학습에 대한 결과를 확인해보자

acc = history.history['accuracy']
val_acc = history.history['val_accuracy']

loss=history.history['loss']
val_loss=history.history['val_loss']

epochs_range = range(len(acc))

plt.figure(figsize=(8, 8))
plt.subplot(1, 2, 1)
plt.plot(epochs_range, acc, label='Training Accuracy')
plt.plot(epochs_range, val_acc, label='Validation Accuracy')
plt.legend(loc='lower right')
plt.title('Training and Validation Accuracy')

plt.subplot(1, 2, 2)
plt.plot(epochs_range, loss, label='Training Loss')
plt.plot(epochs_range, val_loss, label='Validation Loss')
plt.legend(loc='upper right')
plt.title('Training and Validation Loss')
plt.show()

Evaluation

• Test dataset을 이용해서 모델의 성능을 평가합니다.

model.load_weights(checkpoint_dir)

results = model.evaluate(test_dataset)

스코어 결과

• 위의 스코어는 분류모델에 적용되는 스코어입니다.
• 모델의 크기 (MB) 와 정확도를 이용해 스코어를 출력합니다.

def final_score():
    print("Model params num : " + str(model.count_params()))
    print("Accuracy : " + str(results[1]))

    s = (model.count_params() * 32) / (1024 ** 2)
    score = 50 * (results[1] + min((1/s), 1))

    print("score : " + str(score))

final_score()