<jupyter_start><jupyter_text>Keyword Transformer (KWT) https://arxiv.org/pdf/2104.00769v2.pdfQuickstart Colab ----> https://colab.research.google.com/drive/1DJLse4a-9i9Ro-UN6KOsO4WBni9P1d1u?usp=sharing<jupyter_code>%load_ext autoreload
%autoreload 2
%matplotlib inline
import os
import pathlib

import matplotlib.pyplot as plt
import numpy as np
import seaborn as sns
import tensorflow as tf

from tensorflow.keras.layers.experimental import preprocessing
from tensorflow.keras import layers
from tensorflow.keras import models
from IPython import display

from utils import mel_features

from tqdm.notebook import tqdm


# Set seed for experiment reproducibility
seed = 42
tf.random.set_seed(seed)
np.random.seed(seed)
gpus = tf.config.experimental.list_physical_devices('GPU')
tf.config.experimental.set_visible_devices(gpus[0], 'GPU')
tf.config.experimental.set_memory_growth(gpus[0], True)
# set some paths and variables
data_dir = pathlib.Path('/home/vitto/tensorflow_datasets/speech_commands_v2') # change your path accordingly

labels = ['house', 'tree', 'dog','nine', 'sheila', 'four','seven','backward','wow','stop','eight','on',
 'down','bed','zero','off','six','one','five','two','marvin','forward','up','right','three','cat', 'learn',
 'bird','yes','no','left', 'follow', 'go', 'happy']<jupyter_output><empty_output><jupyter_text>Import the Dataset Dowload the dataset with 'wget' and copying the download link from ----> http://download.tensorflow.org/data/speech_commands_v0.02.tar.gz TensorFlow Speech Command dataset is a set of one-second .wav audio files, each containing a single spoken English word. These words are from a small set of commands, and are spoken by a variety of different speakers. 20 of the words are core words, while 10 words are auxiliary words that could act as tests for algorithms in ignoring speeches that do not contain triggers. Included along with the 30 words is a collection of background noise audio files. The dataset was originally designed for limited vocabulary speech recognition tasks. The audio clips were originally collected by Google, and recorded by volunteers in uncontrolled locations around the world.<jupyter_code>data_dir.joinpath('README.md').open('r').readlines()
filenames = tf.io.gfile.glob(str(data_dir) + '/*/*')
filenames = tf.random.shuffle(filenames)
num_samples = len(filenames)
print('Number of total examples:', num_samples)
print('Number of examples per label (average):',
      len(tf.io.gfile.listdir(str(data_dir/labels[0]))))
print('Example file tensor:', filenames[0])<jupyter_output><empty_output><jupyter_text>Play random samples<jupyter_code>def decode_audio(audio_binary):                                                                                                                                                                                                                                                                                                                                                                                                                                                     
    audio, s_rate = tf.audio.decode_wav(audio_binary)
    return tf.squeeze(audio, axis=-1)
def play_random(folder, n=10):
    files = os.listdir(data_dir.joinpath(folder))
    ch = np.random.randint(0, len(files), n)
    for i in ch:
        audio_binary = tf.io.read_file(data_dir.joinpath(folder, files[i]).as_posix())  # binary
        waveform = decode_audio(audio_binary) # decoded
        display.display(display.Audio(waveform, rate=16000))                   
play_random('zero')<jupyter_output><empty_output><jupyter_text>Load dataset<jupyter_code>def get_label(file_path):
    parts = tf.strings.split(file_path, os.path.sep)
    return parts[-2]

def get_waveform_and_label(file_path):
    label = get_label(file_path)
    audio_binary = tf.io.read_file(file_path)  # binary
    waveform = decode_audio(audio_binary) # decoded
    return waveform, label

def import_dataset(file_paths):
    waveforms = []
    labels = []
    for file_path in tqdm(file_paths):
        waveform, label = get_waveform_and_label(file_path)
        waveforms.append(waveform)
        labels.append(label)
    return np.array(waveforms), np.array(labels)
X, y = import_dataset(filenames)
print(X.shape, y.shape)<jupyter_output><empty_output><jupyter_text>Visualize the dataset<jupyter_code>rows = 3
cols = 3
n = rows*cols
fig, axes = plt.subplots(rows, cols, figsize=(10, 12))
for i, (audio, label) in enumerate(zip(X[:n], y[:n])):
    r = i // cols
    c = i % cols
    ax = axes[r][c]
    ax.plot(audio.numpy())
    label = label.numpy().decode()
    ax.set_title(label)

plt.show()<jupyter_output><empty_output><jupyter_text>Prepare the Dataset  Extract spectrograms You'll convert the waveform into a spectrogram, which shows frequency changes over time and can be represented as a 2D image. This can be done by applying the short-time Fourier transform (STFT) to convert the audio into the time-frequency domain.A Fourier transform (tf.signal.fft) converts a signal to its component frequencies, but loses all time information. The STFT (tf.signal.stft) splits the signal into windows of time and runs a Fourier transform on each window, preserving some time information, and returning a 2D tensor that you can run standard convolutions on.STFT produces an array of complex numbers representing magnitude and phase. However, you'll only need the magnitude for this tutorial, which can be derived by applying tf.abs on the output of tf.signal.stft.Choose frame_length and frame_step parameters such that the generated spectrogram "image" is almost square. For more information on STFT parameters choice, you can refer to this video on audio signal processing.You also want the waveforms to have the same length, so that when you convert it to a spectrogram image, the results will have similar dimensions. This can be done by simply zero padding the audio clips that are shorter than one second.<jupyter_code>def get_spectrogram(waveform):
    # Padding for files with less than 16000 samples
    zero_padding = tf.zeros([16000] - tf.shape(waveform), dtype=tf.float32)

    # Concatenate audio with padding so that all audio clips will be of the 
    # same length
    waveform = tf.cast(waveform, tf.float32)

    equal_length = tf.concat([waveform, zero_padding], 0)
    equal_length = equal_length.numpy().flatten()
    spectrogram = 30 * (
        mel_features.log_mel_spectrogram(
        equal_length,
            16000,
            log_offset=0.001,
            window_length_secs=0.025,
            hop_length_secs=0.010,
            num_mel_bins=32,
            lower_edge_hertz=60,
            upper_edge_hertz=3800) - np.log(1e-3))

    return spectrogram
def get_spectrogram_df(X, y):
    audios = []
    for audio in tqdm(X):
        spectrogram = get_spectrogram(audio)
        audios.append(tf.expand_dims(spectrogram, -1))
    return np.array(audios), y
X_pre, y_pre = get_spectrogram_df(X.copy(), y.copy())<jupyter_output><empty_output><jupyter_text>Visualize spectrograms<jupyter_code>rows = 3
cols = 3
n = rows*cols
fig, axes = plt.subplots(rows, cols, figsize=(15, 8))
for i, (audio, label) in enumerate(zip(X_pre[:9], y_pre[:9])):
    r = i // cols
    c = i % cols
    ax = axes[r][c]
    ax.imshow(audio.T[0,...,None])
    ax.set_title(label.numpy().decode())

plt.show()<jupyter_output><empty_output><jupyter_text>One-hot encoding<jupyter_code>y_enc = np.array([labels.index(l) for l in y_pre])
y_enc = tf.one_hot(y_enc, len(labels))
print(y_enc.shape)<jupyter_output><empty_output><jupyter_text>Split the dataset<jupyter_code>def train_test_split(X, y, test_size):
    n_test = int(X.shape[0] * test_size)
    X_test = X[:n_test]
    y_test = y[:n_test]
    X_train = X[n_test:]
    y_train = y[n_test:]
    return X_train, X_test, y_train, y_test
X_train, X_test, y_train, y_test = train_test_split(X_pre, y_enc, 0.1)
print(X_train.shape, y_train.shape)
print(X_test.shape, y_test.shape)<jupyter_output><empty_output><jupyter_text>Standardize data<jupyter_code>def standardize(X, y):
    X -= np.mean(X, axis=1, keepdims=True)
    X /= np.std(X, axis=1, keepdims=True)
    return X, y
X_train_norm, y_train_norm = standardize(X_train, y_train)
X_test_norm, y_test_norm = standardize(X_test, y_test)

print(X_train_norm.shape, y_train_norm.shape)
print(X_test_norm.shape, y_test_norm.shape)<jupyter_output><empty_output><jupyter_text>Build KWT Model<jupyter_code>from utils.transformer import TransformerEncoder, PatchClassEmbedding
from utils.tools import CustomSchedule
# model configurations
n_heads = 1 # select kwt 1 2 or 3
d_model = 64 * n_heads
d_ff = d_model * 4
mlp_head_size = 256
dropout = 0.1
activation = tf.nn.gelu
n_layers = 12
def build_kwt(transformer, input_size):
    # Input
    inputs = tf.keras.layers.Input(shape=input_size)
    
    # Linear Projection of Flattened Patches
    x = tf.keras.layers.Dense(d_model)(inputs)
    
    # Position Embedding + Extra learnable class embedding
    x = PatchClassEmbedding(d_model, input_size[0])(x)
    
    # Transformer Model
    x = transformer(x)
    
    # Take only the Extra Learnable Class
    x = tf.keras.layers.Lambda(lambda x: x[:,0,:])(x)
    
    # MLP Head
    x = tf.keras.layers.Dense(mlp_head_size)(x)
    outputs = tf.keras.layers.Dense(len(labels), activation='softmax')(x)
    
    return tf.keras.models.Model(inputs, outputs)

transformer = TransformerEncoder(d_model, n_heads, d_ff, dropout, activation, n_layers)
model = build_kwt(transformer, input_size=X_train_norm.shape[1:-1])
model.summary()<jupyter_output><empty_output><jupyter_text>Train the Model<jupyter_code># set some variables
batch_size = 64
n_epochs = 50
lr = CustomSchedule(d_model, warmup_steps=20000.0)
optimizer = tf.keras.optimizers.Adam(learning_rate=lr)
model.compile(
    optimizer=optimizer,
    loss=tf.keras.losses.CategoricalCrossentropy(from_logits=False, label_smoothing=0.1),
    metrics=['accuracy'])
history = model.fit(
    X_train_norm, y_train_norm, 
    validation_data=(X_test_norm, y_test_norm),  
    epochs=n_epochs, initial_epoch=0)
metrics = history.history
plt.plot(history.epoch, metrics['loss'], metrics['val_loss'])
plt.legend(['loss', 'val_loss'])
plt.show()<jupyter_output><empty_output><jupyter_text>Test the model<jupyter_code>model.evaluate(X_test, y_test)<jupyter_output><empty_output><jupyter_text>Convert to TensorFlow-Lite<jupyter_code>converter = tf.lite.TFLiteConverter.from_keras_model(model)
# start conversion
tflite_model = converter.convert()
# save model
tflite_model_file = pathlib.Path('bin/model_fp32_kwt.tflite')
tflite_model_file.write_bytes(tflite_model)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10691941357736483, 'fraction_numerical': 0.01354001307311607, 'mean_word_length': 3.9675324675324677, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 59, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 23, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '546134', 'license': 'Apache-2.0', 'n_tokens_mistral': 3625, 'n_tokens_neox': 3466, 'n_words': 848, 'path': 'Notebooks/1 - TF2.X DeepAI-Quickie/tf_2_keyword_transformer.ipynb', 'repo_name': 'EscVM/EscVM_YT', 'size': '10709'}