# -*- coding: utf-8 -*-
"""
Final Corrected SER Model v2 (Fix Gradio Feature Extraction).ipynb
"""
# @title 1. Setup: Install Libraries
# Install necessary libraries including audiomentations and ensure XGBoost is up-to-date
#!pip install librosa soundfile scikit-learn matplotlib seaborn pandas tqdm gradio xgboost audiomentations --upgrade -q
# @title 2. Setup: Imports and GPU Check
import os
import librosa
import numpy as np
import pandas as pd
from tqdm import tqdm
import [Link] as plt
import seaborn as sns
import tarfile
import warnings
import pickle
import gradio as gr
import time
import torch
# Augmentation
from audiomentations import Compose, AddGaussianNoise, TimeStretch, PitchShift
# ML Components
from sklearn.model_selection import train_test_split, GridSearchCV, StratifiedKFold
from xgboost import XGBClassifier
from [Link] import StandardScaler, LabelEncoder
from [Link] import accuracy_score, classification_report, confusion_matrix, precision_score, recall_score, f1_score
# Ignore warnings
[Link]('ignore', category=UserWarning)
[Link]('ignore', category=FutureWarning)
# --- GPU Check ---
if [Link].is_available():
print(f"GPU detected: {[Link].get_device_name(0)}")
device = 'cuda'
else:
print("No GPU detected, XGBoost will run on CPU (slower).")
device = 'cpu'
# @title 3. Setup: Constants and Paths
DRIVE_TAR_GZ_PATH = "/content/drive/MyDrive/hi_kia_1.[Link]"
LOCAL_TAR_GZ_PATH = "hi_kia_1.[Link]"
N_MFCC = 13
AUGMENT_RATIO = 0.6
EXTRACT_DIR_COLAB = "/content/hi_kia_dataset"
EXTRACT_DIR_LOCAL = "hi_kia_dataset"
SCALER_PATH = "feature_scaler.pkl"
LABEL_ENCODER_PATH = "label_encoder.pkl"
BEST_MODEL_PATH = "best_tuned_ser_model.pkl"
FEATURE_LENGTH_PATH = "expected_feature_length.pkl" # Path to save the feature length
# Initial state for feature length
EXPECTED_FEATURE_LENGTH = -1
# @title 4. Setup: Mount Drive & Determine Paths
colab_env = False
try:
from [Link] import drive
[Link]('/content/drive', force_remount=True)
TAR_GZ_PATH = DRIVE_TAR_GZ_PATH
EXTRACT_DIR = EXTRACT_DIR_COLAB
print("Google Drive mounted.")
colab_env = True
except ImportError:
print("Not running in Colab, using local paths.")
TAR_GZ_PATH = LOCAL_TAR_GZ_PATH
EXTRACT_DIR = EXTRACT_DIR_LOCAL
colab_env = False
ANNOTATION_CSV_PATH = [Link](EXTRACT_DIR, "hi_kia", "annotation", "[Link]")
AUDIO_BASE_PATH = [Link](EXTRACT_DIR, "hi_kia", "wav")
# @title 5. Setup: Extract Dataset (if necessary)
[Link](EXTRACT_DIR, exist_ok=True)
print(f"Checking for dataset in {EXTRACT_DIR}...")
if not [Link](ANNOTATION_CSV_PATH) or not [Link](AUDIO_BASE_PATH):
print(f"Dataset files not found. Attempting extraction from {TAR_GZ_PATH}...")
if [Link](TAR_GZ_PATH):
try:
with [Link](TAR_GZ_PATH, "r:gz") as tar:
[Link](path=EXTRACT_DIR)
print(f"Extracted files to: {EXTRACT_DIR}")
if not [Link](ANNOTATION_CSV_PATH) or not [Link](AUDIO_BASE_PATH):
raise FileNotFoundError("Extraction failed or core dataset files missing.")
except Exception as e:
print(f"ERROR during extraction: {e}")
raise
else:
raise FileNotFoundError(f"ERROR: Tar GZ file not found at {TAR_GZ_PATH}.")
else:
print("Dataset already appears to be extracted.")
# @title 6. Load Annotations
try:
df_annotations = pd.read_csv(ANNOTATION_CSV_PATH)
print("\nAnnotation data loaded successfully.")
print(f"Total samples in annotation file: {len(df_annotations)}")
print("Emotion distribution:")
print(df_annotations['emo'].value_counts())
if 'Unnamed: 0' not in df_annotations.columns:
raise KeyError("'Unnamed: 0' column missing.")
except Exception as e:
print(f"ERROR loading annotations: {e}")
raise
# @title 7. Define Augmentation Pipeline & Enhanced Feature Extraction Function
# --- Augmentation Pipeline ---
augmenter = Compose([
AddGaussianNoise(min_amplitude=0.001, max_amplitude=0.010, p=0.3),
TimeStretch(min_rate=0.85, max_rate=1.15, p=0.3),
PitchShift(min_semitones=-3, max_semitones=3, p=0.3),
])
# --- Feature Extraction Helper (takes audio data AND expected length) ---
def extract_enhanced_features_from_data(y, sr, expected_length=-1):
""" Extracts features from pre-loaded audio data. Checks against expected_length. """
try:
if [Link]([Link](y)) < 1e-5: return None # Skip silent audio
mfccs = [Link](y=y, sr=sr, n_mfcc=N_MFCC)
mfccs_delta = [Link](mfccs)
mfccs_delta2 = [Link](mfccs, order=2)
chroma = [Link].chroma_stft(y=y, sr=sr)
mel = [Link](y=y, sr=sr)
contrast = [Link].spectral_contrast(y=y, sr=sr)
zcr = [Link].zero_crossing_rate(y)
rms = [Link](y=y)
def get_stats(feature_matrix):
shape = feature_matrix.shape
if feature_matrix is None or feature_matrix.size == 0:
return [Link](shape[0] * 2 if len(shape) > 1 else 2)
means = [Link](feature_matrix.T, axis=0)
stds = [Link](feature_matrix.T, axis=0)
stds = np.nan_to_num(stds)
return [Link]((means, stds))
mfccs_stats, mfccs_delta_stats, mfccs_delta2_stats = get_stats(mfccs), get_stats(mfccs_delta), get_stats(mfccs_delta2)
chroma_stats, mel_stats, contrast_stats = get_stats(chroma), get_stats(mel), get_stats(contrast)
zcr_stats, rms_stats = get_stats(zcr), get_stats(rms)
all_features = [Link]((
mfccs_stats, mfccs_delta_stats, mfccs_delta2_stats,
chroma_stats, mel_stats, contrast_stats, zcr_stats, rms_stats
))
current_length = all_features.shape[0]
# Check against expected length if provided and valid
if expected_length > 0 and current_length != expected_length:
print(f"Warning: Feature length mismatch. Expected {expected_length}, got {current_length}. Skipping sample.")
return None
# Final check for NaN/inf
if not [Link]([Link](all_features)):
# print(f"Warning: NaN/inf found in features after combining. Replacing with zeros.") # Verbose
all_features = np.nan_to_num(all_features, nan=0.0, posinf=0.0, neginf=0.0)
# Check shape again after potential fix
if expected_length > 0 and all_features.shape[0] != expected_length:
print("Warning: Feature length mismatch after NaN handling. Skipping sample.")
return None
return all_features
except Exception as e:
# print(f"Error during feature extraction from data: {e}") # Verbose
return None
# @title 8. Prepare Data: Extract Features with Augmentation
print("\nExtracting ENHANCED features (with augmentation)...")
features_list = []
labels_list = []
indices_to_keep = []
available_wav_files = set([Link](AUDIO_BASE_PATH))
# --- Preliminary split for augmentation target ID ---
temp_indices = df_annotations.index
temp_labels = df_annotations['emo'].values
train_indices_set = set()
if len(temp_indices) > 1 and len([Link](temp_labels)) > 1:
train_indices, _, _, _ = train_test_split(temp_indices, temp_labels, test_size=0.2, random_state=42, stratify=temp_labels)
train_indices_set = set(train_indices)
print(f"Identified {len(train_indices_set)} indices for potential augmentation.")
# Use a local variable for expected length during this phase
_local_expected_feature_length = -1
for index, row in tqdm(df_annotations.iterrows(), total=len(df_annotations), desc="Processing Audio"):
base_filename = row['Unnamed: 0']
file_name = f"{base_filename}.wav"
file_path = [Link](AUDIO_BASE_PATH, file_name)
emotion_label = row['emo'].lower()
is_training_sample = index in train_indices_set
should_augment = is_training_sample and ([Link]() < AUGMENT_RATIO)
if file_name in available_wav_files:
try:
y_audio, sr = [Link](file_path, sr=None, res_type='kaiser_fast')
current_features = None
if should_augment:
y_augmented = augmenter(samples=y_audio, sample_rate=sr)
# Pass the current local expected length to the helper
current_features = extract_enhanced_features_from_data(y_augmented, sr, _local_expected_feature_length)
else:
current_features = extract_enhanced_features_from_data(y_audio, sr, _local_expected_feature_length)
# Set the expected length on the first success
if _local_expected_feature_length <= 0 and current_features is not None:
_local_expected_feature_length = current_features.shape[0]
print(f"Determined feature vector length: {_local_expected_feature_length}")
# Append if successful AND matches the determined length
if current_features is not None and current_features.shape[0] == _local_expected_feature_length:
features_list.append(current_features)
labels_list.append(emotion_label)
indices_to_keep.append(index)
elif current_features is not None:
# Mismatch detected by helper or this check
print(f"Skipping {file_name} due to feature length mismatch.")
except Exception as e:
# print(f"Skipping {file_name} due to error during loading/augmentation: {e}") # Verbose
continue
# --- Final Array Creation & Save Feature Length ---
X = [Link](features_list)
y = [Link](labels_list)
if len(X) > 0:
df_processed = df_annotations.loc[indices_to_keep].reset_index(drop=True)
# Save the definitively determined feature length
EXPECTED_FEATURE_LENGTH = [Link][1] # Use the actual shape after stacking
print(f"\nSaving expected feature length ({EXPECTED_FEATURE_LENGTH}) to {FEATURE_LENGTH_PATH}")
with open(FEATURE_LENGTH_PATH, 'wb') as f:
[Link](EXPECTED_FEATURE_LENGTH, f)
else:
df_processed = [Link](columns=df_annotations.columns)
EXPECTED_FEATURE_LENGTH = -1 # Indicate failure
print(f"\nSuccessfully extracted features for {len(X)} samples.")
if len(X) > 0:
print(f"Feature matrix shape (X): {[Link]}")
print(f"Label array shape (y): {[Link]}")
if len(X) != len(y) or len(X) != len(df_processed):
raise ValueError("Mismatch in final data lengths")
if [Link][1] != EXPECTED_FEATURE_LENGTH:
print(f"WARNING: Final X width ({[Link][1]}) != saved expected ({EXPECTED_FEATURE_LENGTH}).")
else:
print("WARNING: No features were successfully extracted.")
# @title 9. Encode Labels and Scale Features
if len(X) > 0:
print("\nEncoding labels...")
label_encoder = LabelEncoder()
y_encoded = label_encoder.fit_transform(y)
print(f"Label mapping: {dict(zip(label_encoder.classes_, label_encoder.transform(label_encoder.classes_)))}")
print("\nScaling features...")
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
print("Features scaled.")
print(f"Saving Scaler to {SCALER_PATH}")
with open(SCALER_PATH, 'wb') as f: [Link](scaler, f)
print(f"Saving Label Encoder to {LABEL_ENCODER_PATH}")
with open(LABEL_ENCODER_PATH, 'wb') as f: [Link](label_encoder, f)
else:
print("\nSkipping label encoding, scaling, and training.")
# @title 10. Hyperparameter Tuning (XGBoost GPU) & Model Training/Evaluation
best_model_name = "N/A"
accuracy = 0.0
BEST_MODEL_PATH = None
if len(X) > 1 and len([Link](y_encoded)) > 1:
start_time_grid = [Link]()
print("\nSplitting scaled data into training and testing sets (80/20)...")
X_train, X_test, y_train, y_test = train_test_split(
X_scaled, y_encoded, test_size=0.2, random_state=42, stratify=y_encoded
)
print(f"Training data shape: {X_train.shape}, Test data shape: {X_test.shape}")
print("\n--- Tuning XGBoost Hyperparameters using GridSearchCV (GPU if available) ---")
param_grid = {
'n_estimators': [150, 300], 'max_depth': [5, 8],
'learning_rate': [0.1, 0.15], 'subsample': [0.8, 1.0],
'colsample_bytree': [0.8, 1.0]
}
cv = StratifiedKFold(n_splits=3, shuffle=True, random_state=42)
xgb_params = {
'objective': 'multi:softmax', 'num_class': len(label_encoder.classes_),
'eval_metric': 'mlogloss', 'random_state': 42, 'use_label_encoder': False
}
if device == 'cuda':
xgb_params['tree_method'] = 'gpu_hist'
print("XGBoost will use GPU.")
else:
xgb_params['tree_method'] = 'hist'
print("XGBoost will use CPU.")
xgb = XGBClassifier(**xgb_params)
grid_search = GridSearchCV(estimator=xgb, param_grid=param_grid, scoring='accuracy',
cv=cv, n_jobs=1, verbose=2)
print(f"Starting GridSearchCV (n_jobs=1, tree_method='{xgb_params['tree_method']}')...")
grid_search.fit(X_train, y_train)
end_time_grid = [Link]()
print(f"GridSearchCV finished in {end_time_grid - start_time_grid:.2f} seconds.")
print(f"\nBest Parameters found: {grid_search.best_params_}")
print(f"Best Cross-validation Accuracy: {grid_search.best_score_:.4f}")
best_model = grid_search.best_estimator_
best_model_name = f"XGBoost (Tuned, {[Link]()})"
print(f"\n--- Evaluating Best Tuned Model ({best_model_name}) on Test Set ---")
y_pred = best_model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred, average='weighted')
recall = recall_score(y_test, y_pred, average='weighted')
f1 = f1_score(y_test, y_pred, average='weighted')
class_report = classification_report(y_test, y_pred, target_names=label_encoder.classes_)
conf_matrix = confusion_matrix(y_test, y_pred)
class_labels = label_encoder.classes_
print(f"\n{best_model_name} Test Set Evaluation Metrics:")
print(f"Accuracy: {accuracy:.4f}")
print(f"Precision: {precision:.4f}")
print(f"Recall: {recall:.4f}")
print(f"F1-Score: {f1:.4f}")
print("\nClassification Report:\n", class_report)
[Link](figsize=(8, 6))
[Link](conf_matrix, annot=True, fmt="d", cmap="Blues",
xticklabels=class_labels, yticklabels=class_labels)
[Link]("Predicted Label")
[Link]("True Label")
[Link](f"Confusion Matrix - {best_model_name}")
[Link]()
BEST_MODEL_PATH = "best_tuned_ser_model.pkl"
print(f"Saving Best Tuned Model to {BEST_MODEL_PATH}")
with open(BEST_MODEL_PATH, 'wb') as f: [Link](best_model, f)
else:
print("\nSkipping model training and evaluation due to insufficient data.")
# @title 11. Gradio UI: Load Components and Define Interface
try:
# Check if necessary files/variables exist before loading
if not all([[Link](SCALER_PATH),
[Link](LABEL_ENCODER_PATH),
[Link](FEATURE_LENGTH_PATH), # Check for feature length file
BEST_MODEL_PATH and [Link](BEST_MODEL_PATH)]):
raise FileNotFoundError("Model/Scaler/Encoder/FeatureLength file not found. Ensure training completed.")
with open(SCALER_PATH, 'rb') as f: loaded_scaler = [Link](f)
with open(LABEL_ENCODER_PATH, 'rb') as f: loaded_label_encoder = [Link](f)
with open(BEST_MODEL_PATH, 'rb') as f: loaded_model = [Link](f)
with open(FEATURE_LENGTH_PATH, 'rb') as f: loaded_expected_feature_length = [Link](f) # Load the length
print("\nScaler, Label Encoder, Feature Length, and Best Tuned Model loaded successfully.")
# Define Gradio Prediction Function
def predict_emotion_gradio(audio_filepath): # Input is filepath from [Link]
if audio_filepath is None: return "No audio input provided.", {}
source_info = [Link](audio_filepath)
print(f"Predicting for: {source_info}")
try:
# 1. Load audio using the filepath
y_audio, sr = [Link](audio_filepath, sr=None, res_type='kaiser_fast')
# 2. Extract features using the loaded expected length
features = extract_enhanced_features_from_data(y_audio, sr, loaded_expected_feature_length) # Pass loaded length
if features is None:
print(f"ERROR: Feature extraction returned None for {source_info}")
return "Error extracting features (check logs for details).", {}
# Feature length check is now inside the helper using the passed argument
# 3. Scale features
features_scaled = loaded_scaler.transform([Link](1, -1))
# 4. Predict
prediction_encoded = loaded_model.predict(features_scaled)
predicted_emotion = loaded_label_encoder.inverse_transform(prediction_encoded)[0].capitalize()
# 5. Probabilities
probabilities = {}
if hasattr(loaded_model, "predict_proba"):
probs = loaded_model.predict_proba(features_scaled)[0]
probabilities = {loaded_label_encoder.classes_[i]: float(prob) for i, prob in enumerate(probs)}
probabilities = dict(sorted([Link](), key=lambda item: item[1], reverse=True))
print(f" Predicted Emotion: {predicted_emotion}")
return predicted_emotion, probabilities
except FileNotFoundError:
print(f"ERROR: Gradio temp file not found? Path: {audio_filepath}")
return f"Error loading audio file {source_info}.", {}
except Exception as e:
print(f"Error during prediction process for {source_info}: {e}")
import traceback
traceback.print_exc() # Print full traceback for debugging
return f"Prediction Error: {e}", {}
# --- Create Gradio Interface ---
print("\nSetting up Gradio Interface...")
iface = [Link](
fn=predict_emotion_gradio,
# Use type="filepath" for simpler handling in the prediction function
inputs=[Link](sources=["microphone", "upload"], type="filepath", label="Input Audio (Record or Upload)"),
outputs=[
[Link](label="Predicted Emotion"),
[Link](label="Confidence Scores", num_top_classes=len(loaded_label_encoder.classes_))
],
title=f"Speech Emotion Recognition ({best_model_name})",
description=f"Record or upload audio. Model: {best_model_name} | Features: Enhanced | Augmentation: Yes | Tuning: Yes | Test Accuracy: {accuracy:.4f}",
allow_flagging="never"
)
print("\nLaunching Gradio Interface...")
[Link](share=colab_env, debug=True) # Set debug=True temporarily for detailed logs
except FileNotFoundError as e:
print(f"\nERROR loading files for Gradio: {e}. Ensure previous cells ran successfully and saved files.")
except NameError as e:
print(f"\nERROR: Required variables not defined. Run training cells first. Details: {e}")
except Exception as e:
print(f"\nAn unexpected error occurred setting up Gradio: {e}")
raise