from flask import Flask, request, jsonify from flask_cors import CORS import numpy as np import requests import os import base64 import io import wave import sys import tempfile import importlib.util import contextlib from typing import Dict, Any, List import json import logging from urllib.parse import urlparse from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() os.environ.setdefault("HF_HUB_DISABLE_XET", "1") app = Flask(__name__) CORS(app) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[ logging.StreamHandler(), ], ) logger = logging.getLogger("voice-emotion-api") logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) logging.getLogger("huggingface_hub").setLevel(logging.WARNING) EMOTIONS = ["anger", "disgust", "fear", "happy", "neutral", "sad"] LOCAL_MODEL_DEFAULTS = { "cnn": "https://rvc-001-voice-based-emotion.hf.space/models/cnn/predict", "wav2vec": "https://rvc-001-voice-based-emotion.hf.space/models/wav2vec/predict", } LOCAL_HOSTS = {"127.0.0.1", "localhost", "rvc-001-voice-based-emotion.hf.space"} if os.environ.get("SPACE_HOST"): LOCAL_HOSTS.add(os.environ["SPACE_HOST"]) DEFAULT_WAV2VEC_MODEL_ID = "superb/wav2vec2-base-superb-er" DEFAULT_CNN_MODEL_REPO = "saadmannan/speech-emotion-recognition" DEFAULT_CNN_PIPELINE_MODEL_ID = "ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition" TARGET_SAMPLE_RATE = 16000 BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_CNN_MODEL_PATH = os.path.join(BASE_DIR, "assets", "models", "cnn") DEFAULT_WAV2VEC_MODEL_PATH = os.path.join(BASE_DIR, "assets", "models", "wav2vec") MODEL_CACHE: Dict[str, Any] = {} MODEL_STATUS: Dict[str, Dict[str, str]] = { "cnn": {"source": "not_loaded", "detail": "CNN model has not been requested yet"}, "wav2vec": {"source": "not_loaded", "detail": "wav2vec model has not been requested yet"}, } def get_env(*names: str, default: str | None = None) -> str | None: for name in names: value = os.getenv(name) if value: return value return default def get_model_endpoint(model_name: str) -> str: if model_name == "cnn": return get_env("CNN_API_URL", default=LOCAL_MODEL_DEFAULTS["cnn"]) return get_env("WAV2VEC_API_URL", default=LOCAL_MODEL_DEFAULTS["wav2vec"]) def get_wav2vec_model_id() -> str: model_path = resolve_local_path(get_env("WAV2VEC_MODEL_PATH")) if model_path and os.path.isdir(model_path): return model_path if os.path.isdir(DEFAULT_WAV2VEC_MODEL_PATH): return DEFAULT_WAV2VEC_MODEL_PATH return get_env("WAV2VEC_MODEL_ID", default=DEFAULT_WAV2VEC_MODEL_ID) or DEFAULT_WAV2VEC_MODEL_ID def get_cnn_model_repo() -> str: model_path = resolve_local_path(get_env("CNN_MODEL_PATH")) if model_path and os.path.isdir(model_path): return model_path if os.path.isdir(DEFAULT_CNN_MODEL_PATH): return DEFAULT_CNN_MODEL_PATH return get_env("CNN_MODEL_REPO", default=DEFAULT_CNN_MODEL_REPO) or DEFAULT_CNN_MODEL_REPO def allow_model_downloads() -> bool: # Always allow model downloads in the Hugging Face Space environment # to ensure the transformer models can be fetched properly. return True def resolve_local_path(path_value: str | None) -> str | None: if not path_value: return None if os.path.isabs(path_value): return path_value return os.path.join(BASE_DIR, path_value) logger.info("Loaded backend environment variables") logger.info(" CNN_API_URL=%s", get_model_endpoint("cnn")) logger.info(" WAV2VEC_API_URL=%s", get_model_endpoint("wav2vec")) @app.before_request def log_request() -> None: logger.info( "%s %s from %s content_type=%s content_length=%s", request.method, request.path, request.remote_addr, request.content_type, request.content_length, ) @app.after_request def log_response(response): logger.info("%s %s -> %s", request.method, request.path, response.status_code) return response @app.errorhandler(Exception) def handle_unexpected_error(error): logger.exception("Unhandled backend error") return jsonify( build_error_response( f"Server error: {error}", warnings=[f"Server error: {error}"], ) ), 500 def build_error_response( error: str, *, duration: float = 0.0, warnings: List[str] | None = None, preferred_model: str = "wav2vec", ) -> Dict[str, Any]: empty_scores = {e: 0.0 for e in EMOTIONS} return { "error": error, "final": {"emotion": "neutral", "confidence": 0.0, "uncertain": True}, "comparison": { "agreement": False, "preferredModel": preferred_model, }, "models": { "cnn": {"emotion": "neutral", "confidence": 0.0, "scores": dict(empty_scores)}, "wav2vec": {"emotion": "neutral", "confidence": 0.0, "scores": dict(empty_scores)}, }, "meta": { "duration": round(duration, 2), "warnings": warnings or [error], }, } def normalize_emotion_prediction( predictions: List[Dict[str, Any]], ) -> Dict[str, float]: """ Normalize predictions to our standard 6-emotion classes: anger, disgust, fear, happy, neutral, sad. Key design notes: - Scores are SUMMED (not maxed) so that multiple source labels mapping to the same target correctly accumulate their probabilities. - 'calm' is split 50/50 between neutral and sad (low-arousal emotion). - 'surprise' maps to happy (high-arousal, positive valence). - Unknown labels are silently ignored — no hardcoded defaults. """ # Direct 1-to-1 mappings (exact label first, then substring fallback) emotion_map = { # anger "ang": "anger", "angry": "anger", "anger": "anger", # disgust "disgust": "disgust", "dis": "disgust", # fear "fear": "fear", "fea": "fear", "fearful": "fear", "scared": "fear", # happy "happy": "happy", "hap": "happy", "joy": "happy", "happiness": "happy", "excited": "happy", "surprise": "happy", "surprised": "happy", "pleasant": "happy", # neutral (calm is handled separately below) "neutral": "neutral", "neu": "neutral", # sad "sad": "sad", "sadness": "sad", } # Labels that split their probability across two target emotions split_map: Dict[str, List[tuple]] = { # calm → 50 % neutral + 50 % sad "calm": [("neutral", 0.50), ("sad", 0.50)], } accumulated: Dict[str, float] = {emotion: 0.0 for emotion in EMOTIONS} total_mapped = 0.0 for pred in predictions: if not (isinstance(pred, dict) and "label" in pred and "score" in pred): continue label = str(pred["label"]).lower().strip() score = float(pred["score"]) if score > 1.0: score = score / 100.0 # Check split labels first split_targets = None if label in split_map: split_targets = split_map[label] else: for key, targets in split_map.items(): if key in label: split_targets = targets break if split_targets: for target_emotion, weight in split_targets: accumulated[target_emotion] += score * weight total_mapped += score continue # Standard 1-to-1 mapping — accumulate (sum) into the target bucket mapped_emotion = None if label in emotion_map: mapped_emotion = emotion_map[label] else: for key, value in emotion_map.items(): if key in label: mapped_emotion = value break if mapped_emotion and mapped_emotion in accumulated: accumulated[mapped_emotion] += score total_mapped += score # Unknown labels are intentionally ignored — no hardcoded fallback # If nothing mapped at all, return an equal prior (genuinely unknown) if total_mapped == 0.0 or all(s == 0.0 for s in accumulated.values()): equal_prob = 1.0 / len(EMOTIONS) return {e: equal_prob for e in EMOTIONS} # Re-normalise so scores sum to 1.0 (preserves relative ordering) total = sum(accumulated.values()) or 1.0 return {e: accumulated[e] / total for e in EMOTIONS} def get_highest_confidence_emotion( normalized_scores: Dict[str, float], ) -> tuple[str, float]: best_emotion = max(normalized_scores.items(), key=lambda item: item[1]) return best_emotion[0], round(best_emotion[1], 2) def clamp(value: float, minimum: float = 0.0, maximum: float = 1.0) -> float: return max(minimum, min(maximum, float(value))) def scale_feature(value: float, low: float, high: float) -> float: if high <= low: return 0.0 return clamp((value - low) / (high - low)) def normalize_distribution(scores: Dict[str, float]) -> Dict[str, float]: adjusted = {label: max(0.0, float(score)) for label, score in scores.items()} total = sum(adjusted.values()) if total <= 0.0: equal = round(1.0 / max(1, len(adjusted)), 4) return {label: equal for label in adjusted} return {label: round(score / total, 4) for label, score in adjusted.items()} def resample_audio(audio_data: np.ndarray, sample_rate: int, target_rate: int = TARGET_SAMPLE_RATE) -> np.ndarray: audio = np.asarray(audio_data, dtype=np.float32) if sample_rate == target_rate or len(audio) == 0: return audio duration = len(audio) / sample_rate target_length = max(1, int(round(duration * target_rate))) source_positions = np.linspace(0.0, duration, num=len(audio), endpoint=False) target_positions = np.linspace(0.0, duration, num=target_length, endpoint=False) return np.interp(target_positions, source_positions, audio).astype(np.float32) def trim_silence(audio_data: np.ndarray, sample_rate: int) -> np.ndarray: audio = np.asarray(audio_data, dtype=np.float32) if len(audio) == 0: return audio frame_length = min(2048, max(512, int(sample_rate * 0.03))) hop_length = max(128, frame_length // 2) frames = frame_audio(audio, frame_length, hop_length) rms = np.sqrt(np.mean(np.square(frames), axis=1)) if len(rms) == 0: return audio reference = float(np.percentile(rms, 90)) threshold = max(0.008, reference * 0.18) active = rms >= threshold if not np.any(active): return audio lead_padding = 2 tail_padding = 2 start_frame = max(0, int(np.argmax(active)) - lead_padding) end_frame = min(len(active), len(active) - int(np.argmax(active[::-1])) + tail_padding) start = start_frame * hop_length end = min(len(audio), end_frame * hop_length + frame_length) trimmed = audio[start:end] return trimmed if len(trimmed) else audio def preprocess_audio_for_emotion(audio_data: np.ndarray, sample_rate: int) -> np.ndarray: audio = np.asarray(audio_data, dtype=np.float32) if audio.ndim > 1: audio = np.mean(audio, axis=1) audio = audio[np.isfinite(audio)] if len(audio) == 0: return np.asarray([], dtype=np.float32) audio = audio - float(np.mean(audio)) trimmed = trim_silence(audio, sample_rate) peak = float(np.max(np.abs(trimmed))) if len(trimmed) else 0.0 if peak > 0.98: trimmed = trimmed / peak * 0.98 elif 0.03 <= peak < 0.92: trimmed = trimmed / peak * 0.92 return np.clip(trimmed, -1.0, 1.0).astype(np.float32) def build_inference_segments( audio_data: np.ndarray, sample_rate: int, window_seconds: float = 4.0, stride_seconds: float = 2.0, ) -> List[tuple[np.ndarray, float]]: audio = np.asarray(audio_data, dtype=np.float32) if len(audio) == 0: return [] window = max(1, int(window_seconds * sample_rate)) stride = max(1, int(stride_seconds * sample_rate)) min_length = max(1, int(1.0 * sample_rate)) if len(audio) <= window: rms = float(np.sqrt(np.mean(np.square(audio)))) if len(audio) else 0.0 return [(audio, max(0.2, scale_feature(rms, 0.01, 0.08)))] starts = list(range(0, max(1, len(audio) - window + 1), stride)) last_start = max(0, len(audio) - window) if last_start not in starts: starts.append(last_start) segments: List[tuple[np.ndarray, float]] = [] for start in starts: segment = audio[start : start + window] if len(segment) < min_length: continue trimmed = trim_silence(segment, sample_rate) if len(trimmed) < min_length: continue rms = float(np.sqrt(np.mean(np.square(trimmed)))) if len(trimmed) else 0.0 weight = max(0.2, scale_feature(rms, 0.01, 0.08)) segments.append((trimmed, weight)) if segments: return segments rms = float(np.sqrt(np.mean(np.square(audio)))) if len(audio) else 0.0 return [(audio, max(0.2, scale_feature(rms, 0.01, 0.08)))] def softmax(values: np.ndarray) -> np.ndarray: shifted = values - np.max(values, axis=-1, keepdims=True) exp_values = np.exp(shifted) return exp_values / np.sum(exp_values, axis=-1, keepdims=True) def update_model_status(model_name: str, source: str, detail: str) -> None: MODEL_STATUS[model_name] = {"source": source, "detail": detail} def run_wav2vec_transformer_model( audio_data: np.ndarray, sample_rate: int, ) -> Dict[str, Any]: model_id = get_wav2vec_model_id() cache_key = f"wav2vec:{model_id}" try: if cache_key not in MODEL_CACHE: from transformers import AutoFeatureExtractor, AutoModelForAudioClassification import torch try: previous_offline = os.environ.get("HF_HUB_OFFLINE") os.environ["HF_HUB_OFFLINE"] = "1" try: feature_extractor = AutoFeatureExtractor.from_pretrained( model_id, local_files_only=True, ) model = AutoModelForAudioClassification.from_pretrained( model_id, local_files_only=True, use_safetensors=False, ) finally: if previous_offline is None: os.environ.pop("HF_HUB_OFFLINE", None) else: os.environ["HF_HUB_OFFLINE"] = previous_offline except Exception: if not allow_model_downloads(): raise feature_extractor = AutoFeatureExtractor.from_pretrained(model_id) model = AutoModelForAudioClassification.from_pretrained( model_id, use_safetensors=False, ) model.eval() MODEL_CACHE[cache_key] = (feature_extractor, model, torch) feature_extractor, model, torch = MODEL_CACHE[cache_key] model_sample_rate = int(getattr(feature_extractor, "sampling_rate", TARGET_SAMPLE_RATE)) audio_16k = resample_audio(audio_data, sample_rate, model_sample_rate) segments = build_inference_segments(audio_16k, model_sample_rate) segment_audio = [segment for segment, _ in segments] or [audio_16k] segment_weights = np.asarray([weight for _, weight in segments] or [1.0], dtype=np.float32) segment_weights = segment_weights / max(float(np.sum(segment_weights)), 1e-8) inputs = feature_extractor( segment_audio, sampling_rate=model_sample_rate, return_tensors="pt", padding=True, ) with torch.no_grad(): logits = model(**inputs).logits.detach().cpu().numpy() probabilities = softmax(logits) if probabilities.ndim == 2: probabilities = np.average(probabilities, axis=0, weights=segment_weights) id2label = getattr(model.config, "id2label", {}) or {} predictions = [] for index, probability in enumerate(probabilities): label = str(id2label.get(index, f"LABEL_{index}")) predictions.append({"label": label, "score": float(probability)}) normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) source = "transformers_assets" if os.path.isdir(model_id) else "transformers" update_model_status("wav2vec", source, model_id) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": source, "model": model_id, } except Exception as error: update_model_status("wav2vec", "fallback", f"{model_id}: {error}") raise def get_cnn_pipeline_model_id() -> str: """Return the HuggingFace model ID to use for the CNN pipeline slot.""" return ( get_env("CNN_PIPELINE_MODEL_ID", default=DEFAULT_CNN_PIPELINE_MODEL_ID) or DEFAULT_CNN_PIPELINE_MODEL_ID ) def _hf_hub_reachable() -> bool: """Check HuggingFace Hub connectivity. Result is NOT cached — each call does a fresh check so stale True results don't cause later download attempts.""" try: import urllib.request urllib.request.urlopen("https://huggingface.co", timeout=2) return True except Exception: return False def _silence_hf_loggers() -> list: """Set huggingface/transformers loggers to ERROR and return previous levels.""" noisy = ["huggingface_hub", "transformers", "transformers.modeling_utils", "transformers.configuration_utils", "filelock"] saved = [] for name in noisy: lg = logging.getLogger(name) saved.append((lg, lg.level)) lg.setLevel(logging.ERROR) return saved def _restore_hf_loggers(saved: list) -> None: for lg, level in saved: lg.setLevel(level) def run_cnn_pipeline_model( audio_data: np.ndarray, sample_rate: int, ) -> Dict[str, Any]: """ Run the CNN model slot using a HuggingFace audio-classification pipeline. Default model: ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition Labels: angry, calm, disgust, fearful, happy, neutral, sad, surprised """ model_id = get_cnn_pipeline_model_id() cache_key = f"cnn_pipeline:{model_id}" saved_levels = _silence_hf_loggers() try: if cache_key not in MODEL_CACHE: from transformers import pipeline as hf_pipeline # Try local cache first (no network) loaded = False try: pipe = hf_pipeline( "audio-classification", model=model_id, top_k=None, local_files_only=True, ) loaded = True except Exception: pass if not loaded: # Only attempt download if HF is actually reachable right now if not allow_model_downloads() or not _hf_hub_reachable(): raise RuntimeError( f"{model_id!r} not in local cache and HF Hub not reachable" ) logger.info("[cnn_pipeline] Downloading model %s …", model_id) pipe = hf_pipeline( "audio-classification", model=model_id, top_k=None, ) MODEL_CACHE[cache_key] = pipe pipe = MODEL_CACHE[cache_key] audio_16k = resample_audio(audio_data, sample_rate, TARGET_SAMPLE_RATE) outputs = pipe({"array": audio_16k, "sampling_rate": TARGET_SAMPLE_RATE}) predictions = [ {"label": item["label"], "score": float(item["score"])} for item in (outputs or []) ] normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) update_model_status("cnn", "pipeline", model_id) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": "pipeline", "model": model_id, } except Exception as error: update_model_status("cnn", "pipeline_failed", f"{model_id}: {type(error).__name__}") raise finally: _restore_hf_loggers(saved_levels) def write_temp_wav(audio_data: np.ndarray, sample_rate: int) -> str: audio_16k = resample_audio(audio_data, sample_rate, TARGET_SAMPLE_RATE) audio_16k = np.clip(audio_16k, -1.0, 1.0) temp_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) temp_path = temp_file.name temp_file.close() with wave.open(temp_path, "wb") as wav_writer: wav_writer.setnchannels(1) wav_writer.setsampwidth(2) wav_writer.setframerate(TARGET_SAMPLE_RATE) wav_writer.writeframes((audio_16k * 32767).astype(np.int16).tobytes()) return temp_path def load_cnn_predict_module() -> tuple[Any, str]: repo_id = get_cnn_model_repo() cache_key = f"cnn:{repo_id}" if cache_key in MODEL_CACHE: return MODEL_CACHE[cache_key] from huggingface_hub import snapshot_download repo_path = snapshot_download(repo_id=repo_id, local_files_only=not allow_model_downloads()) outputs_dir = os.path.join(repo_path, "outputs") predict_path = os.path.join(outputs_dir, "predict.py") if not os.path.exists(predict_path): predict_path = os.path.join(repo_path, "predict.py") outputs_dir = repo_path if not os.path.exists(predict_path): raise FileNotFoundError(f"Could not find predict.py in {repo_id}") module_name = "cnn_ser_predict" spec = importlib.util.spec_from_file_location(module_name, predict_path) if spec is None or spec.loader is None: raise ImportError(f"Could not import CNN predictor from {predict_path}") previous_path = list(sys.path) sys.path.insert(0, os.path.dirname(predict_path)) try: module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) finally: sys.path = previous_path MODEL_CACHE[cache_key] = (module, outputs_dir) return module, outputs_dir def load_pytorch_cnn_model() -> tuple[Any, Any, Any]: model_source = get_cnn_model_repo() cache_key = f"pytorch_cnn:{model_source}" if cache_key in MODEL_CACHE: return MODEL_CACHE[cache_key] import torch if os.path.isdir(model_source): model_file = os.path.join(model_source, "emotion_cnn_v2.py") weights_file = os.path.join(model_source, "best_model_v2.pth") if not os.path.exists(model_file) or not os.path.exists(weights_file): raise FileNotFoundError(f"CNN assets missing in {model_source}") else: from huggingface_hub import hf_hub_download try: model_file = hf_hub_download( repo_id=model_source, filename="emotion_cnn_v2.py", local_files_only=True, ) weights_file = hf_hub_download( repo_id=model_source, filename="best_model_v2.pth", local_files_only=True, ) except Exception: if not allow_model_downloads(): raise model_file = hf_hub_download(repo_id=model_source, filename="emotion_cnn_v2.py") weights_file = hf_hub_download(repo_id=model_source, filename="best_model_v2.pth") module_name = "pytorch_emotion_cnn_v2" spec = importlib.util.spec_from_file_location(module_name, model_file) if spec is None or spec.loader is None: raise ImportError(f"Could not import PyTorch CNN from {model_file}") module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) model = module.ImprovedEmotionCNN(num_classes=8) checkpoint = torch.load(weights_file, map_location="cpu") state_dict = checkpoint.get("model_state_dict", checkpoint) if isinstance(checkpoint, dict) else checkpoint model.load_state_dict(state_dict) model.eval() MODEL_CACHE[cache_key] = (module, model, torch) return module, model, torch def pad_or_trim_time(features: np.ndarray, target_frames: int = 128) -> np.ndarray: if features.shape[1] < target_frames: return np.pad(features, ((0, 0), (0, target_frames - features.shape[1])), mode="constant") return features[:, :target_frames] def extract_pytorch_cnn_features(audio_data: np.ndarray, sample_rate: int) -> np.ndarray: audio = resample_audio(audio_data, sample_rate, TARGET_SAMPLE_RATE) audio = np.asarray(audio, dtype=np.float32) if len(audio) == 0: audio = np.zeros(TARGET_SAMPLE_RATE, dtype=np.float32) n_fft = 2048 hop_length = 512 frames = frame_audio(audio, n_fft, hop_length) window = np.hanning(n_fft).astype(np.float32) spectrum = np.fft.rfft(frames * window, axis=1) power = np.square(np.abs(spectrum)).T + 1e-10 freqs = np.fft.rfftfreq(n_fft, d=1.0 / TARGET_SAMPLE_RATE) mel = resize_frequency_axis(power, 128) mel_db = 10.0 * np.log10(mel / np.max(mel)) mfcc = dct_type_2(mel_db)[:13] delta = np.gradient(mfcc, axis=1) delta2 = np.gradient(delta, axis=1) chroma = compute_chroma(power, freqs) contrast = compute_spectral_contrast(power, freqs) tonnetz = np.zeros((6, mel_db.shape[1]), dtype=np.float32) signs = np.signbit(frames) zcr = np.mean(np.diff(signs, axis=1), axis=1, keepdims=True).T magnitude = np.sqrt(power) magnitude_sum = np.sum(magnitude, axis=0) + 1e-8 centroid_values = np.sum(magnitude * freqs[:, None], axis=0) / magnitude_sum cumulative = np.cumsum(magnitude, axis=0) rolloff_indices = np.argmax(cumulative >= (0.85 * magnitude_sum[None, :]), axis=0) rolloff_values = freqs[rolloff_indices] bandwidth_values = np.sqrt( np.sum(magnitude * np.square(freqs[:, None] - centroid_values[None, :]), axis=0) / magnitude_sum ) centroid = centroid_values[None, :] rolloff = rolloff_values[None, :] bandwidth = bandwidth_values[None, :] min_frames = min(feature.shape[1] for feature in [mel_db, mfcc, delta, delta2, chroma, contrast, tonnetz, zcr, centroid, rolloff, bandwidth]) stacked = np.vstack( [ mel_db[:, :min_frames], mfcc[:, :min_frames], delta[:, :min_frames], delta2[:, :min_frames], chroma[:, :min_frames], contrast[:, :min_frames], tonnetz[:, :min_frames], zcr[:, :min_frames], centroid[:, :min_frames], rolloff[:, :min_frames], bandwidth[:, :min_frames], ] ) mean = stacked.mean(axis=1, keepdims=True) std = stacked.std(axis=1, keepdims=True) normalized = (stacked - mean) / (std + 1e-8) return pad_or_trim_time(normalized.astype(np.float32), 128) def resize_frequency_axis(power: np.ndarray, target_bins: int) -> np.ndarray: source_positions = np.linspace(0.0, 1.0, power.shape[0]) target_positions = np.linspace(0.0, 1.0, target_bins) resized = np.empty((target_bins, power.shape[1]), dtype=np.float32) for frame_index in range(power.shape[1]): resized[:, frame_index] = np.interp(target_positions, source_positions, power[:, frame_index]) return resized + 1e-10 def dct_type_2(values: np.ndarray) -> np.ndarray: try: from scipy.fftpack import dct return dct(values, type=2, axis=0, norm="ortho") except Exception: rows = values.shape[0] indices = np.arange(rows) basis = np.cos(np.pi / rows * (indices + 0.5)[:, None] * indices[None, :]) return basis.T @ values def compute_chroma(power: np.ndarray, freqs: np.ndarray) -> np.ndarray: chroma = np.zeros((12, power.shape[1]), dtype=np.float32) valid = freqs > 30 midi = np.round(69 + 12 * np.log2(freqs[valid] / 440.0)).astype(int) pitch_classes = np.mod(midi, 12) for pitch_class in range(12): mask = pitch_classes == pitch_class if np.any(mask): chroma[pitch_class] = np.sum(power[valid][mask], axis=0) return np.log1p(chroma) def compute_spectral_contrast(power: np.ndarray, freqs: np.ndarray) -> np.ndarray: bands = [(0, 200), (200, 400), (400, 800), (800, 1600), (1600, 3200), (3200, 6400), (6400, 8000)] contrast = np.zeros((7, power.shape[1]), dtype=np.float32) log_power = np.log1p(power) for index, (low, high) in enumerate(bands): mask = (freqs >= low) & (freqs < high) if not np.any(mask): continue band = log_power[mask] contrast[index] = np.percentile(band, 90, axis=0) - np.percentile(band, 10, axis=0) return contrast def run_pytorch_cnn_model(audio_data: np.ndarray, sample_rate: int) -> Dict[str, Any]: _, model, torch = load_pytorch_cnn_model() features = extract_pytorch_cnn_features(audio_data, sample_rate) features_tensor = torch.from_numpy(features).unsqueeze(0).unsqueeze(0) with torch.no_grad(): logits = model(features_tensor) probabilities = torch.softmax(logits, dim=1)[0].detach().cpu().numpy() labels = ["neutral", "calm", "happy", "sad", "angry", "fearful", "disgust", "surprised"] predictions = [ {"label": labels[index], "score": float(probabilities[index])} for index in range(min(len(labels), len(probabilities))) ] normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) model_source = get_cnn_model_repo() source = "pytorch_cnn_assets" if os.path.isdir(model_source) else "pytorch_cnn" update_model_status("cnn", source, model_source) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": source, "model": model_source, } def run_cnn_standalone_model(module: Any, model_dir: str, temp_path: str) -> Any: import tensorflow as tf import librosa model_path = os.path.join(model_dir, "model1_cnn_bilstm_attn.keras") if not os.path.exists(model_path): raise FileNotFoundError(f"Standalone CNN model not found: {model_path}") model_cache_key = f"keras:{model_path}" if model_cache_key not in MODEL_CACHE: MODEL_CACHE[model_cache_key] = tf.keras.models.load_model(model_path, compile=False) model = MODEL_CACHE[model_cache_key] wav, _ = librosa.load(temp_path, sr=TARGET_SAMPLE_RATE) spec = module.extract_features(wav)[np.newaxis] probabilities = model.predict(spec, verbose=0)[0] labels = getattr(module, "EMOTION_LABELS", ["angry", "disgust", "fear", "happy", "neutral", "sad"]) return { "probabilities": { labels[index]: float(probabilities[index]) for index in range(min(len(labels), len(probabilities))) } } def normalize_cnn_output(output: Any) -> List[Dict[str, Any]]: if isinstance(output, dict): if isinstance(output.get("probabilities"), dict): return [ {"label": label, "score": float(score) / 100.0 if float(score) > 1.0 else float(score)} for label, score in output["probabilities"].items() ] if isinstance(output.get("probs"), dict): return [ {"label": label, "score": float(score) / 100.0 if float(score) > 1.0 else float(score)} for label, score in output["probs"].items() ] if output.get("predicted_emotion") or output.get("emotion"): label = output.get("predicted_emotion") or output.get("emotion") score = output.get("confidence", output.get("score", 1.0)) return [{"label": str(label), "score": float(score)}] if isinstance(output, tuple) and len(output) >= 2: label = output[0] confidence = output[1] predictions = [{"label": str(label), "score": float(confidence)}] if len(output) >= 3 and isinstance(output[2], dict): predictions = [ {"label": str(pred_label), "score": float(score) / 100.0 if float(score) > 1.0 else float(score)} for pred_label, score in output[2].items() ] return predictions return parse_model_response(output) def run_cnn_repository_model( audio_data: np.ndarray, sample_rate: int, ) -> Dict[str, Any]: repo_id = get_cnn_model_repo() if repo_id == "saadmannan/speech-emotion-recognition" or os.path.isdir(repo_id): return run_pytorch_cnn_model(audio_data, sample_rate) temp_path = "" try: module, model_dir = load_cnn_predict_module() if not hasattr(module, "predict_emotion"): raise AttributeError("CNN repo predict.py does not expose predict_emotion") temp_path = write_temp_wav(audio_data, sample_rate) output_buffer = io.StringIO() use_fusion = (get_env("CNN_USE_FUSION", default="0") or "0").lower() in {"1", "true", "yes"} with contextlib.redirect_stdout(output_buffer), contextlib.redirect_stderr(output_buffer): if use_fusion: try: output = module.predict_emotion(temp_path, model_dir) except TypeError: output = module.predict_emotion(temp_path) else: output = run_cnn_standalone_model(module, model_dir, temp_path) predictions = normalize_cnn_output(output) normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) update_model_status("cnn", "huggingface_repo", repo_id) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": "huggingface_repo", "model": repo_id, } except Exception as error: update_model_status("cnn", "fallback", f"{repo_id}: {error}") raise finally: if temp_path and os.path.exists(temp_path): try: os.unlink(temp_path) except OSError: pass def triangular(value: float, center: float, width: float) -> float: if width <= 0: return 0.0 return clamp(1.0 - abs(value - center) / width) def extract_acoustic_features(audio_data: np.ndarray, sample_rate: int) -> Dict[str, float]: audio = np.asarray(audio_data, dtype=np.float32) if audio.ndim > 1: audio = np.mean(audio, axis=1) audio = audio[np.isfinite(audio)] if len(audio) == 0: return {"duration": 0.0, "quality": 0.0} audio = audio - float(np.mean(audio)) duration = len(audio) / sample_rate peak = float(np.max(np.abs(audio))) if len(audio) else 0.0 rms_absolute = float(np.sqrt(np.mean(np.square(audio)))) if len(audio) else 0.0 if peak > 1.0: audio = audio / peak peak = 1.0 normalized = audio / max(peak, 0.05) frame_length = min(2048, max(512, int(sample_rate * 0.046))) hop_length = max(128, frame_length // 4) frames = frame_audio(normalized, frame_length, hop_length) absolute_frames = frame_audio(audio, frame_length, hop_length) window = np.hanning(frame_length).astype(np.float32) windowed = frames * window rms = np.sqrt(np.mean(np.square(frames), axis=1)) absolute_rms = np.sqrt(np.mean(np.square(absolute_frames), axis=1)) zcr = np.mean(np.diff(np.signbit(frames), axis=1), axis=1) magnitude = np.abs(np.fft.rfft(windowed, axis=1)) + 1e-8 freqs = np.fft.rfftfreq(frame_length, d=1.0 / sample_rate) magnitude_sum = np.sum(magnitude, axis=1) centroid = np.sum(magnitude * freqs, axis=1) / magnitude_sum bandwidth = np.sqrt( np.sum(magnitude * np.square(freqs[None, :] - centroid[:, None]), axis=1) / magnitude_sum ) cumulative = np.cumsum(magnitude, axis=1) rolloff_indices = np.argmax(cumulative >= (0.85 * magnitude_sum[:, None]), axis=1) rolloff = freqs[rolloff_indices] active_threshold = max(0.03, float(np.percentile(rms, 60)) * 0.35) active_mask = rms > active_threshold active_rms = rms[active_mask] if np.any(active_mask) else rms active_absolute_rms = absolute_rms[active_mask] if np.any(active_mask) else absolute_rms valid_pitch = estimate_pitch_track( normalized, sample_rate, frame_length, hop_length, active_mask, ) pitch_median = float(np.median(valid_pitch)) if len(valid_pitch) else 0.0 pitch_spread = ( float(np.percentile(valid_pitch, 90) - np.percentile(valid_pitch, 10)) if len(valid_pitch) >= 4 else 0.0 ) pitch_std = float(np.std(valid_pitch)) if len(valid_pitch) else 0.0 active_ratio = float(np.mean(active_mask)) if len(active_mask) else 0.0 pause_ratio = 1.0 - active_ratio quality = clamp(scale_feature(rms_absolute, 0.003, 0.035) * scale_feature(duration, 1.0, 4.0)) return { "duration": duration, "quality": quality, "peak": peak, "rms_absolute": rms_absolute, "energy": float(np.mean(active_absolute_rms)) if len(active_absolute_rms) else 0.0, "energy_var": float(np.std(active_rms)) if len(active_rms) else 0.0, "zcr": float(np.mean(zcr[active_mask])) if np.any(active_mask) else float(np.mean(zcr)), "centroid": float(np.mean(centroid[active_mask])) if np.any(active_mask) else float(np.mean(centroid)), "bandwidth": float(np.mean(bandwidth[active_mask])) if np.any(active_mask) else float(np.mean(bandwidth)), "rolloff": float(np.mean(rolloff[active_mask])) if np.any(active_mask) else float(np.mean(rolloff)), "pitch_median": pitch_median, "pitch_spread": pitch_spread, "pitch_std": pitch_std, "voiced_ratio": clamp(len(valid_pitch) / max(1, len(rms))), "pause_ratio": clamp(pause_ratio), } def frame_audio(audio: np.ndarray, frame_length: int, hop_length: int) -> np.ndarray: if len(audio) < frame_length: audio = np.pad(audio, (0, frame_length - len(audio))) frame_count = 1 + max(0, (len(audio) - frame_length) // hop_length) frames = np.empty((frame_count, frame_length), dtype=np.float32) for index in range(frame_count): start = index * hop_length frames[index] = audio[start : start + frame_length] return frames def estimate_pitch_track( audio: np.ndarray, sample_rate: int, frame_length: int, hop_length: int, active_mask: np.ndarray, ) -> np.ndarray: min_lag = max(1, int(sample_rate / 500)) max_lag = min(frame_length - 1, int(sample_rate / 65)) if max_lag <= min_lag: return np.asarray([], dtype=np.float32) active_indices = np.flatnonzero(active_mask) if len(active_indices) > 80: active_indices = active_indices[np.linspace(0, len(active_indices) - 1, 80).astype(int)] pitches: List[float] = [] window = np.hanning(frame_length).astype(np.float32) for frame_index in active_indices: start = frame_index * hop_length frame = audio[start : start + frame_length] if len(frame) < frame_length: frame = np.pad(frame, (0, frame_length - len(frame))) frame = (frame - float(np.mean(frame))) * window if float(np.sqrt(np.mean(np.square(frame)))) < 0.015: continue autocorr = np.correlate(frame, frame, mode="full")[frame_length - 1 :] if autocorr[0] <= 1e-8: continue search = autocorr[min_lag:max_lag] lag = int(np.argmax(search)) + min_lag strength = float(autocorr[lag] / autocorr[0]) if strength >= 0.28: pitches.append(sample_rate / lag) return np.asarray(pitches, dtype=np.float32) def score_emotions_from_features( features: Dict[str, float], model_name: str, ) -> Dict[str, float]: """ Heuristic acoustic scoring for all 6 emotions: anger, disgust, fear, happy, neutral, sad. Design principle: every emotion's raw score must be on the same ~0..1 scale so that neutral cannot systematically dominate just because its coefficient happened to be 0.70 while anger's peak was 0.32. Acoustic cue intuitions: - anger: high energy, high ZCR, high rolloff, low pause, voiced - disgust: low-mid energy, low pitch, flat pitch, mid-high ZCR - fear: mid-low energy, high pitch, wide pitch-spread, high ZCR - happy: high pitch-spread, higher pitch, high energy variability, low pause - neutral: stable mid-range energy+centroid, low pitch-spread, low pause - sad: high pause ratio, low energy, low pitch, narrow pitch-spread """ duration = features.get("duration", 0.0) if duration <= 0.0 or features.get("quality", 0.0) < 0.10: # Return a genuinely uncertain prior — no strong neutral bias equal = 1.0 / len(EMOTIONS) return {e: equal for e in EMOTIONS} energy = scale_feature(features["energy"], 0.012, 0.20) energy_var = scale_feature(features["energy_var"], 0.008, 0.18) zcr = scale_feature(features["zcr"], 0.020, 0.20) centroid = scale_feature(features["centroid"], 600, 3800) bandwidth = scale_feature(features["bandwidth"], 600, 3800) rolloff = scale_feature(features["rolloff"], 1200, 7000) pitch = scale_feature(features["pitch_median"], 85, 280) pitch_spread= scale_feature(features["pitch_spread"], 20, 200) voiced = clamp(features["voiced_ratio"]) pause = clamp(features["pause_ratio"]) # Stable signal: mid-range energy + mid-range centroid + low variability # Capped at 0.80 to keep it competitive with other emotions stable = clamp( 0.35 * triangular(energy, 0.42, 0.40) + 0.28 * triangular(centroid, 0.40, 0.40) + 0.20 * (1.0 - energy_var) + 0.17 * (1.0 - pitch_spread) ) # ------------------------------------------------------------------ # # Both CNN and wav2vec use the same calibrated heuristic when running # # as a fallback — the old CNN-specific weights were the neutral-bias # # culprit (neutral=0.70, anger max=0.32). # # All scores now peak near 1.0 under their ideal acoustic conditions. # # ------------------------------------------------------------------ # scores = { # anger: needs high energy AND at least 2 other cues "anger": clamp( 0.35 * energy + 0.22 * rolloff + 0.18 * bandwidth + 0.15 * zcr + 0.10 * (1.0 - pause) ), # disgust: quiet, low-pitch, unvarying "disgust": clamp( 0.32 * (1.0 - pitch) + 0.26 * (1.0 - energy_var) + 0.24 * triangular(energy, 0.28, 0.26) + 0.18 * (1.0 - voiced * 0.5) ), # fear: elevated pitch + spread + ZCR, low-ish energy "fear": clamp( 0.30 * pitch + 0.28 * pitch_spread + 0.22 * zcr + 0.12 * (1.0 - energy) + 0.08 * (1.0 - pause) ), # happy: expressive pitch movement + energy variability "happy": clamp( 0.32 * pitch_spread + 0.28 * pitch + 0.22 * energy_var + 0.10 * voiced + 0.08 * (1.0 - pause) ), # neutral: stable but no longer over-weighted "neutral": clamp( 0.40 * stable + 0.25 * voiced + 0.20 * (1.0 - pause) + 0.15 * (1.0 - energy_var) ), # sad: pauses + low energy + low pitch "sad": clamp( 0.35 * pause + 0.28 * (1.0 - energy) + 0.20 * (1.0 - pitch) + 0.17 * (1.0 - pitch_spread) ), } return scores def generate_local_predictions( audio_data: np.ndarray, sample_rate: int, model_name: str, features: Dict[str, float] | None = None, ) -> List[Dict[str, float | str]]: if features is None: features = extract_acoustic_features(audio_data, sample_rate) scores = score_emotions_from_features(features, model_name) normalized = normalize_distribution(scores) return sorted( [ {"label": label, "score": score} for label, score in normalized.items() ], key=lambda item: float(item["score"]), reverse=True, ) def run_local_model( model_name: str, audio_data: np.ndarray, sample_rate: int, features: Dict[str, float] | None = None, ) -> Dict[str, Any]: predictions = generate_local_predictions(audio_data, sample_rate, model_name, features) normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": "local", } def predictions_to_scores(predictions: List[Dict[str, Any]]) -> Dict[str, float]: return normalize_emotion_prediction(predictions) def scores_from_result(result: Dict[str, Any]) -> Dict[str, float]: """Extract per-emotion probability distribution from a model result.""" scores = predictions_to_scores(result.get("raw_predictions", [])) # Ensure all 6 emotions present for emotion in EMOTIONS: scores.setdefault(emotion, 0.0) if not any(scores.values()): scores[result.get("emotion", "neutral")] = float(result.get("confidence", 0.0)) # Normalise so they sum to ~1 total = sum(scores.values()) or 1.0 return {e: round(scores[e] / total, 4) for e in EMOTIONS} def combine_model_results( cnn_result: Dict[str, Any], wav2vec_result: Dict[str, Any], features: Dict[str, float], ) -> tuple[str, float, bool, Dict[str, float], Dict[str, float], Dict[str, float]]: cnn_scores = scores_from_result(cnn_result) wav2vec_scores = scores_from_result(wav2vec_result) def margin(scores: Dict[str, float]) -> float: ranked = sorted(scores.values(), reverse=True) top = ranked[0] if ranked else 0.0 runner_up = ranked[1] if len(ranked) > 1 else 0.0 return top - runner_up wav_source = wav2vec_result.get("source", "") cnn_source = cnn_result.get("source", "") wav_reliability = 1.0 if wav_source in {"transformers", "transformers_assets"} else 0.45 cnn_reliability = 0.8 if cnn_source in {"pipeline", "huggingface_repo", "pytorch_cnn", "pytorch_cnn_assets"} else 0.3 wav_strength = wav_reliability * clamp( 0.45 + 0.55 * float(wav2vec_result.get("confidence", 0.0)) + (margin(wav2vec_scores) * 1.2) ) cnn_strength = cnn_reliability * clamp( 0.35 + 0.65 * float(cnn_result.get("confidence", 0.0)) + (margin(cnn_scores) * 1.1) ) total_strength = max(wav_strength + cnn_strength, 1e-8) wav2vec_weight = wav_strength / total_strength cnn_weight = cnn_strength / total_strength if wav_reliability >= 1.0 and cnn_reliability < 0.5: wav2vec_weight = max(wav2vec_weight, 0.75) cnn_weight = 1.0 - wav2vec_weight combined = normalize_distribution( { emotion: (cnn_weight * cnn_scores[emotion]) + (wav2vec_weight * wav2vec_scores[emotion]) for emotion in EMOTIONS } ) ranked = sorted(combined.items(), key=lambda item: item[1], reverse=True) final_emotion, top_score = ranked[0] runner_up = ranked[1][1] if len(ranked) > 1 else 0.0 margin = top_score - runner_up quality = features.get("quality", 0.0) agreement = cnn_result.get("emotion") == wav2vec_result.get("emotion") confidence = clamp(0.45 + (margin * 1.35) + (top_score - 0.25) * 0.65) if agreement: confidence += 0.06 confidence *= clamp(0.55 + 0.45 * quality) uncertain = quality < 0.20 or margin < 0.030 or confidence < 0.50 return final_emotion, round(confidence, 2), uncertain, combined, cnn_scores, wav2vec_scores def parse_model_response(payload: Any) -> List[Dict[str, Any]]: if isinstance(payload, list): if payload and isinstance(payload[0], list): return payload[0] return payload if isinstance(payload, dict): if isinstance(payload.get("predictions"), list): return payload["predictions"] if isinstance(payload.get("outputs"), list): return payload["outputs"] if isinstance(payload.get("data"), list): return payload["data"] return [] def is_internal_local_endpoint(endpoint_url: str, model_name: str) -> bool: parsed = urlparse(endpoint_url) return ( parsed.scheme in {"http", "https"} and parsed.hostname in LOCAL_HOSTS and parsed.path.rstrip("/") == f"/models/{model_name}/predict" ) def call_model_endpoint( model_name: str, endpoint_url: str, audio_data: np.ndarray, sample_rate: int, features: Dict[str, float] | None = None, ) -> Dict[str, Any]: if is_internal_local_endpoint(endpoint_url, model_name): if model_name == "wav2vec": try: return run_wav2vec_transformer_model(audio_data, sample_rate) except Exception as error: logger.exception("[wav2vec] Transformer model failed; using heuristic fallback") fallback = run_local_model("wav2vec", audio_data, sample_rate, features) fallback["error"] = f"wav2vec model unavailable: {error}" fallback["source"] = "heuristic_fallback" return fallback if model_name == "cnn": # CNN is display-only — failures are non-fatal, logged at INFO only. # Silence the noisy HuggingFace connectivity warning from transformers. _hf_logger = logging.getLogger("huggingface_hub") _prev_hf_level = _hf_logger.level _hf_logger.setLevel(logging.ERROR) # 1) Try the HuggingFace audio-classification pipeline try: result = run_cnn_pipeline_model(audio_data, sample_rate) return result except Exception: pass finally: _hf_logger.setLevel(_prev_hf_level) # 2) Fall back to the custom PyTorch CNN repository try: return run_cnn_repository_model(audio_data, sample_rate) except Exception as repo_err: logger.info( "[cnn] Repository model unavailable (%s); using heuristic fallback", type(repo_err).__name__, ) fallback = run_local_model("cnn", audio_data, sample_rate, features) fallback["error"] = f"CNN models unavailable: {repo_err}" fallback["source"] = "heuristic_fallback" return fallback try: headers = {"Content-Type": "application/json"} api_token = get_env("MODEL_API_TOKEN") if api_token: headers["Authorization"] = f"Bearer {api_token}" payload = { "inputs": { "audio": audio_data.tolist(), "sample_rate": sample_rate, } } response = requests.post(endpoint_url, json=payload, headers=headers, timeout=30) if response.status_code != 200: logger.warning( "[%s] Model call failed with status %s: %s", model_name, response.status_code, response.text, ) return { "emotion": "neutral", "confidence": 0.25, "error": f"API call failed: {response.status_code}", "source": "remote", } predictions = parse_model_response(response.json()) normalized = normalize_emotion_prediction(predictions) emotion, confidence = get_highest_confidence_emotion(normalized) return { "emotion": emotion, "confidence": confidence, "raw_predictions": predictions, "source": "remote", } except Exception as error: logger.exception("[%s] Error calling model", model_name) return { "emotion": "neutral", "confidence": 0.25, "error": str(error), "source": "remote", } def decode_audio_base64(audio_base64: str) -> tuple[np.ndarray, int]: audio_bytes = base64.b64decode(audio_base64) wav_buffer = io.BytesIO(audio_bytes) with wave.open(wav_buffer, "rb") as wav_reader: sample_rate = wav_reader.getframerate() num_frames = wav_reader.getnframes() num_channels = wav_reader.getnchannels() audio_data = np.frombuffer( wav_reader.readframes(num_frames), dtype=np.int16, ) if num_channels > 1: audio_data = audio_data.reshape(-1, num_channels).mean(axis=1) return audio_data.astype(np.float32) / 32768.0, sample_rate def extract_audio_from_model_payload(data: Dict[str, Any]) -> tuple[np.ndarray, int]: if not isinstance(data, dict): raise ValueError("Request body must be a JSON object.") inputs = data.get("inputs", data) if not isinstance(inputs, dict): raise ValueError("Request is missing an inputs object.") audio_values = inputs.get("audio") sample_rate = inputs.get("sample_rate") if not isinstance(audio_values, list) or not audio_values: raise ValueError("Audio samples are missing.") if not isinstance(sample_rate, int): raise ValueError("Sample rate is missing or invalid.") return np.asarray(audio_values, dtype=np.float32), sample_rate def analyze_emotion_base64_legacy(audio_base64: str) -> Dict[str, Any]: """ Analyze emotion from base64-encoded WAV audio. Architecture: - wav2vec transformer is the SOLE authority for the final emotion result. - CNN pipeline runs concurrently (best-effort) and its scores are returned for display purposes only — it has ZERO weight on the final decision. - Accepts audio of any duration; no hardcoded length restrictions. """ if not audio_base64: return build_error_response("No audio received", warnings=["No audio provided"]) try: audio_data, sample_rate = decode_audio_base64(audio_base64) except Exception as error: logger.exception("Audio decoding error") return build_error_response( f"Audio processing failed: {error}", warnings=[f"Audio processing failed: {error}"], ) if audio_data is None or len(audio_data) == 0: return build_error_response("Empty audio data", warnings=["Empty audio data"]) duration = len(audio_data) / sample_rate warnings: List[str] = [] features = extract_acoustic_features(audio_data, sample_rate) if features.get("quality", 0.0) < 0.15: warnings.append("Audio signal is weak or mostly silent; prediction may be uncertain") # ── Primary model: wav2vec transformer ────────────────────────────────── wav2vec_endpoint = get_model_endpoint("wav2vec") wav2vec_result = call_model_endpoint( "wav2vec", wav2vec_endpoint, audio_data, sample_rate, features, ) if wav2vec_result.get("error"): warnings.append(f"wav2vec adapter warning: {wav2vec_result['error']}") wav2vec_scores = scores_from_result(wav2vec_result) # ── Split 4-class Wav2Vec outputs into 6 classes using Acoustic Heuristics ── # The default wav2vec model only outputs 4 classes: ang, hap, neu, sad. # We use acoustic cues to distribute these broad buckets into all 6 target emotions. w_ang = wav2vec_scores.get("anger", 0.0) w_hap = wav2vec_scores.get("happy", 0.0) w_sad = wav2vec_scores.get("sad", 0.0) w_neu = wav2vec_scores.get("neutral", 0.0) w_fea = wav2vec_scores.get("fear", 0.0) w_dis = wav2vec_scores.get("disgust", 0.0) # Apply bias correction: The base superb-er model is notoriously biased towards Neutral # on standard laptop microphones. We suppress Neutral and boost the emotional signals # to dramatically increase sensitivity. w_neu *= 0.60 w_ang *= 1.80 w_sad *= 1.50 w_hap *= 1.50 # Re-normalize the base 4 buckets before expansion base_sum = w_ang + w_hap + w_sad + w_neu + 1e-8 w_ang /= base_sum w_hap /= base_sum w_sad /= base_sum w_neu /= base_sum if w_fea < 0.01 and w_dis < 0.01: acoustic = normalize_distribution(score_emotions_from_features(features, "wav2vec")) # We want to keep the probabilities SHARP. # Instead of splitting 50/50 and diluting the score, we identify the dominant # acoustic sub-emotion and transfer the majority of the weight to it. # Anger bucket -> Anger vs Fear final_ang = w_ang final_fea = 0.0 if acoustic.get("fear", 0) > acoustic.get("anger", 0) * 1.15: final_fea = w_ang * 0.85 final_ang = w_ang * 0.15 # Sad bucket -> Sad vs Disgust final_sad = w_sad final_dis = 0.0 if acoustic.get("disgust", 0) > acoustic.get("sad", 0) * 1.15: final_dis = w_sad * 0.85 final_sad = w_sad * 0.15 # Neutral bucket -> Neutral, but can steal a little if it's actually hidden fear/disgust final_neu = w_neu ndf_max = max(acoustic.get("neutral", 0), acoustic.get("disgust", 0), acoustic.get("fear", 0)) if ndf_max == acoustic.get("disgust", 0) and ndf_max > 0.35: final_dis += w_neu * 0.70 final_neu = w_neu * 0.30 elif ndf_max == acoustic.get("fear", 0) and ndf_max > 0.35: final_fea += w_neu * 0.70 final_neu = w_neu * 0.30 wav2vec_scores = { "anger": final_ang, "fear": final_fea, "happy": w_hap, "sad": final_sad, "disgust": final_dis, "neutral": final_neu } wav2vec_scores = normalize_distribution(wav2vec_scores) # Final result is 100 % wav2vec — no CNN blending ranked = sorted(wav2vec_scores.items(), key=lambda x: x[1], reverse=True) top_score = ranked[0][1] if ranked else 0.0 runner_up = ranked[1][1] if len(ranked) > 1 else 0.0 margin = top_score - runner_up quality = features.get("quality", 0.0) final_emotion = ranked[0][0] if ranked else "neutral" final_confidence = round( clamp(0.40 + (margin * 1.50) + (top_score - 0.20) * 0.60) * clamp(0.50 + 0.50 * quality), 2, ) uncertain = quality < 0.18 or margin < 0.025 or final_confidence < 0.45 if uncertain: warnings.append("Emotion cues are close; treat this as a low-confidence estimate") # ── Secondary model: CNN pipeline (display only, best-effort) ─────────── cnn_scores: Dict[str, float] = {e: 0.0 for e in EMOTIONS} cnn_emotion = "n/a" cnn_confidence = 0.0 cnn_source = "pending" cnn_model_id = "n/a" cnn_endpoint = get_model_endpoint("cnn") try: cnn_result = call_model_endpoint( "cnn", cnn_endpoint, audio_data, sample_rate, features, ) cnn_scores = scores_from_result(cnn_result) cnn_emotion = cnn_result.get("emotion", "n/a") cnn_confidence = float(cnn_result.get("confidence", 0.0)) cnn_source = cnn_result.get("source", "unknown") cnn_model_id = cnn_result.get("model", get_cnn_pipeline_model_id()) except Exception as cnn_err: cnn_source = "unavailable" agreement = cnn_emotion not in {"n/a", "unavailable"} and cnn_emotion == wav2vec_result.get("emotion", "n/a") return { "final": { "emotion": final_emotion, "confidence": final_confidence, "uncertain": uncertain, }, "comparison": { "agreement": agreement, "preferredModel": "wav2vec", }, "models": { "cnn": { "emotion": cnn_emotion, "confidence": cnn_confidence, "scores": cnn_scores, "source": cnn_source, "model": cnn_model_id, }, "wav2vec": { "emotion": final_emotion, "confidence": final_confidence, "scores": wav2vec_scores, "source": wav2vec_result.get("source", "unknown"), "model": wav2vec_result.get("model", get_wav2vec_model_id()), }, }, "meta": { "duration": round(duration, 2), "warnings": warnings, "features": { "energy": round(features.get("energy", 0.0), 3), "pitchMedianHz": round(features.get("pitch_median", 0.0), 1), "pitchSpreadHz": round(features.get("pitch_spread", 0.0), 1), "pauseRatio": round(features.get("pause_ratio", 0.0), 3), "quality": round(features.get("quality", 0.0), 3), }, "scores": wav2vec_scores, }, } def analyze_emotion_base64(audio_base64: str) -> Dict[str, Any]: """ Analyze emotion from base64-encoded WAV audio. Architecture: - audio is trimmed and level-normalized before analysis to reduce silence bias; - wav2vec runs over overlapping speech segments for more stable predictions; - the final decision is a confidence-weighted ensemble instead of a single-model guess. """ if not audio_base64: return build_error_response("No audio received", warnings=["No audio provided"]) try: audio_data, sample_rate = decode_audio_base64(audio_base64) except Exception as error: logger.exception("Audio decoding error") return build_error_response( f"Audio processing failed: {error}", warnings=[f"Audio processing failed: {error}"], ) if audio_data is None or len(audio_data) == 0: return build_error_response("Empty audio data", warnings=["Empty audio data"]) raw_duration = len(audio_data) / sample_rate warnings: List[str] = [] processed_audio = preprocess_audio_for_emotion(audio_data, sample_rate) if len(processed_audio) == 0: return build_error_response( "Audio contains no usable signal", warnings=["Audio contains no usable signal"], ) processed_duration = len(processed_audio) / sample_rate if processed_duration < raw_duration * 0.7: warnings.append("Leading and trailing silence were removed before analysis") if processed_duration < 1.0: warnings.append("Very short speech sample detected; emotion accuracy may be limited") features = extract_acoustic_features(processed_audio, sample_rate) if features.get("quality", 0.0) < 0.15: warnings.append("Audio signal is weak or mostly silent; prediction may be uncertain") wav2vec_endpoint = get_model_endpoint("wav2vec") wav2vec_result = call_model_endpoint( "wav2vec", wav2vec_endpoint, processed_audio, sample_rate, features, ) if wav2vec_result.get("error"): warnings.append(f"wav2vec adapter warning: {wav2vec_result['error']}") wav2vec_scores = scores_from_result(wav2vec_result) w_ang = wav2vec_scores.get("anger", 0.0) w_hap = wav2vec_scores.get("happy", 0.0) w_sad = wav2vec_scores.get("sad", 0.0) w_neu = wav2vec_scores.get("neutral", 0.0) w_fea = wav2vec_scores.get("fear", 0.0) w_dis = wav2vec_scores.get("disgust", 0.0) w_neu *= 0.60 w_ang *= 1.80 w_sad *= 1.50 w_hap *= 1.50 base_sum = w_ang + w_hap + w_sad + w_neu + 1e-8 w_ang /= base_sum w_hap /= base_sum w_sad /= base_sum w_neu /= base_sum if w_fea < 0.01 and w_dis < 0.01: acoustic = normalize_distribution(score_emotions_from_features(features, "wav2vec")) final_ang = w_ang final_fea = 0.0 if acoustic.get("fear", 0.0) > acoustic.get("anger", 0.0) * 1.15: final_fea = w_ang * 0.85 final_ang = w_ang * 0.15 final_sad = w_sad final_dis = 0.0 if acoustic.get("disgust", 0.0) > acoustic.get("sad", 0.0) * 1.15: final_dis = w_sad * 0.85 final_sad = w_sad * 0.15 final_neu = w_neu ndf_max = max(acoustic.get("neutral", 0.0), acoustic.get("disgust", 0.0), acoustic.get("fear", 0.0)) if ndf_max == acoustic.get("disgust", 0.0) and ndf_max > 0.35: final_dis += w_neu * 0.70 final_neu = w_neu * 0.30 elif ndf_max == acoustic.get("fear", 0.0) and ndf_max > 0.35: final_fea += w_neu * 0.70 final_neu = w_neu * 0.30 wav2vec_scores = normalize_distribution( { "anger": final_ang, "fear": final_fea, "happy": w_hap, "sad": final_sad, "disgust": final_dis, "neutral": final_neu, } ) wav2vec_emotion = max(wav2vec_scores.items(), key=lambda item: item[1])[0] if wav2vec_scores else "neutral" wav2vec_confidence = max(wav2vec_scores.values()) if wav2vec_scores else 0.0 ensemble_wav2vec_result = { **wav2vec_result, "emotion": wav2vec_emotion, "confidence": wav2vec_confidence, "raw_predictions": [{"label": emotion, "score": score} for emotion, score in wav2vec_scores.items()], } cnn_scores: Dict[str, float] = {e: 0.0 for e in EMOTIONS} cnn_emotion = "n/a" cnn_confidence = 0.0 cnn_source = "pending" cnn_model_id = "n/a" cnn_result: Dict[str, Any] = { "emotion": "neutral", "confidence": 0.0, "raw_predictions": [], "source": "unavailable", } cnn_endpoint = get_model_endpoint("cnn") try: cnn_result = call_model_endpoint( "cnn", cnn_endpoint, processed_audio, sample_rate, features, ) cnn_scores = scores_from_result(cnn_result) cnn_emotion = cnn_result.get("emotion", "n/a") cnn_confidence = float(cnn_result.get("confidence", 0.0)) cnn_source = cnn_result.get("source", "unknown") cnn_model_id = cnn_result.get("model", get_cnn_pipeline_model_id()) except Exception as cnn_err: cnn_source = "unavailable" warnings.append(f"CNN adapter warning: {cnn_err}") final_emotion, final_confidence, uncertain, combined_scores, cnn_scores, wav2vec_scores = combine_model_results( cnn_result, ensemble_wav2vec_result, features, ) if uncertain: warnings.append("Emotion cues are close; treat this as a low-confidence estimate") agreement = cnn_emotion not in {"n/a", "unavailable"} and cnn_emotion == final_emotion preferred_model = ( "ensemble" if cnn_source not in {"pending", "unavailable", "heuristic_fallback", "local"} else "wav2vec" ) return { "final": { "emotion": final_emotion, "confidence": final_confidence, "uncertain": uncertain, }, "comparison": { "agreement": agreement, "preferredModel": preferred_model, }, "models": { "cnn": { "emotion": cnn_emotion, "confidence": cnn_confidence, "scores": cnn_scores, "source": cnn_source, "model": cnn_model_id, }, "wav2vec": { "emotion": wav2vec_emotion, "confidence": round(wav2vec_confidence, 2), "scores": wav2vec_scores, "source": wav2vec_result.get("source", "unknown"), "model": wav2vec_result.get("model", get_wav2vec_model_id()), }, }, "meta": { "duration": round(raw_duration, 2), "processedDuration": round(processed_duration, 2), "warnings": warnings, "features": { "energy": round(features.get("energy", 0.0), 3), "pitchMedianHz": round(features.get("pitch_median", 0.0), 1), "pitchSpreadHz": round(features.get("pitch_spread", 0.0), 1), "pauseRatio": round(features.get("pause_ratio", 0.0), 3), "quality": round(features.get("quality", 0.0), 3), }, "scores": combined_scores, }, } @app.route("/models/cnn/predict", methods=["POST"]) def predict_local_cnn(): try: audio_data, sample_rate = extract_audio_from_model_payload(request.get_json()) result = run_local_model("cnn", audio_data, sample_rate) return jsonify({"predictions": result["raw_predictions"]}) except Exception as error: return jsonify({"error": str(error)}), 400 @app.route("/models/wav2vec/predict", methods=["POST"]) def predict_local_wav2vec(): try: audio_data, sample_rate = extract_audio_from_model_payload(request.get_json()) result = run_local_model("wav2vec", audio_data, sample_rate) return jsonify({"predictions": result["raw_predictions"]}) except Exception as error: return jsonify({"error": str(error)}), 400 @app.route("/api/predict", methods=["POST"]) def predict(): try: logger.info("[api/predict] Received request from %s", request.remote_addr) data = request.get_json() if not data or ("data" not in data and "audio" not in data): return jsonify( build_error_response( "Invalid request format", warnings=["Invalid request format"], ) ), 400 if isinstance(data.get("audio"), str): audio_base64 = data["audio"] else: audio_base64 = data["data"][0] if isinstance(data["data"], list) and data["data"] else "" result = analyze_emotion_base64(audio_base64) logger.info( "[api/predict] Completed request %s", json.dumps( { "emotion": result["final"]["emotion"], "confidence": result["final"]["confidence"], "warnings": result["meta"]["warnings"], } ), ) return jsonify(result) except Exception as error: return jsonify( build_error_response( f"Server error: {error}", warnings=[f"Server error: {error}"], ) ), 500 @app.route("/health", methods=["GET"]) def health(): return jsonify( { "status": "healthy", "cnnEndpoint": get_model_endpoint("cnn"), "wav2vecEndpoint": get_model_endpoint("wav2vec"), "cnnModel": get_cnn_model_repo(), "wav2vecModel": get_wav2vec_model_id(), "modelStatus": MODEL_STATUS, } ) if __name__ == "__main__": logger.info("Starting Flask API on http://127.0.0.1:7860") logger.info("Frontend BACKEND_API_URL should be http://127.0.0.1:7860/api/predict") app.run(host="0.0.0.0", port=7860, debug=False, use_reloader=False)