api_server.py

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)