File size: 4,567 Bytes
bae0f63 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
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()
|