title speaker recognition fadi badinehttps twitter comfadibadine date created 14062020 last modified 19072023 description classify speakers using fast fourier transform fft and a 1d convnet accelerator gpu converted to keras 3 by fadi badinehttps twitter comfadibadine introduction this example demonstrates how to create a model to classify speakers from the frequency domain representation of speech recordings obtained via fast fourier transform fft it shows the following how to use tf data to load preprocess and feed audio streams into a model how to create a 1d convolutional network with residual connections for audio classification our process we prepare a dataset of speech samples from different speakers with the speaker as label we add background noise to these samples to augment our data we take the fft of these samples we train a 1d convnet to predict the correct speaker given a noisy fft speech sample note this example should be run with tensorflow 2 3 or higher or tfnightly the noise samples in the dataset need to be resampled to a sampling rate of 16000 hz before using the code in this example in order to do this you will need to have installed ffmpeg setup get the data from https www kaggle comkongaevansspeakerrecognitiondataset and save it to speakerrecognitiondataset zip then unzip it to 16000pcmspeeches shell kaggle datasets download d kongaevansspeakerrecognitiondataset unzip qq speakerrecognitiondataset zip the folders in which we will put the audio samples and the noise samples percentage of samples to use for validation seed to use when shuffling the dataset and the noise the sampling rate to use this is the one used in all the audio samples we will resample all the noise to this sampling rate this will also be the output size of the audio wave samples since all samples are of 1 second long the factor to multiply the noise with according to noisysample sample noise prop scale where prop sampleamplitude noiseamplitude data preparation the dataset is composed of 7 folders divided into 2 groups speech samples with 5 folders for 5 different speakers each folder contains 1500 audio files each 1 second long and sampled at 16000 hz background noise samples with 2 folders and a total of 6 files these files are longer than 1 second and originally not sampled at 16000 hz but we will resample them to 16000 hz we will use those 6 files to create 354 1secondlong noise samples to be used for training let s sort these 2 categories into 2 folders an audio folder which will contain all the perspeaker speech sample folders a noise folder which will contain all the noise samples before sorting the audio and noise categories into 2 folders we have the following directory structure maindirectory speakera speakerb speakerc speakerd speakere other backgroundnoise after sorting we end up with the following structure maindirectory audio speakera speakerb speakerc speakerd speakere noise other backgroundnoise if folder is audio or noise do nothing if folder is one of the folders that contains noise samples move it to the noise folder otherwise it should be a speaker folder then move it to audio folder noise preparation in this section we load all noise samples which should have been resampled to 16000 we split those noise samples to chunks of 16000 samples which correspond to 1 second duration each get the list of all noise files resample all noise samples to 16000 hz note that this requires ffmpeg split noise into chunks of 16 000 steps each number of slices of 16000 each that can be generated from the noise sample dataset generation constructs a dataset of audios and labels pathds tf data dataset fromtensorslicesaudiopaths audiods pathds map lambda x pathtoaudiox numparallelcallstf data autotune labelds tf data dataset fromtensorsliceslabels return tf data dataset zipaudiods labelds def pathtoaudiopath create a random tensor of the same size as audio ranging from 0 to the number of noise stream samples that we have get the amplitude proportion between the audio and the noise adding the rescaled noise to audio since tf signal fft applies fft on the innermost dimension we need to squeeze the dimensions and then expand them again after fft return the absolute value of the first half of the fft which represents the positive frequencies get the list of audio file paths along with their corresponding labels shuffle split into training and validation create 2 datasets one for training and the other for validation add noise to the training set transform audio wave to the frequency domain using audiotofft model definition shortcut compile the model using adam s default learning rate add callbacks earlystopping to stop training when the model is not enhancing anymore modelcheckpoint to always keep the model that has the best valaccuracy training evaluation we get 98 validation accuracy demonstration let s take some samples and predict the speaker compare the prediction with the real speaker listen to the audio to see that despite the samples being noisy the model is still pretty accurate get the signal fft predict take random samples for every sample print the true and predicted label as well as run the voice with the noise introduction this example demonstrates how to create a model to classify speakers from the frequency domain representation of speech recordings obtained via fast fourier transform fft it shows the following how to use tf data to load preprocess and feed audio streams into a model how to create a 1d convolutional network with residual connections for audio classification our process we prepare a dataset of speech samples from different speakers with the speaker as label we add background noise to these samples to augment our data we take the fft of these samples we train a 1d convnet to predict the correct speaker given a noisy fft speech sample note this example should be run with tensorflow 2 3 or higher or tf nightly the noise samples in the dataset need to be resampled to a sampling rate of 16000 hz before using the code in this example in order to do this you will need to have installed ffmpeg setup get the data from https www kaggle com kongaevans speaker recognition dataset and save it to speaker recognition dataset zip then unzip it to 16000_pcm_speeches shell kaggle datasets download d kongaevans speaker recognition dataset unzip qq speaker recognition dataset zip the folders in which we will put the audio samples and the noise samples percentage of samples to use for validation seed to use when shuffling the dataset and the noise the sampling rate to use this is the one used in all the audio samples we will resample all the noise to this sampling rate this will also be the output size of the audio wave samples since all samples are of 1 second long the factor to multiply the noise with according to noisy_sample sample noise prop scale where prop sample_amplitude noise_amplitude data preparation the dataset is composed of 7 folders divided into 2 groups speech samples with 5 folders for 5 different speakers each folder contains 1500 audio files each 1 second long and sampled at 16000 hz background noise samples with 2 folders and a total of 6 files these files are longer than 1 second and originally not sampled at 16000 hz but we will resample them to 16000 hz we will use those 6 files to create 354 1 second long noise samples to be used for training let s sort these 2 categories into 2 folders an audio folder which will contain all the per speaker speech sample folders a noise folder which will contain all the noise samples before sorting the audio and noise categories into 2 folders we have the following directory structure main_directory speaker_a speaker_b speaker_c speaker_d speaker_e other _background_noise_ after sorting we end up with the following structure main_directory audio speaker_a speaker_b speaker_c speaker_d speaker_e noise other _background_noise_ if folder is audio or noise do nothing if folder is one of the folders that contains noise samples move it to the noise folder otherwise it should be a speaker folder then move it to audio folder noise preparation in this section we load all noise samples which should have been resampled to 16000 we split those noise samples to chunks of 16000 samples which correspond to 1 second duration each get the list of all noise files resample all noise samples to 16000 hz note that this requires ffmpeg split noise into chunks of 16 000 steps each number of slices of 16000 each that can be generated from the noise sample dataset generation constructs a dataset of audios and labels reads and decodes an audio file create a random tensor of the same size as audio ranging from 0 to the number of noise stream samples that we have get the amplitude proportion between the audio and the noise adding the rescaled noise to audio since tf signal fft applies fft on the innermost dimension we need to squeeze the dimensions and then expand them again after fft return the absolute value of the first half of the fft which represents the positive frequencies get the list of audio file paths along with their corresponding labels shuffle split into training and validation create 2 datasets one for training and the other for validation add noise to the training set transform audio wave to the frequency domain using audio_to_fft model definition shortcut compile the model using adam s default learning rate add callbacks earlystopping to stop training when the model is not enhancing anymore modelcheckpoint to always keep the model that has the best val_accuracy training evaluation we get 98 validation accuracy demonstration let s take some samples and predict the speaker compare the prediction with the real speaker listen to the audio to see that despite the samples being noisy the model is still pretty accurate get the signal fft predict take random samples for every sample print the true and predicted label as well as run the voice with the noise
import os

os.environ["KERAS_BACKEND"] = "tensorflow"

import shutil
import numpy as np

import tensorflow as tf
import keras

from pathlib import Path
from IPython.display import display, Audio






DATASET_ROOT = "16000_pcm_speeches"


AUDIO_SUBFOLDER = "audio"
NOISE_SUBFOLDER = "noise"

DATASET_AUDIO_PATH = os.path.join(DATASET_ROOT, AUDIO_SUBFOLDER)
DATASET_NOISE_PATH = os.path.join(DATASET_ROOT, NOISE_SUBFOLDER)


VALID_SPLIT = 0.1


SHUFFLE_SEED = 43






SAMPLING_RATE = 16000




SCALE = 0.5

BATCH_SIZE = 128
EPOCHS = 1






for folder in os.listdir(DATASET_ROOT):
    if os.path.isdir(os.path.join(DATASET_ROOT, folder)):
        if folder in [AUDIO_SUBFOLDER, NOISE_SUBFOLDER]:
            
            continue
        elif folder in ["other", "_background_noise_"]:
            
            
            shutil.move(
                os.path.join(DATASET_ROOT, folder),
                os.path.join(DATASET_NOISE_PATH, folder),
            )
        else:
            
            
            shutil.move(
                os.path.join(DATASET_ROOT, folder),
                os.path.join(DATASET_AUDIO_PATH, folder),
            )




noise_paths = []
for subdir in os.listdir(DATASET_NOISE_PATH):
    subdir_path = Path(DATASET_NOISE_PATH) / subdir
    if os.path.isdir(subdir_path):
        noise_paths += [
            os.path.join(subdir_path, filepath)
            for filepath in os.listdir(subdir_path)
            if filepath.endswith(".wav")
        ]
if not noise_paths:
    raise RuntimeError(f"Could not find any files at {DATASET_NOISE_PATH}")
print(
    "Found {} files belonging to {} directories".format(
        len(noise_paths), len(os.listdir(DATASET_NOISE_PATH))
    )
)



command = (
    "for dir in `ls -1 " + DATASET_NOISE_PATH + "`; do "
    "for file in `ls -1 " + DATASET_NOISE_PATH + "/$dir/*.wav`; do "
    "sample_rate=`ffprobe -hide_banner -loglevel panic -show_streams "
    "$file | grep sample_rate | cut -f2 -d=`; "
    "if [ $sample_rate -ne 16000 ]; then "
    "ffmpeg -hide_banner -loglevel panic -y "
    "-i $file -ar 16000 temp.wav; "
    "mv temp.wav $file; "
    "fi; done; done"
)
os.system(command)



def load_noise_sample(path):
    sample, sampling_rate = tf.audio.decode_wav(
        tf.io.read_file(path), desired_channels=1
    )
    if sampling_rate == SAMPLING_RATE:
        
        slices = int(sample.shape[0] / SAMPLING_RATE)
        sample = tf.split(sample[: slices * SAMPLING_RATE], slices)
        return sample
    else:
        print("Sampling rate for {} is incorrect. Ignoring it".format(path))
        return None


noises = []
for path in noise_paths:
    sample = load_noise_sample(path)
    if sample:
        noises.extend(sample)
noises = tf.stack(noises)

print(
    "{} noise files were split into {} noise samples where each is {} sec. long".format(
        len(noise_paths), noises.shape[0], noises.shape[1] // SAMPLING_RATE
    )
)




def paths_and_labels_to_dataset(audio_paths, labels):
    
    path_ds = tf.data.Dataset.from_tensor_slices(audio_paths)
    audio_ds = path_ds.map(
        lambda x: path_to_audio(x), num_parallel_calls=tf.data.AUTOTUNE
    )
    label_ds = tf.data.Dataset.from_tensor_slices(labels)
    return tf.data.Dataset.zip((audio_ds, label_ds))


def path_to_audio(path):
    
    audio = tf.io.read_file(path)
    audio, _ = tf.audio.decode_wav(audio, 1, SAMPLING_RATE)
    return audio


def add_noise(audio, noises=None, scale=0.5):
    if noises is not None:
        
        
        tf_rnd = tf.random.uniform(
            (tf.shape(audio)[0],), 0, noises.shape[0], dtype=tf.int32
        )
        noise = tf.gather(noises, tf_rnd, axis=0)

        
        prop = tf.math.reduce_max(audio, axis=1) / tf.math.reduce_max(
            noise, axis=1
        )
        prop = tf.repeat(
            tf.expand_dims(prop, axis=1), tf.shape(audio)[1], axis=1
        )

        
        audio = audio + noise * prop * scale

    return audio


def audio_to_fft(audio):
    
    
    
    audio = tf.squeeze(audio, axis=-1)
    fft = tf.signal.fft(
        tf.cast(tf.complex(real=audio, imag=tf.zeros_like(audio)), tf.complex64)
    )
    fft = tf.expand_dims(fft, axis=-1)

    
    
    return tf.math.abs(fft[:, : (audio.shape[1] // 2), :])




class_names = os.listdir(DATASET_AUDIO_PATH)
print(
    "Our class names: {}".format(
        class_names,
    )
)

audio_paths = []
labels = []
for label, name in enumerate(class_names):
    print(
        "Processing speaker {}".format(
            name,
        )
    )
    dir_path = Path(DATASET_AUDIO_PATH) / name
    speaker_sample_paths = [
        os.path.join(dir_path, filepath)
        for filepath in os.listdir(dir_path)
        if filepath.endswith(".wav")
    ]
    audio_paths += speaker_sample_paths
    labels += [label] * len(speaker_sample_paths)

print(
    "Found {} files belonging to {} classes.".format(
        len(audio_paths), len(class_names)
    )
)


rng = np.random.RandomState(SHUFFLE_SEED)
rng.shuffle(audio_paths)
rng = np.random.RandomState(SHUFFLE_SEED)
rng.shuffle(labels)


num_val_samples = int(VALID_SPLIT * len(audio_paths))
print("Using {} files for training.".format(len(audio_paths) - num_val_samples))
train_audio_paths = audio_paths[:-num_val_samples]
train_labels = labels[:-num_val_samples]

print("Using {} files for validation.".format(num_val_samples))
valid_audio_paths = audio_paths[-num_val_samples:]
valid_labels = labels[-num_val_samples:]


train_ds = paths_and_labels_to_dataset(train_audio_paths, train_labels)
train_ds = train_ds.shuffle(
    buffer_size=BATCH_SIZE * 8, seed=SHUFFLE_SEED
).batch(BATCH_SIZE)

valid_ds = paths_and_labels_to_dataset(valid_audio_paths, valid_labels)
valid_ds = valid_ds.shuffle(buffer_size=32 * 8, seed=SHUFFLE_SEED).batch(32)



train_ds = train_ds.map(
    lambda x, y: (add_noise(x, noises, scale=SCALE), y),
    num_parallel_calls=tf.data.AUTOTUNE,
)


train_ds = train_ds.map(
    lambda x, y: (audio_to_fft(x), y), num_parallel_calls=tf.data.AUTOTUNE
)
train_ds = train_ds.prefetch(tf.data.AUTOTUNE)

valid_ds = valid_ds.map(
    lambda x, y: (audio_to_fft(x), y), num_parallel_calls=tf.data.AUTOTUNE
)
valid_ds = valid_ds.prefetch(tf.data.AUTOTUNE)




def residual_block(x, filters, conv_num=3, activation="relu"):
    
    s = keras.layers.Conv1D(filters, 1, padding="same")(x)
    for i in range(conv_num - 1):
        x = keras.layers.Conv1D(filters, 3, padding="same")(x)
        x = keras.layers.Activation(activation)(x)
    x = keras.layers.Conv1D(filters, 3, padding="same")(x)
    x = keras.layers.Add()([x, s])
    x = keras.layers.Activation(activation)(x)
    return keras.layers.MaxPool1D(pool_size=2, strides=2)(x)


def build_model(input_shape, num_classes):
    inputs = keras.layers.Input(shape=input_shape, name="input")

    x = residual_block(inputs, 16, 2)
    x = residual_block(x, 32, 2)
    x = residual_block(x, 64, 3)
    x = residual_block(x, 128, 3)
    x = residual_block(x, 128, 3)

    x = keras.layers.AveragePooling1D(pool_size=3, strides=3)(x)
    x = keras.layers.Flatten()(x)
    x = keras.layers.Dense(256, activation="relu")(x)
    x = keras.layers.Dense(128, activation="relu")(x)

    outputs = keras.layers.Dense(
        num_classes, activation="softmax", name="output"
    )(x)

    return keras.models.Model(inputs=inputs, outputs=outputs)


model = build_model((SAMPLING_RATE // 2, 1), len(class_names))

model.summary()


model.compile(
    optimizer="Adam",
    loss="sparse_categorical_crossentropy",
    metrics=["accuracy"],
)




model_save_filename = "model.keras"

earlystopping_cb = keras.callbacks.EarlyStopping(
    patience=10, restore_best_weights=True
)
mdlcheckpoint_cb = keras.callbacks.ModelCheckpoint(
    model_save_filename, monitor="val_accuracy", save_best_only=True
)



history = model.fit(
    train_ds,
    epochs=EPOCHS,
    validation_data=valid_ds,
    callbacks=[earlystopping_cb, mdlcheckpoint_cb],
)



print(model.evaluate(valid_ds))





SAMPLES_TO_DISPLAY = 10

test_ds = paths_and_labels_to_dataset(valid_audio_paths, valid_labels)
test_ds = test_ds.shuffle(buffer_size=BATCH_SIZE * 8, seed=SHUFFLE_SEED).batch(
    BATCH_SIZE
)

test_ds = test_ds.map(
    lambda x, y: (add_noise(x, noises, scale=SCALE), y),
    num_parallel_calls=tf.data.AUTOTUNE,
)

for audios, labels in test_ds.take(1):
    
    ffts = audio_to_fft(audios)
    
    y_pred = model.predict(ffts)
    
    rnd = np.random.randint(0, BATCH_SIZE, SAMPLES_TO_DISPLAY)
    audios = audios.numpy()[rnd, :, :]
    labels = labels.numpy()[rnd]
    y_pred = np.argmax(y_pred, axis=-1)[rnd]

    for index in range(SAMPLES_TO_DISPLAY):
        
        
        print(
            "Speaker:\33{} {}\33[0m\tPredicted:\33{} {}\33[0m".format(
                "[92m" if labels[index] == y_pred[index] else "[91m",
                class_names[labels[index]],
                "[92m" if labels[index] == y_pred[index] else "[91m",
                class_names[y_pred[index]],
            )
        )
        display(Audio(audios[index, :, :].squeeze(), rate=SAMPLING_RATE))