fuzzy_engine.py

"""Fuzzy Logic Fusion Engine — the core of Hermyon.

Replaces weighted averaging with a full fuzzy inference system:
  1. Fuzzification: map modality outputs to fuzzy membership degrees
  2. Rule evaluation: fire IF-THEN rules against fuzzy states
  3. Conflict detection: identify cross-modal disagreements
  4. Aggregation: combine rule outputs
  5. Defuzzification: produce crisp 28-dim GoEmotions probabilities
"""

from __future__ import annotations

import numpy as np
from typing import Optional

from constants import GOEMOTIONS_LABELS, NUM_GOEMOTIONS
from schemas import (
    ModalityResult,
    FuzzyFusionResult,
    FuzzyEmotionState,
    CrossModalConflict,
)
from projector import EmotionProjector
from membership import FuzzyMembershipFunctions, LEVEL_CENTROIDS
from rule_base import FuzzyRuleBase
from conflict_detector import ConflictDetector
from defuzzifier import Defuzzifier


class FuzzyFusionEngine:
    """Mamdani-style fuzzy inference system for multimodal emotion fusion."""

    def __init__(
        self,
        defuzzification_method: str = "centroid",
        conflict_threshold: float = 0.3,
        culture_id: str | None = None,
        base_weights: dict[str, float] | None = None,
    ):
        self.projector = EmotionProjector()
        self.mf = FuzzyMembershipFunctions
        self.rule_base = FuzzyRuleBase()
        self.conflict_detector = ConflictDetector(severity_threshold=conflict_threshold)
        self.defuzzifier = Defuzzifier(method=defuzzification_method)

        self.base_weights = base_weights or {
            "face": 0.20,
            "voice": 0.30,
            "text": 0.35,
            "posture": 0.15,
        }
        self.culture_id = culture_id

        if culture_id:
            self.rule_base.add_cultural_rules(culture_id)

    def fuse(
        self,
        face_result: ModalityResult | None = None,
        voice_result: ModalityResult | None = None,
        text_result: ModalityResult | None = None,
        posture_result: ModalityResult | None = None,
    ) -> FuzzyFusionResult:
        projected = {}
        confidences = {}
        modality_results = {}
        available = []

        if face_result is not None and face_result.confidence > 0:
            face_28 = self.projector.project_face(np.array(face_result.probabilities))
            projected["face"] = face_28
            confidences["face"] = face_result.confidence
            modality_results["face"] = face_result
            available.append("face")

        if voice_result is not None and voice_result.confidence > 0:
            voice_28 = self.projector.project_voice(np.array(voice_result.probabilities))
            projected["voice"] = voice_28
            confidences["voice"] = voice_result.confidence
            modality_results["voice"] = voice_result
            available.append("voice")

        if text_result is not None and text_result.confidence > 0:
            projected["text"] = np.array(text_result.probabilities)
            confidences["text"] = text_result.confidence
            modality_results["text"] = text_result
            available.append("text")

        if posture_result is not None and posture_result.confidence > 0:
            projected["posture"] = np.array(posture_result.probabilities)
            confidences["posture"] = posture_result.confidence
            modality_results["posture"] = posture_result
            available.append("posture")

        if not available:
            uniform = np.ones(NUM_GOEMOTIONS) / NUM_GOEMOTIONS
            return FuzzyFusionResult(
                labels=list(GOEMOTIONS_LABELS),
                probabilities=uniform.tolist(),
                fuzzy_states=[],
                conflicts=[],
                fired_rules=[],
                modality_weights={},
                modality_results={},
                conflict_score=0.0,
            )

        # Fuzzification
        modality_fuzzy: dict[str, dict[str, dict[str, float]]] = {}
        for mod_name in available:
            vec = projected[mod_name]
            fuzzy_states = {}
            for i, label in enumerate(GOEMOTIONS_LABELS):
                fuzzy_states[label] = self.mf.fuzzify(float(vec[i]))
            modality_fuzzy[mod_name] = fuzzy_states

        # Baseline weighted blend
        weights = {}
        for mod_name in available:
            base_w = self.base_weights.get(mod_name, 1.0 / len(available))
            weights[mod_name] = base_w * confidences[mod_name]

        total_w = sum(weights.values())
        if total_w > 0:
            weights = {k: v / total_w for k, v in weights.items()}

        base_crisp = np.zeros(NUM_GOEMOTIONS)
        for mod_name in available:
            base_crisp += weights[mod_name] * projected[mod_name]

        base_fuzzy = {}
        for i, label in enumerate(GOEMOTIONS_LABELS):
            base_fuzzy[label] = self.mf.fuzzify(float(base_crisp[i]))

        # 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 = self.rule_base.evaluate(face_fuzzy, voice_fuzzy, text_fuzzy, posture_fuzzy)

        # Conflict Detection
        conflicts = self.conflict_detector.detect(modality_fuzzy, available)
        conflict_score = self.conflict_detector.overall_conflict_score(conflicts)

        # Defuzzification
        crisp_result, fired_names = self.defuzzifier.defuzzify(base_fuzzy, fired_rules, base_crisp)

        # Build fuzzy state trace
        fuzzy_states = []
        for i, label in enumerate(GOEMOTIONS_LABELS):
            memberships = base_fuzzy.get(label, {"absent": 1.0})
            dominant = max(memberships, key=memberships.get)
            fuzzy_states.append(FuzzyEmotionState(
                emotion=label,
                memberships=memberships,
                dominant_level=dominant,
                crisp_value=float(crisp_result[i]),
            ))

        return FuzzyFusionResult(
            labels=list(GOEMOTIONS_LABELS),
            probabilities=crisp_result.tolist(),
            fuzzy_states=fuzzy_states,
            conflicts=conflicts,
            fired_rules=fired_names,
            modality_weights=weights,
            modality_results=modality_results,
            conflict_score=conflict_score,
        )

    def set_culture(self, culture_id: str) -> None:
        self.culture_id = culture_id
        self.rule_base = FuzzyRuleBase()
        self.rule_base.add_cultural_rules(culture_id)

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