import os import glob import pandas as pd import numpy as np import librosa from pathlib import Path import sys # Ensure config is importable sys.path.append(str(Path(__file__).resolve().parent.parent.parent)) from src.config import TEXT_DATASET_PATH, AUDIO_FEATURES_PATH DATASET_ROOT = os.path.join(str(Path(__file__).resolve().parent.parent.parent), "Dataset") from src.data_prep.audio_processor import extract_195_features_from_audio def extract_audio_features(file_path): """ Wrapper around the standardized 195-feature extraction function. """ return extract_195_features_from_audio(file_path) def ingest_audio_data(): print("Ingesting Real Audio Data from Dataset/Audio...") audio_dir = os.path.join(DATASET_ROOT, "Audio") if not os.path.exists(audio_dir): print(f"Audio directory not found: {audio_dir}") return features = [] labels = [] # Process RAVDESS # RAVDESS format: 03-01-01-01-01-01-01.wav # Emotion is 3rd identifier: 01 = neutral, 02 = calm, 03 = happy, 04 = sad, 05 = angry, 06 = fearful, 07 = disgust, 08 = surprised ravdess_files = glob.glob(os.path.join(audio_dir, '**', '03-01-*.wav'), recursive=True) print(f"Found {len(ravdess_files)} RAVDESS files.") for file in ravdess_files: filename = os.path.basename(file) try: emotion_code = int(filename.split("-")[2]) # Map to our 5 categories: 0=Normal, 1=Stress, 2=Anxiety, 3=Depression, 4=Emotional Distress if emotion_code in [1, 2, 3]: label = "Normal" elif emotion_code == 4: label = "Depression" elif emotion_code == 5: label = "Stress" elif emotion_code == 6: label = "Anxiety" elif emotion_code in [7, 8]: label = "Emotional Distress" else: continue feature = extract_audio_features(file) if feature is not None: features.append(feature) labels.append(label) except Exception: continue # Process TESS # TESS format: OAF_angry_...wav tess_files = glob.glob(os.path.join(audio_dir, '**', '*_*.wav'), recursive=True) print(f"Found {len(tess_files)} TESS/CREMA-D potential files.") for file in tess_files: filename = os.path.basename(file).lower() if "angry" in filename or "ang" in filename: label = "Stress" elif "fear" in filename or "fea" in filename: label = "Anxiety" elif "sad" in filename: label = "Depression" elif "disgust" in filename or "ps" in filename: label = "Emotional Distress" elif "neutral" in filename or "neu" in filename or "happy" in filename or "hap" in filename: label = "Normal" else: continue # Avoid double-counting RAVDESS if filename.startswith("03-01-"): continue feature = extract_audio_features(file) if feature is not None: features.append(feature) labels.append(label) if len(features) == 0: print("No valid audio files found. Skipping.") return print(f"Successfully extracted features from {len(features)} audio files.") # Save to CSV expected by audio_classifier.py data = [] for f, l in zip(features, labels): row = {f"feature_{i+1}": f[i] for i in range(len(f))} row["emotion"] = l data.append(row) df = pd.DataFrame(data) os.makedirs(os.path.dirname(AUDIO_FEATURES_PATH), exist_ok=True) df.to_csv(AUDIO_FEATURES_PATH, index=False) print(f"Saved audio features CSV to {AUDIO_FEATURES_PATH}") def ingest_text_data(): print("Ingesting Real Text Data from Dataset/Text...") text_dir = os.path.join(DATASET_ROOT, "Text") if not os.path.exists(text_dir): print(f"Text directory not found: {text_dir}") return csv_files = glob.glob(os.path.join(text_dir, '**', '*.csv'), recursive=True) if not csv_files: print("No CSV files found in Dataset/Text/") return print(f"Found CSV files: {csv_files}") combined_texts = [] combined_labels = [] for file in csv_files: try: df = pd.read_csv(file) print(f"Processing {os.path.basename(file)} with columns: {df.columns.tolist()}") # Handle Dreaddit if 'text' in df.columns and 'label' in df.columns and 'subreddit' in df.columns: print("Detected Dreaddit format.") # Label 1 = Stress, 0 = Non-Stress (Normal) for _, row in df.iterrows(): combined_texts.append(row['text']) combined_labels.append("Stress" if row['label'] == 1 else "Normal") # Handle Mental Health Text Classification elif 'text' in df.columns and 'label' in df.columns: print("Detected Mental Health Classification format.") for _, row in df.iterrows(): # Map labels to our string format if they are numeric, or keep if string label_val = str(row['label']).strip() if label_val in ["0", "Normal"]: combined_labels.append("Normal") elif label_val in ["1", "Depression"]: combined_labels.append("Depression") elif label_val in ["2", "Suicidal"]: combined_labels.append("Suicidal") elif label_val in ["3", "Anxiety"]: combined_labels.append("Anxiety") elif label_val in ["4", "Stress"]: combined_labels.append("Stress") else: combined_labels.append(label_val) # Fallback except Exception as e: print(f"Failed to process {file}: {e}") if not combined_texts: print("Could not extract any text data. Check CSV column names.") return final_df = pd.DataFrame({ "text": combined_texts, "category": combined_labels }) # Calculate metadata features required by the linguistic classifier print("Calculating linguistic metadata...") first_person_words = {"i", "me", "my", "mine", "myself", "we", "our", "us"} neg_words = {"stress", "stressed", "overwhelmed", "anxiety", "anxious", "depressed", "depression", "fear", "terrified", "hopeless", "lonely", "isolation", "panic", "fatigue", "falling", "failing", "pain", "sadness", "emptiness", "burnout"} def calc_fp(text): words = str(text).lower().split() if not words: return 0.0 return round(sum(1 for w in words if w in first_person_words) / len(words), 4) def calc_neg(text): words = str(text).lower().split() if not words: return 0.0 return round(sum(1 for w in words if w in neg_words) / len(words), 4) final_df['first_person_ratio'] = final_df['text'].apply(calc_fp) final_df['negative_word_density'] = final_df['text'].apply(calc_neg) final_df['word_count'] = final_df['text'].apply(lambda x: len(str(x).split())) os.makedirs(os.path.dirname(TEXT_DATASET_PATH), exist_ok=True) final_df.to_csv(TEXT_DATASET_PATH, index=False) print(f"Successfully processed {len(final_df)} real text samples and saved to {TEXT_DATASET_PATH}") print("Category distribution:") print(final_df['category'].value_counts()) if __name__ == "__main__": ingest_text_data() ingest_audio_data()