pip install librosa numpy tensorflow scikit-learn soundfile matplotlib pandas

import os
import glob
import numpy as np
import pandas as pd
import librosa
import soundfile as sf
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder

# =============================================
# 1. RAVDESS 감정 레이블 매핑
# =============================================
EMOTION_MAP = {
'01': 'neutral',    # 중립
'02': 'calm',       # 차분
'03': 'happy',      # 기쁨
'04': 'sad',        # 슬픔
'05': 'angry',      # 분노
'06': 'fearful',    # 공포
'07': 'disgust',    # 혐오
'08': 'surprised'   # 놀람
}

# =============================================
# 2. 데이터셋 경로에서 파일 목록 & 레이블 추출
# =============================================
def load_ravdess_data(data_path):
"""
RAVDESS 데이터셋 로드 함수
data_path: RAVDESS 폴더 경로 (Actor_01 ~ Actor_24 포함)
"""
file_paths = []
labels = []

# 모든 wav 파일 탐색
for wav_file in glob.glob(os.path.join(data_path, '**', '*.wav'), recursive=True):
filename = os.path.basename(wav_file)
parts = filename.replace('.wav', '').split('-')

# 파일명 형식 검증
if len(parts) >= 3:
emotion_code = parts[2]  # 세 번째 요소가 감정 코드
if emotion_code in EMOTION_MAP:
file_paths.append(wav_file)
labels.append(EMOTION_MAP[emotion_code])

print(f"총 {len(file_paths)}개 파일 로드 완료")

# 감정별 분포 확인
label_series = pd.Series(labels)
print("\n감정별 데이터 분포:")
print(label_series.value_counts())

return file_paths, labels

# 사용 예시
# file_paths, labels = load_ravdess_data('./RAVDESS')

# =============================================
# 3. 음성 특징 추출 함수 (핵심!)
# =============================================
def extract_features(file_path, sr=22050, duration=3.0):
"""
음성 파일에서 다양한 특징을 추출하는 함수

추출 특징:
- MFCC (Mel-Frequency Cepstral Coefficients): 40개
- Mel Spectrogram: 128개
- Chroma Features: 12개
- Spectral Contrast: 7개
- Zero Crossing Rate: 1개
- RMS Energy: 1개
"""
try:
# 오디오 로드 (고정 길이로 패딩/자르기)
audio, sample_rate = librosa.load(
file_path, 
sr=sr, 
duration=duration,
mono=True
)

# 고정 길이 맞추기 (패딩 또는 자르기)
target_length = int(sr * duration)
if len(audio) < target_length:
# 짧으면 0으로 패딩
audio = np.pad(audio, (0, target_length - len(audio)))
else:
# 길면 자르기
audio = audio[:target_length]

# ---- MFCC 추출 ----
# 음성의 스펙트럼 포락선을 표현, 감정 인식에 가장 중요한 특징
mfcc = librosa.feature.mfcc(y=audio, sr=sample_rate, n_mfcc=40)
mfcc_mean = np.mean(mfcc, axis=1)      # 시간축 평균
mfcc_std = np.std(mfcc, axis=1)         # 시간축 표준편차
mfcc_delta = np.mean(librosa.feature.delta(mfcc), axis=1)  # 1차 미분

# ---- Mel Spectrogram ----
mel_spec = librosa.feature.melspectrogram(
y=audio, sr=sample_rate, n_mels=128
)
mel_spec_db = librosa.power_to_db(mel_spec, ref=np.max)
mel_mean = np.mean(mel_spec_db, axis=1)

# ---- Chroma Features (음정 관련) ----
chroma = librosa.feature.chroma_stft(y=audio, sr=sample_rate)
chroma_mean = np.mean(chroma, axis=1)

# ---- Spectral Contrast ----
contrast = librosa.feature.spectral_contrast(y=audio, sr=sample_rate)
contrast_mean = np.mean(contrast, axis=1)

# ---- Zero Crossing Rate (목소리 거칠기) ----
zcr = librosa.feature.zero_crossing_rate(audio)
zcr_mean = np.mean(zcr)

# ---- RMS Energy (음량/에너지) ----
rms = librosa.feature.rms(y=audio)
rms_mean = np.mean(rms)

# ---- Pitch (F0) 추출 ----
pitches, magnitudes = librosa.piptrack(y=audio, sr=sample_rate)
pitch_mean = np.mean(pitches[pitches > 0]) if np.any(pitches > 0) else 0
pitch_std = np.std(pitches[pitches > 0]) if np.any(pitches > 0) else 0

# 모든 특징 결합
features = np.concatenate([
mfcc_mean,          # 40
mfcc_std,           # 40
mfcc_delta,         # 40
mel_mean,           # 128
chroma_mean,        # 12
contrast_mean,      # 7
[zcr_mean],         # 1
[rms_mean],         # 1
[pitch_mean],       # 1
[pitch_std]         # 1
])
# 총 특징 벡터 크기: 271

return features

except Exception as e:
print(f"특징 추출 오류 ({file_path}): {e}")
return None


# =============================================
# 4. 전체 데이터셋 특징 추출
# =============================================
def prepare_dataset(file_paths, labels):
"""전체 데이터셋의 특징을 추출하고 배열로 반환"""
X = []
y = []

print("특징 추출 중...")
for i, (fp, label) in enumerate(zip(file_paths, labels)):
if i % 100 == 0:
print(f"  진행: {i}/{len(file_paths)}")

features = extract_features(fp)
if features is not None:
X.append(features)
y.append(label)

X = np.array(X)
y = np.array(y)

print(f"\n특징 추출 완료!")
print(f"X shape: {X.shape}")
print(f"y shape: {y.shape}")

return X, y

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers, Model
from sklearn.preprocessing import LabelEncoder, StandardScaler
from sklearn.metrics import classification_report, confusion_matrix
import numpy as np

# =============================================
# 5. 데이터 전처리 (스케일링 & 인코딩)
# =============================================
def preprocess_data(X, y, test_size=0.2, val_size=0.1):
"""
데이터 스케일링, 레이블 인코딩, 데이터 분할
"""
# 레이블 인코딩 (문자열 → 숫자)
le = LabelEncoder()
y_encoded = le.fit_transform(y)
y_onehot = keras.utils.to_categorical(y_encoded)

print(f"감정 클래스: {le.classes_}")
print(f"클래스 수: {len(le.classes_)}")

# 특징 정규화 (StandardScaler)
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# 학습/검증/테스트 분할
X_temp, X_test, y_temp, y_test = train_test_split(
X_scaled, y_onehot, test_size=test_size, 
random_state=42, stratify=y_encoded
)

val_ratio = val_size / (1 - test_size)
X_train, X_val, y_train, y_val = train_test_split(
X_temp, y_temp, test_size=val_ratio,
random_state=42
)

print(f"\n데이터 분할 결과:")
print(f"  학습: {X_train.shape[0]}개")
print(f"  검증: {X_val.shape[0]}개")
print(f"  테스트: {X_test.shape[0]}개")

return X_train, X_val, X_test, y_train, y_val, y_test, le, scaler


# =============================================
# 6. CNN + LSTM 하이브리드 모델
# =============================================
def build_cnn_lstm_model(input_dim, num_classes):
"""
CNN + LSTM 하이브리드 감정 분류 모델

구조:
Input → Reshape → Conv1D × 3 → LSTM × 2 → Dense → Output
"""
inputs = keras.Input(shape=(input_dim,), name='audio_features')

# 1D 배열을 시퀀스 형태로 변환 (CNN/LSTM 입력용)
x = layers.Reshape((input_dim, 1))(inputs)

# ---- CNN 블록 1 ----
x = layers.Conv1D(64, kernel_size=5, padding='same', activation='relu')(x)
x = layers.BatchNormalization()(x)
x = layers.MaxPooling1D(pool_size=2)(x)
x = layers.Dropout(0.3)(x)

# ---- CNN 블록 2 ----
x = layers.Conv1D(128, kernel_size=5, padding='same', activation='relu')(x)
x = layers.BatchNormalization()(x)
x = layers.MaxPooling1D(pool_size=2)(x)
x = layers.Dropout(0.3)(x)

# ---- CNN 블록 3 ----
x = layers.Conv1D(256, kernel_size=3, padding='same', activation='relu')(x)
x = layers.BatchNormalization()(x)
x = layers.MaxPooling1D(pool_size=2)(x)
x = layers.Dropout(0.3)(x)

# ---- LSTM 블록 1 ----
x = layers.LSTM(128, return_sequences=True, dropout=0.3)(x)
x = layers.BatchNormalization()(x)

# ---- LSTM 블록 2 ----
x = layers.LSTM(64, return_sequences=False, dropout=0.3)(x)
x = layers.BatchNormalization()(x)

# ---- Fully Connected 레이어 ----
x = layers.Dense(256, activation='relu')(x)
x = layers.Dropout(0.4)(x)
x = layers.Dense(128, activation='relu')(x)
x = layers.Dropout(0.3)(x)

# ---- 출력 레이어 ----
outputs = layers.Dense(num_classes, activation='softmax', name='emotion_output')(x)

model = Model(inputs=inputs, outputs=outputs, name='SER_CNN_LSTM')

return model


# =============================================
# 7. 모델 컴파일 & 학습
# =============================================
def train_model(X_train, X_val, y_train, y_val, num_classes, epochs=100):
"""모델 학습 함수"""

input_dim = X_train.shape[1]
model = build_cnn_lstm_model(input_dim, num_classes)

# 모델 구조 출력
model.summary()

# 컴파일
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=0.001),
loss='categorical_crossentropy',
metrics=['accuracy']
)

# 콜백 설정
callbacks = [
# 검증 손실이 개선되지 않으면 학습률 감소
keras.callbacks.ReduceLROnPlateau(
monitor='val_loss',
factor=0.5,
patience=10,
min_lr=1e-6,
verbose=1
),
# 과적합 방지: 검증 손실 기준 조기 종료
keras.callbacks.EarlyStopping(
monitor='val_loss',
patience=20,
restore_best_weights=True,
verbose=1
),
# 최고 성능 모델 저장
keras.callbacks.ModelCheckpoint(
'best_ser_model.h5',
monitor='val_accuracy',
save_best_only=True,
verbose=1
)
]

# 학습 시작!
history = model.fit(
X_train, y_train,
validation_data=(X_val, y_val),
epochs=epochs,
batch_size=32,
callbacks=callbacks,
verbose=1
)

return model, history

# =============================================
# 8. 음성 데이터 증강 기법들
# =============================================
def augment_audio(audio, sr=22050, augment_type='random'):
"""
음성 데이터 증강 함수

증강 기법:
1. 노이즈 추가 (Noise Addition)
2. 시간 이동 (Time Shifting)
3. 피치 변환 (Pitch Shifting)
4. 속도 변환 (Time Stretching)
5. 볼륨 조절 (Volume Scaling)
"""
import random

if augment_type == 'random':
augment_type = random.choice([
'noise', 'shift', 'pitch', 'stretch', 'volume'
])

augmented = audio.copy()

if augment_type == 'noise':
# 가우시안 노이즈 추가 (실제 환경 시뮬레이션)
noise_factor = 0.005
noise = np.random.randn(len(audio))
augmented = audio + noise_factor * noise
augmented = augmented.astype(np.float32)

elif augment_type == 'shift':
# 시간 이동 (앞뒤로 최대 1초 이동)
shift_max = int(sr * 0.5)  # 0.5초
shift = np.random.randint(-shift_max, shift_max)
augmented = np.roll(audio, shift)
# 이동된 부분 0으로 채우기
if shift > 0:
augmented[:shift] = 0
elif shift < 0:
augmented[shift:] = 0

elif augment_type == 'pitch':
# 피치 변환 (-3 ~ +3 반음)
n_steps = np.random.uniform(-3, 3)
augmented = librosa.effects.pitch_shift(
audio, sr=sr, n_steps=n_steps
)

elif augment_type == 'stretch':
# 속도 변환 (0.8x ~ 1.2x)
rate = np.random.uniform(0.8, 1.2)
augmented = librosa.effects.time_stretch(audio, rate=rate)
# 길이 맞추기
target_len = len(audio)
if len(augmented) > target_len:
augmented = augmented[:target_len]
else:
augmented = np.pad(augmented, (0, target_len - len(augmented)))

elif augment_type == 'volume':
# 볼륨 조절 (0.5x ~ 1.5x)
factor = np.random.uniform(0.5, 1.5)
augmented = audio * factor

return augmented


# =============================================
# 9. 증강 포함 특징 추출 (데이터셋 확장)
# =============================================
def extract_features_with_augmentation(file_path, sr=22050, 
duration=3.0, n_augments=2):
"""
원본 + 증강 데이터의 특징을 함께 추출
n_augments: 증강 횟수 (원본 1개 + 증강 n개)
"""
features_list = []

try:
audio, sample_rate = librosa.load(file_path, sr=sr, 
duration=duration, mono=True)
target_length = int(sr * duration)
if len(audio) < target_length:
audio = np.pad(audio, (0, target_length - len(audio)))
else:
audio = audio[:target_length]

# 원본 특징 추출
# (extract_features 함수를 오디오 배열 버전으로 수정 필요)
features_list.append(extract_features_from_array(audio, sr))

# 증강 데이터 특징 추출
for _ in range(n_augments):
aug_audio = augment_audio(audio, sr)
features_list.append(extract_features_from_array(aug_audio, sr))

except Exception as e:
print(f"오류: {e}")

return features_list


def extract_features_from_array(audio, sr=22050):
"""오디오 배열에서 직접 특징 추출 (파일 경로 대신 배열 입력)"""
mfcc = librosa.feature.mfcc(y=audio, sr=sr, n_mfcc=40)
mfcc_mean = np.mean(mfcc, axis=1)
mfcc_std = np.std(mfcc, axis=1)
mfcc_delta = np.mean(librosa.feature.delta(mfcc), axis=1)

mel_spec = librosa.feature.melspectrogram(y=audio, sr=sr, n_mels=128)
mel_spec_db = librosa.power_to_db(mel_spec, ref=np.max)
mel_mean = np.mean(mel_spec_db, axis=1)

chroma = librosa.feature.chroma_stft(y=audio, sr=sr)
chroma_mean = np.mean(chroma, axis=1)

contrast = librosa.feature.spectral_contrast(y=audio, sr=sr)
contrast_mean = np.mean(contrast, axis=1)

zcr_mean = np.mean(librosa.feature.zero_crossing_rate(audio))
rms_mean = np.mean(librosa.feature.rms(y=audio))

pitches, _ = librosa.piptrack(y=audio, sr=sr)
pitch_mean = np.mean(pitches[pitches > 0]) if np.any(pitches > 0) else 0
pitch_std = np.std(pitches[pitches > 0]) if np.any(pitches > 0) else 0

return np.concatenate([
mfcc_mean, mfcc_std, mfcc_delta, mel_mean,
chroma_mean, contrast_mean,
[zcr_mean], [rms_mean], [pitch_mean], [pitch_std]
])

import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import classification_report, confusion_matrix

# =============================================
# 10. 모델 평가 함수
# =============================================
def evaluate_model(model, X_test, y_test, le):
"""
모델 성능 평가 및 시각화
"""
# 예측
y_pred_prob = model.predict(X_test)
y_pred = np.argmax(y_pred_prob, axis=1)
y_true = np.argmax(y_test, axis=1)

# 전체 정확도
accuracy = np.mean(y_pred == y_true)
print(f"\n=== 모델 평가 결과 ===")
print(f"테스트 정확도: {accuracy:.4f} ({accuracy*100:.2f}%)")

# 감정별 상세 성능 (Precision, Recall, F1-Score)
print("\n=== 감정별 성능 ===")
print(classification_report(
y_true, y_pred, 
target_names=le.classes_
))

# 혼동 행렬 시각화
cm = confusion_matrix(y_true, y_pred)

plt.figure(figsize=(10, 8))
sns.heatmap(
cm, 
annot=True, 
fmt='d',
cmap='Blues',
xticklabels=le.classes_,
yticklabels=le.classes_
)
plt.title('감정 분류 혼동 행렬 (Confusion Matrix)', fontsize=14, pad=15)
plt.ylabel('실제 감정', fontsize=12)
plt.xlabel('예측 감정', fontsize=12)
plt.tight_layout()
plt.savefig('confusion_matrix.png', dpi=150, bbox_inches='tight')
plt.show()

return accuracy, y_pred, y_pred_prob


# =============================================
# 11. 학습 곡선 시각화
# =============================================
def plot_training_history(history):
"""학습/검증 손실 및 정확도 시각화"""

fig, axes = plt.subplots(1, 2, figsize=(14, 5))

# 정확도 그래프
axes[0].plot(history.history['accuracy'], 
label='학습 정확도', color='#667eea', linewidth=2)
axes[0].plot(history.history['val_accuracy'], 
label='검증 정확도', color='#764ba2', linewidth=2, linestyle='--')
axes[0].set_title('모델 정확도', fontsize=13, fontweight='bold')
axes[0].set_xlabel('에폭')
axes[0].set_ylabel('정확도')
axes[0].legend()
axes[0].grid(True, alpha=0.3)

# 손실 그래프
axes[1].plot(history.history['loss'], 
label='학습 손실', color='#e17055', linewidth=2)
axes[1].plot(history.history['val_loss'], 
label='검증 손실', color='#fdcb6e', linewidth=2, linestyle='--')
axes[1].set_title('모델 손실', fontsize=13, fontweight='bold')
axes[1].set_xlabel('에폭')
axes[1].set_ylabel('손실')
axes[1].legend()
axes[1].grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('training_history.png', dpi=150, bbox_inches='tight')
plt.show()

import pyaudio
import wave
import threading
import queue
import time

# =============================================
# 12. 실시간 음성 감정 분석기
# =============================================
class RealTimeSER:
"""
실시간 음성 감정 분석 클래스
마이크 입력을 받아 실시간으로 감정을 분류
"""

# 감정 이모지 매핑
EMOTION_EMOJI = {
'neutral':   '😐',
'calm':      '😌',
'happy':     '😊',
'sad':       '😢',
'angry':     '😠',
'fearful':   '😨',
'disgust':   '🤢',
'surprised': '😲'
}

def __init__(self, model, scaler, label_encoder, 
sr=22050, duration=3.0, chunk_size=1024):
self.model = model
self.scaler = scaler
self.le = label_encoder
self.sr = sr
self.duration = duration
self.chunk_size = chunk_size
self.is_running = False
self.audio_queue = queue.Queue()

def record_audio(self):
"""마이크에서 오디오 녹음 (별도 스레드)"""
p = pyaudio.PyAudio()

stream = p.open(
format=pyaudio.paFloat32,
channels=1,
rate=self.sr,
input=True,
frames_per_buffer=self.chunk_size
)

print("🎙️ 녹음 시작! (Ctrl+C로 종료)")

while self.is_running:
frames = []
n_frames = int(self.sr * self.duration / self.chunk_size)

for _ in range(n_frames):
if not self.is_running:
break
data = stream.read(self.chunk_size, exception_on_overflow=False)
frames.append(np.frombuffer(data, dtype=np.float32))

if frames:
audio_data = np.concatenate(frames)
self.audio_queue.put(audio_data)

stream.stop_stream()
stream.close()
p.terminate()

def analyze_emotion(self):
"""큐에서 오디오를 꺼내 감정 분석 (별도 스레드)"""

while self.is_running:
try:
audio_data = self.audio_queue.get(timeout=1.0)

# 특징 추출
features = extract_features_from_array(audio_data, self.sr)
features_scaled = self.scaler.transform(features.reshape(1, -1))

# 감정 예측
pred_prob = self.model.predict(features_scaled, verbose=0)[0]
pred_idx = np.argmax(pred_prob)
pred_emotion = self.le.classes_[pred_idx]
confidence = pred_prob[pred_idx]

# 결과 출력
emoji = self.EMOTION_EMOJI.get(pred_emotion, '❓')
print(f"\r{emoji} 감정: {pred_emotion:10s} | "
f"신뢰도: {confidence:.2%} | "
f"{'█' * int(confidence * 20)}", end='', flush=True)

# 상위 3개 감정 출력
top3_idx = np.argsort(pred_prob)[-3:][::-1]
print(f"\n  상위 3개: ", end='')
for idx in top3_idx:
em = self.le.classes_[idx]
prob = pred_prob[idx]
print(f"{self.EMOTION_EMOJI.get(em,'')} {em}({prob:.1%}) ", end='')
print()

except queue.Empty:
continue
except Exception as e:
print(f"\n분석 오류: {e}")

def start(self):
"""실시간 분석 시작"""
self.is_running = True

# 녹음 스레드
record_thread = threading.Thread(target=self.record_audio)
record_thread.daemon = True
record_thread.start()

# 분석 스레드
analyze_thread = threading.Thread(target=self.analyze_emotion)
analyze_thread.daemon = True
analyze_thread.start()

try:
while True:
time.sleep(0.1)
except KeyboardInterrupt:
print("\n\n분석 종료!")
self.is_running = False

def analyze_file(self, file_path):
"""단일 파일 감정 분석"""
features = extract_features(file_path)
if features is None:
return None

features_scaled = self.scaler.transform(features.reshape(1, -1))
pred_prob = self.model.predict(features_scaled, verbose=0)[0]
pred_idx = np.argmax(pred_prob)
pred_emotion = self.le.classes_[pred_idx]
confidence = pred_prob[pred_idx]

result = {
'emotion': pred_emotion,
'confidence': float(confidence),
'emoji': self.EMOTION_EMOJI.get(pred_emotion, '❓'),
'all_probabilities': {
self.le.classes_[i]: float(pred_prob[i]) 
for i in range(len(self.le.classes_))
}
}

return result


# =============================================
# 사용 예시
# =============================================
# 모델 로드 후 실시간 분석 시작
# model = keras.models.load_model('best_ser_model.h5')
# ser = RealTimeSER(model, scaler, le)
# 
# 파일 분석
# result = ser.analyze_file('test_audio.wav')
# print(f"감정: {result['emoji']} {result['emotion']} ({result['confidence']:.1%})")
#
# 실시간 분석
# ser.start()

# pip install transformers datasets torch torchaudio

from transformers import (
Wav2Vec2Processor, 
Wav2Vec2ForSequenceClassification,
TrainingArguments, 
Trainer
)
import torch
import torchaudio
from datasets import Dataset, DatasetDict
import numpy as np

# =============================================
# 13. Wav2Vec2 기반 감정 분류 모델
# =============================================

# 감정 레이블 정의
EMOTION_LABELS = ['neutral', 'calm', 'happy', 'sad', 
'angry', 'fearful', 'disgust', 'surprised']
label2id = {label: i for i, label in enumerate(EMOTION_LABELS)}
id2label = {i: label for i, label in enumerate(EMOTION_LABELS)}

def load_wav2vec2_model(num_labels=8):
"""
사전학습된 Wav2Vec2 모델 로드 및 분류 헤드 추가
"""
model_name = "facebook/wav2vec2-base"  # 또는 "facebook/wav2vec2-large"

# 프로세서 로드 (오디오 전처리 담당)
processor = Wav2Vec2Processor.from_pretrained(model_name)

# 분류 모델 로드
model = Wav2Vec2ForSequenceClassification.from_pretrained(
model_name,
num_labels=num_labels,
label2id=label2id,
id2label=id2label,
ignore_mismatched_sizes=True
)

# 특징 추출기 레이어 동결 (파인튜닝 시 선택적)
# model.freeze_feature_encoder()

return model, processor


def preprocess_audio_for_wav2vec2(file_path, processor, sr=16000, max_duration=5.0):
"""
Wav2Vec2 입력 형식으로 오디오 전처리
"""
# 오디오 로드 (16kHz 필수!)
audio, orig_sr = torchaudio.load(file_path)

# 모노 변환
if audio.shape[0] > 1:
audio = torch.mean(audio, dim=0, keepdim=True)

# 리샘플링 (Wav2Vec2는 16kHz 입력 필요)
if orig_sr != sr:
resampler = torchaudio.transforms.Resample(orig_sr, sr)
audio = resampler(audio)

audio = audio.squeeze().numpy()

# 길이 제한
max_length = int(sr * max_duration)
if len(audio) > max_length:
audio = audio[:max_length]

# 프로세서로 정규화
inputs = processor(
audio,
sampling_rate=sr,
return_tensors="pt",
padding=True,
max_length=max_length,
truncation=True
)

return inputs


# =============================================
# 14. Trainer API를 활용한 파인튜닝
# =============================================
def finetune_wav2vec2(train_dataset, eval_dataset, model, processor):
"""
Wav2Vec2 파인튜닝 설정 및 학습
"""

training_args = TrainingArguments(
output_dir="./wav2vec2-emotion",
num_train_epochs=10,
per_device_train_batch_size=8,
per_device_eval_batch_size=8,
gradient_accumulation_steps=2,
evaluation_strategy="epoch",
save_strategy="epoch",
load_best_model_at_end=True,
metric_for_best_model="accuracy",
learning_rate=3e-5,
warmup_ratio=0.1,
logging_dir="./logs",
logging_steps=10,
fp16=True,  # GPU 메모리 절약 (CUDA 필요)
dataloader_num_workers=4,
report_to="none"
)

def compute_metrics(eval_pred):
"""평가 지표 계산"""
logits, labels = eval_pred
predictions = np.argmax(logits, axis=-1)
accuracy = np.mean(predictions == labels)
return {"accuracy": accuracy}

trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
eval_dataset=eval_dataset,
compute_metrics=compute_metrics,
)

trainer.train()

# 최종 평가
results = trainer.evaluate()
print(f"\nWav2Vec2 파인튜닝 결과: {results}")

return trainer

from flask import Flask, request, jsonify
import tensorflow as tf
from tensorflow import keras
import numpy as np
import librosa
import pickle
import io
import os
import tempfile

# =============================================
# 15. Flask REST API 서버
# =============================================
app = Flask(__name__)

# 전역 변수로 모델 로드 (서버 시작 시 1회만)
MODEL = None
SCALER = None
LABEL_ENCODER = None

def load_model_and_preprocessors():
"""서버 시작 시 모델 및 전처리기 로드"""
global MODEL, SCALER, LABEL_ENCODER

MODEL = keras.models.load_model('best_ser_model.h5')

with open('scaler.pkl', 'rb') as f:
SCALER = pickle.load(f)

with open('label_encoder.pkl', 'rb') as f:
LABEL_ENCODER = pickle.load(f)

print("✅ 모델 로드 완료!")


@app.route('/health', methods=['GET'])
def health_check():
"""서버 상태 확인"""
return jsonify({
'status': 'healthy',
'model_loaded': MODEL is not None,
'version': '1.0.0'
})


@app.route('/analyze', methods=['POST'])
def analyze_emotion():
"""
음성 파일 감정 분석 API

Request: multipart/form-data
- file: 오디오 파일 (WAV, MP3 등)

Response: JSON
- emotion: 예측된 감정
- confidence: 신뢰도
- all_emotions: 전체 감정별 확률
"""
if 'file' not in request.files:
return jsonify({'error': '파일이 없습니다'}), 400

audio_file = request.files['file']

if audio_file.filename == '':
return jsonify({'error': '파일명이 비어있습니다'}), 400

# 허용된 파일 형식 확인
allowed_extensions = {'.wav', '.mp3', '.flac', '.ogg', '.m4a'}
file_ext = os.path.splitext(audio_file.filename)[1].lower()

if file_ext not in allowed_extensions:
return jsonify({
'error': f'지원하지 않는 파일 형식: {file_ext}'
}), 400

try:
# 임시 파일로 저장
with tempfile.NamedTemporaryFile(
suffix=file_ext, delete=False
) as tmp_file:
audio_file.save(tmp_file.name)
tmp_path = tmp_file.name

# 특징 추출
features = extract_features(tmp_path)

# 임시 파일 삭제
os.unlink(tmp_path)

if features is None:
return jsonify({'error': '특징 추출 실패'}), 500

# 스케일링 & 예측
features_scaled = SCALER.transform(features.reshape(1, -1))
pred_prob = MODEL.predict(features_scaled, verbose=0)[0]
pred_idx = np.argmax(pred_prob)
pred_emotion = LABEL_ENCODER.classes_[pred_idx]
confidence = float(pred_prob[pred_idx])

# 전체 감정 확률 딕셔너리
all_emotions = {
LABEL_ENCODER.classes_[i]: float(pred_prob[i])
for i in range(len(LABEL_ENCODER.classes_))
}

# 상위 3개 감정
top3_idx = np.argsort(pred_prob)[-3:][::-1]
top3 = [
{
'emotion': LABEL_ENCODER.classes_[i],
'probability': float(pred_prob[i])
}
for i in top3_idx
]

return jsonify({
'success': True,
'emotion': pred_emotion,
'confidence': confidence,
'top3_emotions': top3,
'all_emotions': all_emotions
})

except Exception as e:
return jsonify({'error': str(e)}), 500


@app.route('/batch_analyze', methods=['POST'])
def batch_analyze():
"""여러 파일 일괄 분석"""
files = request.files.getlist('files')

if not files:
return jsonify({'error': '파일이 없습니다'}), 400

results = []
for audio_file in files:
# 각 파일 처리 (analyze_emotion 로직 재사용)
# ... (생략)
pass

return jsonify({'results': results})


if __name__ == '__main__':
load_model_and_preprocessors()
app.run(host='0.0.0.0', port=5000, debug=False)

# curl로 테스트
# curl -X POST http://localhost:5000/analyze \
#      -F "file=@test_audio.wav"

# Python requests로 테스트
import requests

url = "http://localhost:5000/analyze"
with open("test_audio.wav", "rb") as f:
response = requests.post(url, files={"file": f})

result = response.json()
print(f"감정: {result['emotion']}")
print(f"신뢰도: {result['confidence']:.2%}")
print(f"상위 3개: {result['top3_emotions']}")

# =============================================
# main.py — 음성 감정 분석 전체 파이프라인
# =============================================
import os
import pickle
import numpy as np
import warnings
warnings.filterwarnings('ignore')

def main():
print("=" * 60)
print("🎙️  음성 감정 분석 시스템 (SER)")
print("=" * 60)

# ---- 설정 ----
DATA_PATH = './RAVDESS'          # 데이터셋 경로
MODEL_PATH = 'best_ser_model.h5' # 모델 저장 경로
SCALER_PATH = 'scaler.pkl'
LE_PATH = 'label_encoder.pkl'

# ---- 1단계: 데이터 로드 ----
print("\n[1/5] 데이터 로드 중...")
file_paths, labels = load_ravdess_data(DATA_PATH)

# ---- 2단계: 특징 추출 ----
print("\n[2/5] 특징 추출 중...")
X, y = prepare_dataset(file_paths, labels)

# ---- 3단계: 데이터 전처리 ----
print("\n[3/5] 데이터 전처리 중...")
X_train, X_val, X_test, y_train, y_val, y_test, le, scaler = \
preprocess_data(X, y)

# 전처리기 저장
with open(SCALER_PATH, 'wb') as f:
pickle.dump(scaler, f)
with open(LE_PATH, 'wb') as f:
pickle.dump(le, f)
print(f"전처리기 저장 완료: {SCALER_PATH}, {LE_PATH}")

# ---- 4단계: 모델 학습 ----
print("\n[4/5] 모델 학습 중...")
num_classes = len(le.classes_)
model, history = train_model(
X_train, X_val, y_train, y_val, 
num_classes=num_classes, 
epochs=100
)

# 학습 곡선 시각화
plot_training_history(history)

# ---- 5단계: 모델 평가 ----
print("\n[5/5] 모델 평가 중...")
accuracy, y_pred, y_pred_prob = evaluate_model(model, X_test, y_test, le)

print(f"\n{'='*60}")
print(f"✅ 학습 완료!")
print(f"   최종 테스트 정확도: {accuracy:.4f} ({accuracy*100:.2f}%)")
print(f"   모델 저장 위치: {MODEL_PATH}")
print(f"{'='*60}")

# ---- 단일 파일 테스트 ----
print("\n[테스트] 단일 파일 감정 분석:")
test_file = file_paths[0]  # 첫 번째 파일로 테스트

ser = RealTimeSER(model, scaler, le)
result = ser.analyze_file(test_file)

if result:
print(f"  파일: {os.path.basename(test_file)}")
print(f"  예측 감정: {result['emoji']} {result['emotion']}")
print(f"  신뢰도: {result['confidence']:.2%}")
print(f"  전체 확률:")
for emotion, prob in sorted(
result['all_probabilities'].items(), 
key=lambda x: x[1], reverse=True
):
bar = '█' * int(prob * 30)
print(f"    {emotion:12s}: {bar} {prob:.2%}")

return model, scaler, le


if __name__ == '__main__':
model, scaler, le = main()

# 한국어 Wav2Vec2 모델 활용 예시
from transformers import Wav2Vec2Processor, Wav2Vec2ForSequenceClassification

# 한국어 사전학습 모델 로드
KO_MODEL_NAME = "kresnik/wav2vec2-large-xlsr-korean"

processor = Wav2Vec2Processor.from_pretrained(KO_MODEL_NAME)
model = Wav2Vec2ForSequenceClassification.from_pretrained(
KO_MODEL_NAME,
num_labels=7,  # 한국어 감정 클래스 수
label2id={
'기쁨': 0, '슬픔': 1, '분노': 2, '공포': 3,
'혐오': 4, '놀람': 5, '중립': 6
},
id2label={
0: '기쁨', 1: '슬픔', 2: '분노', 3: '공포',
4: '혐오', 5: '놀람', 6: '중립'
},
ignore_mismatched_sizes=True
)

# 이후 파인튜닝 과정은 영어 모델과 동일!