fuzzy_fusion.py

"""EmoSphere Fuzzy Fusion Engine — Mamdani fuzzy inference for 9 emotions.

Ported from Hermyon's FuzzyFusionEngine but adapted for EmoSphere's
consumer-grade 9-emotion label set (joy, sadness, surprise, fear,
disgust, anger, neutral, love, calm).

Pipeline:
  1. Fuzzification: map detector scores to 5 membership levels
  2. Rule evaluation: fire agreement/conflict IF-THEN rules
  3. Defuzzification: centroid method to produce crisp scores
  4. Normalization: output valid probability distribution
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Callable, Optional

import numpy as np

from models import (
    EmotionLabel,
    EMOTION_LABELS,
    EmotionScore,
    EmotionDetectionResult,
    FusedDetectionResult,
    CulturalRegion,
)


# =====================================================================
# Trapezoidal Membership Functions (5 levels)
# =====================================================================

FUZZY_LEVELS = ["absent", "low", "moderate", "high", "very_high"]

LEVEL_CENTROIDS = {
    "absent": 0.0,
    "low": 0.15,
    "moderate": 0.35,
    "high": 0.60,
    "very_high": 0.85,
}


def _trapezoid(x: float, a: float, b: float, c: float, d: float) -> float:
    """Standard trapezoidal membership function."""
    if x <= a or x >= d:
        return 0.0
    if b <= x <= c:
        return 1.0
    if a < x < b:
        return (x - a) / (b - a)
    return (d - x) / (d - c)


def _left_shoulder(x: float, a: float, b: float) -> float:
    if x <= a:
        return 1.0
    if x >= b:
        return 0.0
    return (b - x) / (b - a)


def _right_shoulder(x: float, a: float, b: float) -> float:
    if x <= a:
        return 0.0
    if x >= b:
        return 1.0
    return (x - a) / (b - a)


def fuzzify(crisp_value: float) -> dict[str, float]:
    """Convert a crisp probability [0,1] to fuzzy membership degrees."""
    return {
        "absent": _left_shoulder(crisp_value, 0.05, 0.12),
        "low": _trapezoid(crisp_value, 0.05, 0.12, 0.20, 0.30),
        "moderate": _trapezoid(crisp_value, 0.20, 0.30, 0.45, 0.55),
        "high": _trapezoid(crisp_value, 0.45, 0.55, 0.70, 0.80),
        "very_high": _right_shoulder(crisp_value, 0.70, 0.80),
    }


# =====================================================================
# Fuzzy Rule Definitions
# =====================================================================

@dataclass
class FuzzyRule:
    name: str
    category: str  # "agreement" or "conflict"
    condition: Callable  # (face_fuzzy, voice_fuzzy, text_fuzzy, posture_fuzzy) -> float
    consequent: dict[str, tuple[str, float]]  # EmotionLabel.value -> (level, weight_modifier)
    priority: int = 1


def _get(fuzzy_states: Optional[dict], emotion: str, level: str) -> float:
    """Safely get a fuzzy membership degree from a modality's fuzzified output."""
    if fuzzy_states is None:
        return 0.0
    state = fuzzy_states.get(emotion, {})
    return state.get(level, 0.0)


def _above(fuzzy_states: Optional[dict], emotion: str, min_level: str) -> float:
    """Get max membership at or above a given level."""
    if fuzzy_states is None:
        return 0.0
    state = fuzzy_states.get(emotion, {})
    idx = FUZZY_LEVELS.index(min_level)
    return max(state.get(lv, 0.0) for lv in FUZZY_LEVELS[idx:])


def _build_rules() -> list[FuzzyRule]:
    """Build the consumer-grade fuzzy rule set for 9 emotions."""
    rules: list[FuzzyRule] = []

    # ── AGREEMENT RULES ──────────────────────────────────────────

    # R1: Genuine joy -- face+voice+text all show joy
    rules.append(FuzzyRule(
        name="R01_genuine_joy",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "joy", "high"),
            _above(v, "joy", "moderate"),
            _above(t, "joy", "moderate"),
        ),
        consequent={
            "joy": ("very_high", 1.3),
            "love": ("moderate", 0.8),
        },
        priority=2,
    ))

    # R2: Deep sadness -- confirmed across modalities
    rules.append(FuzzyRule(
        name="R02_deep_sadness",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "sadness", "high"),
            _above(v, "sadness", "moderate"),
            _above(t, "sadness", "moderate"),
        ),
        consequent={
            "sadness": ("very_high", 1.3),
            "neutral": ("absent", 0.3),
        },
        priority=2,
    ))

    # R3: Confirmed fear -- face+voice+posture
    rules.append(FuzzyRule(
        name="R03_confirmed_fear",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "fear", "moderate"),
            _above(v, "fear", "moderate"),
            max(_above(p, "fear", "moderate") if p else 0.0,
                _above(t, "fear", "moderate")),
        ),
        consequent={
            "fear": ("very_high", 1.4),
            "calm": ("absent", 0.2),
        },
        priority=2,
    ))

    # R4: Confirmed disgust -- face+voice+text agree
    rules.append(FuzzyRule(
        name="R04_confirmed_disgust",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "disgust", "moderate"),
            _above(v, "disgust", "moderate"),
            _above(t, "disgust", "low"),
        ),
        consequent={
            "disgust": ("very_high", 1.3),
        },
        priority=2,
    ))

    # R5: Genuine surprise -- face+voice agree
    rules.append(FuzzyRule(
        name="R05_genuine_surprise",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "surprise", "high"),
            _above(v, "surprise", "moderate"),
        ),
        consequent={
            "surprise": ("very_high", 1.2),
        },
        priority=2,
    ))

    # R6: Deep love -- text+voice+face gentle
    rules.append(FuzzyRule(
        name="R06_deep_love",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(t, "love", "high"),
            max(_above(f, "joy", "low"), _above(f, "calm", "low"), 0.2),
        ),
        consequent={
            "love": ("very_high", 1.3),
            "joy": ("moderate", 0.8),
            "calm": ("moderate", 0.7),
        },
        priority=2,
    ))

    # R7: Deep calm -- all modalities relaxed
    rules.append(FuzzyRule(
        name="R07_deep_calm",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "calm", "moderate"),
            _above(v, "calm", "moderate"),
            max(_above(t, "calm", "low"), _above(t, "neutral", "moderate")),
        ),
        consequent={
            "calm": ("very_high", 1.2),
            "neutral": ("moderate", 0.7),
        },
        priority=1,
    ))

    # ── CONFLICT RULES ───────────────────────────────────────────

    # R8: Smile masking sadness -- face happy but voice sad
    rules.append(FuzzyRule(
        name="R08_smile_masking_sadness",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(f, "joy", "moderate"),
            _above(v, "sadness", "moderate"),
        ),
        consequent={
            "sadness": ("high", 1.4),
            "joy": ("low", 0.4),
        },
        priority=3,
    ))

    # R9: Suppressed anger/disgust -- face neutral but voice tense
    rules.append(FuzzyRule(
        name="R09_suppressed_disgust",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(f, "neutral", "high"),
            _above(v, "disgust", "moderate"),
        ),
        consequent={
            "disgust": ("high", 1.5),
            "anger": ("moderate", 1.2),
            "neutral": ("low", 0.3),
        },
        priority=3,
    ))

    # R10: Hidden anxiety -- posture tense but face/voice neutral
    rules.append(FuzzyRule(
        name="R10_hidden_anxiety",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(p, "fear", "moderate") if p else 0.0,
            _above(f, "neutral", "moderate"),
        ),
        consequent={
            "fear": ("high", 1.4),
            "neutral": ("low", 0.3),
        },
        priority=3,
    ))

    # R11: Social desirability -- face+voice happy but text reveals negative
    rules.append(FuzzyRule(
        name="R11_social_desirability",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(f, "joy", "moderate"),
            max(_above(v, "joy", "low"), 0.15),
            max(
                _above(t, "sadness", "moderate"),
                _above(t, "fear", "moderate"),
                _above(t, "disgust", "moderate"),
            ),
        ),
        consequent={
            "sadness": ("high", 1.3),
            "fear": ("moderate", 1.2),
            "joy": ("low", 0.4),
        },
        priority=3,
    ))

    # R12: Confirmed anger -- face+voice+text agree
    rules.append(FuzzyRule(
        name="R12_confirmed_anger",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(f, "anger", "moderate"),
            max(_above(v, "anger", "moderate"),
                _above(t, "anger", "moderate")),
        ),
        consequent={
            "anger": ("very_high", 1.3),
            "calm": ("absent", 0.2),
        },
        priority=2,
    ))

    # ── VOICE/TEXT OVERRIDE RULES ────────────────────────────────
    # These ensure that voice and text can strongly influence the
    # result even when face shows a different emotion.

    # R13: Voice-driven emotion -- voice is strong, override face
    for emo in ["joy", "sadness", "fear", "surprise", "anger", "disgust"]:
        rules.append(FuzzyRule(
            name=f"R13_voice_{emo}",
            category="agreement",
            condition=lambda f, v, t, p, e=emo: (
                _above(v, e, "high") * 0.8
                if _above(v, e, "high") > _above(f, e, "low")
                else 0.0
            ),
            consequent={
                emo: ("high", 1.2),
            },
            priority=2,
        ))

    # R14: Text/speech-driven emotion -- text strongly indicates an emotion
    for emo in ["joy", "sadness", "fear", "surprise", "anger", "disgust", "love"]:
        rules.append(FuzzyRule(
            name=f"R14_text_{emo}",
            category="agreement",
            condition=lambda f, v, t, p, e=emo: (
                _above(t, e, "high") * 0.8
                if _above(t, e, "high") > _above(f, e, "low")
                else 0.0
            ),
            consequent={
                emo: ("high", 1.2),
            },
            priority=2,
        ))

    # R15: Posture-driven emotion -- posture/gesture strongly indicates
    for emo in ["fear", "anger", "sadness", "calm"]:
        rules.append(FuzzyRule(
            name=f"R15_posture_{emo}",
            category="agreement",
            condition=lambda f, v, t, p, e=emo: (
                _above(p, e, "high") * 0.7 if p else 0.0
            ),
            consequent={
                emo: ("moderate", 1.1),
            },
            priority=1,
        ))

    # R16: Face contradicted -- face says X but 2+ other modalities say Y
    # This dampens face when it disagrees with majority
    rules.append(FuzzyRule(
        name="R16_face_contradicted_joy",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(f, "joy", "high"),
            max(
                min(_above(v, "sadness", "moderate"), _above(t, "sadness", "low")),
                min(_above(v, "anger", "moderate"), _above(t, "anger", "low")),
                min(_above(v, "fear", "moderate"), _above(t, "fear", "low")),
            ),
        ),
        consequent={
            "joy": ("low", 0.3),
            "sadness": ("moderate", 1.2),
        },
        priority=3,
    ))

    rules.append(FuzzyRule(
        name="R17_face_contradicted_neutral",
        category="conflict",
        condition=lambda f, v, t, p: min(
            _above(f, "neutral", "high"),
            max(
                _above(v, "sadness", "high"),
                _above(v, "anger", "high"),
                _above(v, "fear", "high"),
                _above(t, "sadness", "high"),
                _above(t, "anger", "high"),
            ),
        ),
        consequent={
            "neutral": ("low", 0.3),
        },
        priority=3,
    ))

    # R18: Voice+text agree but face differs -- trust voice+text
    rules.append(FuzzyRule(
        name="R18_voice_text_agree_surprise",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(v, "surprise", "moderate"),
            _above(t, "surprise", "low"),
        ) * 0.9,
        consequent={
            "surprise": ("high", 1.3),
        },
        priority=2,
    ))

    rules.append(FuzzyRule(
        name="R19_voice_text_agree_sadness",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(v, "sadness", "moderate"),
            _above(t, "sadness", "low"),
        ) * 0.9,
        consequent={
            "sadness": ("high", 1.3),
        },
        priority=2,
    ))

    rules.append(FuzzyRule(
        name="R20_voice_text_agree_anger",
        category="agreement",
        condition=lambda f, v, t, p: min(
            _above(v, "anger", "moderate"),
            _above(t, "anger", "low"),
        ) * 0.9,
        consequent={
            "anger": ("high", 1.3),
        },
        priority=2,
    ))

    return rules


# =====================================================================
# Defuzzifier (Centroid Method)
# =====================================================================

def _defuzzify_centroid(
    base_fuzzy: dict[str, dict[str, float]],
    fired_rules: list[tuple[FuzzyRule, float]],
    base_crisp: dict[EmotionLabel, float],
) -> dict[EmotionLabel, float]:
    """Centroid defuzzification with rule blending."""
    result = {label: base_crisp.get(label, 0.0) for label in EMOTION_LABELS}

    # Apply rule adjustments
    rule_targets: dict[str, float] = {}
    rule_activations: dict[str, float] = {}

    for rule, activation in fired_rules:
        for emotion_val, (target_level, weight_mod) in rule.consequent.items():
            target_centroid = LEVEL_CENTROIDS.get(target_level, 0.35)
            effective = target_centroid * activation * weight_mod
            if emotion_val not in rule_targets or effective > rule_targets[emotion_val]:
                rule_targets[emotion_val] = effective
                rule_activations[emotion_val] = activation

    for label in EMOTION_LABELS:
        val = label.value
        if val in rule_activations:
            # Stronger rule blending — rules are the primary decision mechanism
            blend = min(rule_activations[val] * 1.3, 1.0)
            result[label] = (1.0 - blend) * result[label] + blend * rule_targets[val]

    # Centroid refinement from fuzzy memberships (lighter touch)
    for label in EMOTION_LABELS:
        memberships = base_fuzzy.get(label.value, {})
        numerator = 0.0
        denominator = 0.0
        for level, mu in memberships.items():
            if mu > 0 and level in LEVEL_CENTROIDS:
                numerator += mu * LEVEL_CENTROIDS[level]
                denominator += mu
        if denominator > 0:
            cog = numerator / denominator
            result[label] = 0.8 * result[label] + 0.2 * cog

    # Normalize
    result = {k: max(v, 0.0) for k, v in result.items()}
    total = sum(result.values())
    if total > 0:
        result = {k: v / total for k, v in result.items()}
    else:
        result = {label: 1.0 / len(EMOTION_LABELS) for label in EMOTION_LABELS}

    return result


# =====================================================================
# FuzzyFusionEngine
# =====================================================================

class FuzzyFusionEngine:
    """Mamdani-style fuzzy inference for EmoSphere's 9 emotions.

    Replaces simple weighted averaging with fuzzy rule-based fusion that
    can detect cross-modal agreement, conflict, and masking patterns.

    Modality weights (equal contribution when all active):
      face: 0.25, voice: 0.25, text: 0.25, posture: 0.25

    The fuzzy rules are the primary decision mechanism — these weights
    only provide the initial crisp baseline that rules then modify.
    """

    BASE_WEIGHTS = {
        "face": 0.25,
        "voice": 0.25,
        "text": 0.25,
        "posture": 0.25,
    }

    def __init__(self):
        self.rules = _build_rules()

    def fuse(
        self,
        face: Optional[EmotionDetectionResult] = None,
        voice: Optional[EmotionDetectionResult] = None,
        text: Optional[EmotionDetectionResult] = None,
        posture: Optional[EmotionDetectionResult] = None,
    ) -> FusedDetectionResult:
        """Fuse available modality results using fuzzy inference."""
        import time
        start = time.time()

        available: list[tuple[str, EmotionDetectionResult]] = []
        if face:
            available.append(("face", face))
        if voice:
            available.append(("voice", voice))
        if text:
            available.append(("text", text))
        if posture:
            available.append(("posture", posture))

        if not available:
            neutral_scores = [
                EmotionScore(label=label, score=1.0 if label == EmotionLabel.NEUTRAL else 0.0, confidence=0.0)
                for label in EMOTION_LABELS
            ]
            return FusedDetectionResult(
                dominant=EmotionLabel.NEUTRAL,
                dominant_score=1.0,
                scores=neutral_scores,
                modality_weights={},
                confidence=0.0,
                processing_time_ms=0.0,
            )

        # -- Step 1: Extract score dicts from each modality --
        modality_scores: dict[str, dict[EmotionLabel, float]] = {}
        confidences: dict[str, float] = {}
        for mod_name, result in available:
            scores = {s.label: s.score for s in result.scores}
            modality_scores[mod_name] = scores
            confidences[mod_name] = max(result.confidence, 0.01)

        # -- Step 2: Equal weights for all available modalities --
        n = len(modality_scores)
        weights: dict[str, float] = {mod: 1.0 / n for mod in modality_scores}

        # -- Step 3: Weighted baseline blend --
        base_crisp: dict[EmotionLabel, float] = {label: 0.0 for label in EMOTION_LABELS}
        for mod_name, scores in modality_scores.items():
            w = weights.get(mod_name, 0.0)
            for label in EMOTION_LABELS:
                base_crisp[label] += scores.get(label, 0.0) * w

        # -- Step 4: Fuzzification --
        modality_fuzzy: dict[str, dict[str, dict[str, float]]] = {}
        for mod_name, scores in modality_scores.items():
            fuzzy_states = {}
            for label in EMOTION_LABELS:
                fuzzy_states[label.value] = fuzzify(scores.get(label, 0.0))
            modality_fuzzy[mod_name] = fuzzy_states

        base_fuzzy: dict[str, dict[str, float]] = {}
        for label in EMOTION_LABELS:
            base_fuzzy[label.value] = fuzzify(base_crisp[label])

        # -- Step 5: Rule evaluation --
        face_fuzzy = modality_fuzzy.get("face")
        voice_fuzzy = modality_fuzzy.get("voice")
        text_fuzzy = modality_fuzzy.get("text")
        posture_fuzzy = modality_fuzzy.get("posture")

        fired_rules: list[tuple[FuzzyRule, float]] = []
        for rule in self.rules:
            try:
                activation = rule.condition(face_fuzzy, voice_fuzzy, text_fuzzy, posture_fuzzy)
                activation = max(0.0, min(1.0, float(activation)))
                if activation > 0.05:
                    fired_rules.append((rule, activation))
            except (TypeError, KeyError, ValueError):
                continue

        fired_rules.sort(key=lambda x: (x[0].priority, x[1]), reverse=True)

        # -- Step 6: Defuzzification --
        fused = _defuzzify_centroid(base_fuzzy, fired_rules, base_crisp)

        # -- Build result --
        scores = [
            EmotionScore(label=label, score=fused[label], confidence=fused[label])
            for label in EMOTION_LABELS
        ]
        dominant = max(fused, key=fused.get)  # type: ignore

        return FusedDetectionResult(
            dominant=dominant,
            dominant_score=fused[dominant],
            scores=scores,
            face_result=face,
            voice_result=voice,
            text_result=text,
            posture_result=posture,
            modality_weights=weights,
            confidence=max(r.confidence for _, r in available) * 0.95,
            processing_time_ms=(time.time() - start) * 1000,
        )

    @property
    def fired_rule_names(self) -> list[str]:
        """Convenience — last fusion's fired rules. For reporting, call fuse() and inspect."""
        return []

    @property
    def num_rules(self) -> int:
        return len(self.rules)