face_module/label_mapping.py

from __future__ import annotations

import logging
from typing import Mapping

from config import FACE_LABELS

LOGGER = logging.getLogger(__name__)

DEEPFACE_TO_3CLASS = {
    "happy": "positive",
    # In this demo, surprise is treated as positive activation by default.
    "surprise": "positive",
    "neutral": "neutral",
    "sad": "negative",
    "angry": "negative",
    "fear": "negative",
    "disgust": "negative",
}


def map_deepface_emotion_to_3class(raw_emotion: str) -> str:
    """Map a DeepFace emotion label to positive/neutral/negative."""
    normalized = str(raw_emotion or "").strip().lower()
    mapped = DEEPFACE_TO_3CLASS.get(normalized)
    if mapped is None:
        LOGGER.warning("Unknown DeepFace emotion label %r; fallback to neutral.", raw_emotion)
        return "neutral"
    return mapped


def convert_deepface_emotion_scores(emotion_scores: Mapping[str, float] | None) -> dict[str, float]:
    """Aggregate DeepFace emotion scores into normalized 3-class probabilities.

    DeepFace may return values as percentages or probabilities. Because the final
    output is normalized by total mass, both formats are handled consistently.
    """
    if not emotion_scores:
        return {"positive": 0.0, "neutral": 1.0, "negative": 0.0}

    grouped = {label: 0.0 for label in FACE_LABELS}
    for raw_label, value in emotion_scores.items():
        mapped_label = map_deepface_emotion_to_3class(raw_label)
        try:
            numeric_value = float(value)
        except (TypeError, ValueError):
            LOGGER.warning("Invalid emotion score for %r: %r", raw_label, value)
            continue
        if numeric_value > 0:
            grouped[mapped_label] += numeric_value

    total = sum(grouped.values())
    if total <= 0:
        return {"positive": 0.0, "neutral": 1.0, "negative": 0.0}

    normalized = {label: grouped[label] / total for label in FACE_LABELS}
    return _round_and_balance(normalized, FACE_LABELS)


def _round_and_balance(values: dict[str, float], labels: list[str]) -> dict[str, float]:
    rounded = {label: round(max(0.0, min(1.0, values.get(label, 0.0))), 4) for label in labels}
    diff = round(1.0 - sum(rounded.values()), 4)
    if abs(diff) > 0 and labels:
        max_label = max(labels, key=lambda label: rounded[label])
        rounded[max_label] = round(max(0.0, min(1.0, rounded[max_label] + diff)), 4)
    return rounded