app/services/fusion_engine.py

"""
fusion_engine.py — PsyPredict Multimodal Weighted Fusion Engine
Combines text emotion score + face emotion score → final risk score.
Weights are configurable via app config (TEXT_WEIGHT, FACE_WEIGHT).
Speech modality placeholder included for future expansion.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional

from app.config import get_settings

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Face emotion → distress score mapping
# Calibrated: fear/sadness = high distress, happy = minimal distress
# ---------------------------------------------------------------------------

FACE_DISTRESS_SCORES: dict[str, float] = {
    "fear": 0.80,
    "sad": 0.70,
    "angry": 0.50,
    "disgust": 0.40,
    "surprised": 0.30,
    "neutral": 0.20,
    "happy": 0.05,
}

# DistilBERT emotion labels → distress scores
TEXT_EMOTION_DISTRESS_SCORES: dict[str, float] = {
    "sadness": 0.85,
    "fear": 0.80,
    "anger": 0.60,
    "disgust": 0.50,
    "surprise": 0.30,
    "joy": 0.05,
    "love": 0.05,
    "neutral": 0.20,
}


@dataclass
class FusionResult:
    """Result of multimodal fusion scoring."""
    final_risk_score: float        # 0.0–1.0 weighted combined score
    text_score: float              # Raw text distress score
    face_score: float              # Raw face distress score
    speech_score: Optional[float]  # Placeholder — always None for now
    dominant_modality: str         # "text" | "face" | "balanced"
    text_weight: float
    face_weight: float


class FusionEngine:
    """
    Computes the weighted multimodal risk score.

    Formula:
        final_risk_score = (TEXT_WEIGHT * text_distress) + (FACE_WEIGHT * face_distress)

    Weights are loaded from app config at runtime.
    """

    def __init__(self) -> None:
        self.settings = get_settings()

    def _text_distress(self, dominant_text_emotion: str) -> float:
        """Map dominant text emotion label → distress score."""
        return TEXT_EMOTION_DISTRESS_SCORES.get(
            dominant_text_emotion.lower(), 0.20
        )

    def _face_distress(self, face_emotion: str) -> float:
        """Map face emotion label → distress score."""
        return FACE_DISTRESS_SCORES.get(face_emotion.lower(), 0.20)

    def compute(
        self,
        dominant_text_emotion: str,
        face_emotion: str,
        speech_score: Optional[float] = None,  # Future: speech sentiment
    ) -> FusionResult:
        """
        Compute weighted fusion score from available modalities.

        Args:
            dominant_text_emotion: Top emotion from DistilBERT (e.g. "sadness")
            face_emotion: Detected face emotion from Keras CNN (e.g. "sad")
            speech_score: Optional speech distress score (0.0–1.0)

        Returns:
            FusionResult with final weighted score and per-modality breakdown
        """
        tw = self.settings.TEXT_WEIGHT
        fw = self.settings.FACE_WEIGHT

        text_score = self._text_distress(dominant_text_emotion)
        face_score = self._face_distress(face_emotion)

        # If speech is provided in future, re-normalize weights
        if speech_score is not None:
            speech_weight = 1.0 - tw - fw
            if speech_weight > 0:
                final = (tw * text_score) + (fw * face_score) + (speech_weight * speech_score)
            else:
                final = (tw * text_score) + (fw * face_score)
        else:
            # Normalize text + face weights to sum to 1.0
            total = tw + fw
            final = ((tw / total) * text_score) + ((fw / total) * face_score)

        final = round(min(max(final, 0.0), 1.0), 4)

        # Determine dominant modality
        if abs(text_score - face_score) < 0.10:
            dominant = "balanced"
        elif text_score > face_score:
            dominant = "text"
        else:
            dominant = "face"

        logger.debug(
            "Fusion: text_emotion=%s(%.2f) face_emotion=%s(%.2f) → final=%.4f dominant=%s",
            dominant_text_emotion, text_score,
            face_emotion, face_score,
            final, dominant,
        )

        return FusionResult(
            final_risk_score=final,
            text_score=text_score,
            face_score=face_score,
            speech_score=speech_score,
            dominant_modality=dominant,
            text_weight=tw,
            face_weight=fw,
        )


# Singleton
fusion_engine = FusionEngine()