iti110_birds_classifier_model_builder.py

# -*- coding: utf-8 -*-
"""ITI110-Birds-CNN-ResNet-GPT-4o.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/16uUvA-YukBceEUQd8qBCISg9K3raRmAh

Based on the above pipeline, write a python code to create a Gradio UI which performs the following: 1. An input text box for user to input the path to the dataset 2. An output text bx to display the distribution of samples across species 3. Automatically perform 1.2 Audio Processing 4. Put 3 checkboxes for enabling or disabling Pitch Shifting, Time Stretching, and Background Noise 5. Use bullet buttons to allow user to select using MFCC, Spectrograms or Chroma for Feature Extraction 6. Use bullet buttons to allow user to select building either RestNet50 or CNN model 7. Let the user press a button to train and save the model 8. Let the user press a button to evaluate the model, output the results 9. Let the user press a button to perform hyperparameter tuning and save the model file 10. Let the user press a button to perform Learning Rate Reduction Callback and save the model file 11. Let the user press a button to evaluate the two fine-tuned models, output the results

Below is the Python code to create a Gradio UI that performs the tasks as specified. This code assumes you have already installed Gradio (pip install gradio) and other necessary libraries.
"""


import os
import numpy as np
import pandas as pd
import librosa
import soundfile as sf
from pydub import AudioSegment
import gradio as gr
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from tensorflow.keras.utils import to_categorical
from tensorflow.keras.models import Sequential, load_model, Model
from tensorflow.keras.layers import Conv2D, MaxPooling2D, Flatten, Dense, Dropout, Input
from tensorflow.keras.applications import ResNet50
from tensorflow.keras.callbacks import ReduceLROnPlateau
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import classification_report, confusion_matrix

# Global variables
dataset_path = ''
species_folders = []
label_encoder = LabelEncoder()
X_train, X_test, y_train, y_test = None, None, None, None
history = None
model = None

def data_exploration(path):
    global dataset_path, species_folders
    dataset_path = path
    species_folders = [f for f in os.listdir(dataset_path) if os.path.isdir(os.path.join(dataset_path, f))]
    species_count = Counter()
    for species in species_folders:
        species_files = os.listdir(os.path.join(dataset_path, species))
        species_count[species] = len(species_files)
    species_df = pd.DataFrame(list(species_count.items()), columns=['Species', 'Count'])
    species_df.to_csv('species_distribution.csv', index=False)
    return species_df.to_string(index=False)

def audio_processing():
    for species in species_folders:
        species_folder = os.path.join(dataset_path, species)
        for file_name in os.listdir(species_folder):
            if file_name.endswith('.mp3'):
                mp3_path = os.path.join(species_folder, file_name)
                wav_path = os.path.join(species_folder, file_name.replace('.mp3', '.wav'))
                convert_mp3_to_wav(mp3_path, wav_path)
                os.remove(mp3_path)
    for species in species_folders:
        species_folder = os.path.join(dataset_path, species)
        for file_name in os.listdir(species_folder):
            if file_name.endswith('.wav'):
                wav_path = os.path.join(species_folder, file_name)
                preprocess_audio(wav_path, wav_path)
    return "Audio processing completed."

def convert_mp3_to_wav(input_path, output_path):
    audio = AudioSegment.from_mp3(input_path)
    audio.export(output_path, format="wav")

def preprocess_audio(input_path, output_path, sample_rate=22050):
    y, sr = librosa.load(input_path, sr=sample_rate)
    y_trimmed, _ = librosa.effects.trim(y)
    y_normalized = librosa.util.normalize(y_trimmed)
    sf.write(output_path, y_normalized, sample_rate)

def data_augmentation(pitch_shifting, time_stretching, background_noise):
    for species in species_folders:
        species_folder = os.path.join(dataset_path, species)
        if pitch_shifting:
            apply_pitch_shift(species_folder)
        if time_stretching:
            apply_time_stretch(species_folder)
        if background_noise:
            apply_background_noise(species_folder)
    return "Data augmentation completed."

def apply_pitch_shift(species_folder):
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            y_shifted = librosa.effects.pitch_shift(y, sr, n_steps=2)
            shifted_path = wav_path.replace('.wav', '_pitch_shifted.wav')
            sf.write(shifted_path, y_shifted, sr)

def apply_time_stretch(species_folder):
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            y_stretched = librosa.effects.time_stretch(y, rate=1.2)
            stretched_path = wav_path.replace('.wav', '_time_stretched.wav')
            sf.write(stretched_path, y_stretched, sr)

def apply_background_noise(species_folder):
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            noise = np.random.randn(len(y))
            y_noisy = y + 0.005 * noise
            noisy_path = wav_path.replace('.wav', '_noisy.wav')
            sf.write(noisy_path, y_noisy, sr)

def feature_extraction(feature_type):
    output_folder = f'path_to_output_{feature_type.lower()}'
    if feature_type == 'MFCC':
        extract_features = extract_mfcc
    elif feature_type == 'Spectrogram':
        extract_features = extract_spectrogram
    elif feature_type == 'Chroma':
        extract_features = extract_chroma_features

    for species in species_folders:
        species_folder = os.path.join(dataset_path, species)
        species_output_folder = os.path.join(output_folder, species)
        extract_features(species_folder, species_output_folder)

    return f"{feature_type} features extracted and saved."

def extract_mfcc(species_folder, output_folder):
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            mfccs = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
            feature_path = os.path.join(output_folder, file_name.replace('.wav', '.npy'))
            np.save(feature_path, mfccs)

def extract_spectrogram(species_folder, output_folder):
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            S = librosa.feature.melspectrogram(y=y, sr=sr)
            S_DB = librosa.power_to_db(S, ref=np.max)
            feature_path = os.path.join(output_folder, file_name.replace('.wav', '.npy'))
            np.save(feature_path, S_DB)

def extract_chroma_features(species_folder, output_folder):
    if not os.path.exists(output_folder):
        os.makedirs(output_folder)
    for file_name in os.listdir(species_folder):
        if file_name.endswith('.wav'):
            wav_path = os.path.join(species_folder, file_name)
            y, sr = librosa.load(wav_path, sr=None)
            chroma = librosa.feature.chroma_stft(y=y, sr=sr)
            feature_path = os.path.join(output_folder, file_name.replace('.wav', '_chroma.npy'))
            np.save(feature_path, chroma)

def load_features_and_labels(feature_folder):
    X = []
    y = []
    species_list = os.listdir(feature_folder)
    for species in species_list:
        species_folder = os.path.join(feature_folder, species)
        for file_name in os.listdir(species_folder):
            if file_name.endswith('.npy'):
                feature_path = os.path.join(species_folder, file_name)
                features = np.load(feature_path)
                X.append(features)
                y.append(species)
    return np.array(X), np.array(y)

def build_model(model_type, feature_type):
    global X_train, X_test, y_train, y_test, label_encoder, model
    feature_folder = f'path_to_output_{feature_type.lower()}'
    X, y = load_features_and_labels(feature_folder)
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    label_encoder = LabelEncoder()
    y_train_encoded = label_encoder.fit_transform(y_train)
    y_test_encoded = label_encoder.transform(y_test)
    y_train_categorical = to_categorical(y_train_encoded)
    y_test_categorical = to_categorical(y_test_encoded)
    np.save('classes.npy', label_encoder.classes_)
    X_train_reshaped = np.stack([np.repeat(x[..., np.newaxis], 3, axis=-1) for x in X_train])
    X_test_reshaped = np.stack([np.repeat(x[..., np.newaxis], 3, axis=-1) for x in X_test])

    if model_type == 'ResNet50':
        input_tensor = Input(shape=(X_train.shape[1], X_train.shape[2], 3))
        base_model = ResNet50(include_top=False, weights='imagenet', input_tensor=input_tensor)
        x = Flatten()(base_model.output)
        x = Dense(128, activation='relu')(x)
        x = Dropout(0.3)(x)
        output_tensor = Dense(len(label_encoder.classes_), activation='softmax')(x)
        model = Model(inputs=input_tensor, outputs=output_tensor)
    else:
        model = Sequential([
            Conv2D(32, (3, 3), activation='relu', input_shape=(X_train.shape[1], X_train.shape[2], 3)),
            MaxPooling2D((2, 2)),
            Dropout(0.3),
            Conv2D(64, (3, 3), activation='relu'),
            MaxPooling2D((2, 2)),
            Dropout(0.3),
            Flatten(),
            Dense(128, activation='relu'),
            Dropout(0.3),
            Dense(len(label_encoder.classes_), activation='softmax')
        ])

    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
    model.summary()
    return "Model built."

def train_model():
    global history
    history = model.fit(X_train_reshaped, y_train_categorical, epochs=30, batch_size=32, validation_data=(X_test_reshaped, y_test_categorical))
    model.save('bird_species_model.h5')
    return "Model training completed and saved as 'bird_species_model.h5'."

def evaluate_model():
    test_loss, test_accuracy = model.evaluate(X_test_reshaped, y_test_categorical)
    y_pred = model.predict(X_test_reshaped)
    y_pred_classes = np.argmax(y_pred, axis=1)
    y_true_classes = np.argmax(y_test_categorical, axis=1)
    report = classification_report(y_true_classes, y_pred_classes, target_names=label_encoder.classes_)
    conf_matrix = confusion_matrix(y_true_classes, y_pred_classes)
    plt.figure(figsize=(12, 8))
    sns.heatmap(conf_matrix, annot=True, fmt='d', cmap='Blues', xticklabels=label_encoder.classes_, yticklabels=label_encoder.classes_)
    plt.xlabel('Predicted Label')
    plt.ylabel('True Label')
    plt.title('Confusion Matrix')
    plt.show()
    return f"Test Loss: {test_loss}\nTest Accuracy: {test_accuracy}\n\nClassification Report:\n{report}"

def hyperparameter_tuning():
    global history
    reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=5, min_lr=0.001)
    history = model.fit(X_train_reshaped, y_train_categorical, epochs=50, batch_size=32, validation_data=(X_test_reshaped, y_test_categorical), callbacks=[reduce_lr])
    model.save('bird_species_model_fine_tuned_v1.h5')
    return "Model hyperparameter tuning completed and saved as 'bird_species_model_fine_tuned_v1.h5'."

def learning_rate_reduction():
    global history
    reduce_lr = ReduceLROnPlateau(monitor='val_loss', factor=0.2, patience=5, min_lr=0.001)
    history = model.fit(X_train_reshaped, y_train_categorical, epochs=50, batch_size=32, validation_data=(X_test_reshaped, y_test_categorical), callbacks=[reduce_lr])
    model.save('bird_species_model_fine_tuned_v2.h5')
    return "Model learning rate reduction completed and saved as 'bird_species_model_fine_tuned_v2.h5'."

def evaluate_fine_tuned_models():
    model_v1 = load_model('bird_species_model_fine_tuned_v1.h5')
    model_v2 = load_model('bird_species_model_fine_tuned_v2.h5')

    test_loss_v1, test_accuracy_v1 = model_v1.evaluate(X_test_reshaped, y_test_categorical)
    y_pred_v1 = model_v1.predict(X_test_reshaped)
    y_pred_classes_v1 = np.argmax(y_pred_v1, axis=1)
    y_true_classes_v1 = np.argmax(y_test_categorical, axis=1)
    report_v1 = classification_report(y_true_classes_v1, y_pred_classes_v1, target_names=label_encoder.classes_)

    test_loss_v2, test_accuracy_v2 = model_v2.evaluate(X_test_reshaped, y_test_categorical)
    y_pred_v2 = model_v2.predict(X_test_reshaped)
    y_pred_classes_v2 = np.argmax(y_pred_v2, axis=1)
    y_true_classes_v2 = np.argmax(y_test_categorical, axis=1)
    report_v2 = classification_report(y_true_classes_v2, y_pred_classes_v2, target_names=label_encoder.classes_)

    return f"Fine-Tuned Model V1:\nTest Loss: {test_loss_v1}\nTest Accuracy: {test_accuracy_v1}\n\nClassification Report:\n{report_v1}\n\n" \
           f"Fine-Tuned Model V2:\nTest Loss: {test_loss_v2}\nTest Accuracy: {test_accuracy_v2}\n\nClassification Report:\n{report_v2}"

# Gradio UI
with gr.Blocks() as demo:
    with gr.Row():
        dataset_path_input = gr.Textbox(label="Dataset Path")
        data_exploration_output = gr.Textbox(label="Data Exploration Output")
    with gr.Row():
        audio_processing_button = gr.Button("Perform Audio Processing")
    with gr.Row():
        pitch_shifting_checkbox = gr.Checkbox(label="Enable Pitch Shifting")
        time_stretching_checkbox = gr.Checkbox(label="Enable Time Stretching")
        background_noise_checkbox = gr.Checkbox(label="Enable Background Noise")
        data_augmentation_button = gr.Button("Perform Data Augmentation")
    with gr.Row():
        feature_extraction_radio = gr.Radio(["MFCC", "Spectrogram", "Chroma"], label="Feature Extraction")
        feature_extraction_button = gr.Button("Perform Feature Extraction")
    with gr.Row():
        model_type_radio = gr.Radio(["ResNet50", "CNN"], label="Model Type")
        build_model_button = gr.Button("Build Model")
    with gr.Row():
        train_model_button = gr.Button("Train and Save Model")
        evaluate_model_button = gr.Button("Evaluate Model")
    with gr.Row():
        hyperparameter_tuning_button = gr.Button("Hyperparameter Tuning")
        learning_rate_reduction_button = gr.Button("Learning Rate Reduction")
    with gr.Row():
        evaluate_fine_tuned_models_button = gr.Button("Evaluate Fine-Tuned Models")

    dataset_path_input.change(data_exploration, inputs=dataset_path_input, outputs=data_exploration_output)
    audio_processing_button.click(audio_processing, outputs=None)
    data_augmentation_button.click(data_augmentation, inputs=[pitch_shifting_checkbox, time_stretching_checkbox, background_noise_checkbox], outputs=None)
    feature_extraction_button.click(feature_extraction, inputs=feature_extraction_radio, outputs=None)
    build_model_button.click(build_model, inputs=[model_type_radio, feature_extraction_radio], outputs=None)
    train_model_button.click(train_model, outputs=None)
    evaluate_model_button.click(evaluate_model, outputs=None)
    hyperparameter_tuning_button.click(hyperparameter_tuning, outputs=None)
    learning_rate_reduction_button.click(learning_rate_reduction, outputs=None)
    evaluate_fine_tuned_models_button.click(evaluate_fine_tuned_models, outputs=None)

demo.launch()