Blame for emotion.py

import time
from dataclasses import dataclass, field
from typing import Optional, Dict, Tuple, List

import gradio as gr
import numpy as np
from PIL import Image

from deepface import DeepFace


# =========================
# 參數（可微調）
# =========================
ANALYZE_EVERY_SEC = 0.8
DOWNSAMPLE_W = 720
EMA_ALPHA = 0.45
CONF_TH = 40.0
SWITCH_CONFIRM_SEC = 2.5
IOU_TH = 0.25

DETECTOR_BACKEND = "mtcnn"     # 想快一點可改 "opencv"
ALIGN_FACE = True


# =========================
# 情緒中文 + 分數映射（7 -> 5）
# =========================
EMO_ZH = {
    "angry": "生氣",
    "disgust": "厭惡",
    "fear": "害怕",
    "happy": "開心",
    "sad": "難過",
    "surprise": "驚訝",
    "neutral": "平靜",
    "unknown": "未知",
}

# 你要的：把 7 種情緒塞進 5 分心情分數
EMO_TO_SCORE = {
    "angry": 1,
    "disgust": 1,
    "fear": 1,
    "sad": 2,
    "neutral": 3,
    "surprise": 4,
    "happy": 5,
    "unknown": 3,
}

# ### [整合需求 新增] ###
# 為了讓主程式 (app.py) 可以呼叫這個轉換邏輯，將其包裝成函式
def get_emotion_score(emo: str) -> int:
    return EMO_TO_SCORE.get(emo, 3)
# ######################

SCORE_LABELS = {
    1: "1（心情差）",
    2: "2（不太好）",
    3: "3（普通）",
    4: "4（不錯）",
    5: "5（超棒）",
}
SCORE_CHOICES = [SCORE_LABELS[i] for i in [1, 2, 3, 4, 5]]


# =========================
# 狀態
# =========================
@dataclass
class AppState:
    mode: str = "攝影機辨識" # 保留欄位但固定為攝影機

    running: bool = True
    finished: bool = False
    last_analyze_t: float = 0.0

    # tracking
    track_bbox: Optional[Tuple[int, int, int, int]] = None

    # smoothing
    ema_scores: Optional[np.ndarray] = None
    emo_keys: List[str] = field(default_factory=list)

    # stable/candidate
    stable_emo: str = "unknown"
    candidate_emo: Optional[str] = None
    candidate_since: Optional[float] = None

    # final
    final_emo: Optional[str] = None
    final_conf: Optional[float] = None

    # ### [整合需求 新增] ###
    # 用來儲存最後計算出的整數分數 (1~5)，讓外部程式容易讀取
    final_score: int = 3  
    # ######################


# =========================
# UI HTML
# =========================
def _hint_html(msg: str) -> str:
    return f"""
    <div style="border-radius:14px;padding:14px;border:1px dashed #DADADA;background:#FAFAFA;color:#111;">
      {msg}
    </div>
    """


def _camera_hint_before_start() -> str:
    return _hint_html(
        "請先按「Click to access webcam / 允許」開啟攝影機；"
        "接著，請按「錄製（Record）」開始。"
    )


def _camera_hint_done() -> str:
    return _hint_html("已完成辨識，可按「重新攝影機辨識」再次進行辨識。")


def _camera_hint_stopped() -> str:
    return _hint_html("已停止辨識。可按「重新攝影機辨識」再試一次。")


def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
    zh = EMO_ZH.get(emo_key, emo_key)
    score = EMO_TO_SCORE.get(emo_key, 3)
    conf_txt = "" if conf is None else f"{conf:.1f}%"
    score_desc = SCORE_LABELS[score].split("（")[1].rstrip("）")

    # ✅ 結果字體用黑色，避免你說的灰字看不清楚
    return f"""
    <div style="
        border-radius:16px;
        padding:20px;
        border:1px solid #E6E6E6;
        background:#FFFFFF;
        color:#111;
    ">
      <div style="font-size:14px;color:#444;margin-bottom:8px;">
        辨識結果
      </div>

      <div style="
        font-size:42px;
        font-weight:800;
        line-height:1.1;
        margin-bottom:14px;
        color:#000;
      ">
        {zh}
      </div>

      <div style="font-size:16px;color:#111;margin-bottom:6px;">
        心情分數：<b>{score}</b>（{score_desc}）
      </div>

      <div style="font-size:14px;color:#111;">
        信心值：{conf_txt}
      </div>
    </div>
    """


def _score_radio_value(score: int) -> str:
    score = int(score)
    score = min(5, max(1, score))
    return SCORE_LABELS[score]


def _downsample_rgb(frame: np.ndarray, target_w: int) -> np.ndarray:
    if frame is None:
        return None
    h, w = frame.shape[:2]
    if w <= target_w:
        return frame
    scale = target_w / w
    new_h = max(1, int(h * scale))
    img = Image.fromarray(frame)
    img = img.resize((target_w, new_h))
    return np.array(img)


def _iou(a: Tuple[int, int, int, int], b: Tuple[int, int, int, int]) -> float:
    ax, ay, aw, ah = a
    bx, by, bw, bh = b
    ax2, ay2 = ax + aw, ay + ah
    bx2, by2 = bx + bw, by + bh

    ix1, iy1 = max(ax, bx), max(ay, by)
    ix2, iy2 = min(ax2, bx2), min(ay2, by2)
    iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1)
    inter = iw * ih
    if inter <= 0:
        return 0.0
    union = aw * ah + bw * bh - inter
    return inter / max(1e-6, union)


def _pick_face_by_tracking(
    faces: List[Dict],
    prev_bbox: Optional[Tuple[int, int, int, int]]
) -> Tuple[Optional[Dict], Optional[Tuple[int, int, int, int]]]:
    if not faces:
        return None, None

    cand = []
    for f in faces:
        fa = f.get("facial_area", {}) or {}
        x = int(fa.get("x", 0) or 0)
        y = int(fa.get("y", 0) or 0)
        w = int(fa.get("w", 0) or 0)
        h = int(fa.get("h", 0) or 0)
        if w > 0 and h > 0 and f.get("face") is not None:
            cand.append((f, (x, y, w, h), w * h))

    if not cand:
        return None, None

    if prev_bbox is not None:
        best = None
        best_i = 0.0
        for f, bb, area in cand:
            i = _iou(prev_bbox, bb)
            if i > best_i:
                best_i = i
                best = (f, bb)
        if best is not None and best_i >= IOU_TH:
            return best[0], best[1]

    cand.sort(key=lambda x: x[2], reverse=True)
    return cand[0][0], cand[0][1]


def _reset_state_for_camera(st: AppState) -> AppState:
    st.running = True
    st.finished = False
    st.last_analyze_t = 0.0
    st.track_bbox = None
    st.ema_scores = None
    st.emo_keys = []
    st.stable_emo = "unknown"
    st.candidate_emo = None
    st.candidate_since = None
    st.final_emo = None
    st.final_conf = None
    
    # ### [整合需求 新增] ###
    st.final_score = 3
    # ######################
    
    return st


# =========================
# Camera streaming
# =========================
def on_stream(frame_rgb: np.ndarray, st: AppState):
    # ✅ 完成/停止狀態：不要再顯示「Click to access…/Record…」
    if st.finished or (not st.running):
        cam_visible = False

        if st.final_emo:
            result_html = _result_card_html(st.final_emo, st.final_conf)
            mood_val = _score_radio_value(EMO_TO_SCORE.get(st.final_emo, 3))
            hint_html = _camera_hint_done()
        else:
            result_html = _hint_html("尚未完成辨識。")
            mood_val = _score_radio_value(3)
            hint_html = _camera_hint_stopped()

        return (
            gr.update(visible=cam_visible),
            gr.update(value=result_html),
            gr.update(value=mood_val),
            gr.update(value=hint_html),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    # ✅ 還沒拿到 frame：提示使用者按允許/Record（而不是亂寫已開啟）
    if frame_rgb is None:
        return (
            gr.update(visible=True),
            gr.update(value=_hint_html("等待影像…（尚未開始串流）")),
            gr.update(value=_score_radio_value(3)),
            gr.update(value=_camera_hint_before_start()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    now = time.time()
    if now - st.last_analyze_t < ANALYZE_EVERY_SEC:
        show = st.stable_emo or "unknown"
        return (
            gr.update(visible=True),
            gr.update(value=_result_card_html(show, st.final_conf)),
            gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
            gr.update(value=_camera_hint_before_start()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    st.last_analyze_t = now

    small = _downsample_rgb(frame_rgb, DOWNSAMPLE_W)

    faces = DeepFace.extract_faces(
        img_path=small,
        detector_backend=DETECTOR_BACKEND,
        enforce_detection=False,
        align=ALIGN_FACE,
    )
    pick, bbox = _pick_face_by_tracking(faces, st.track_bbox)
    if pick is None or bbox is None or pick.get("face") is None:
        show = st.stable_emo or "unknown"
        return (
            gr.update(visible=True),
            gr.update(value=_result_card_html(show, st.final_conf)),
            gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
            gr.update(value=_camera_hint_before_start()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    st.track_bbox = bbox
    face = pick["face"]
    if face.dtype != np.uint8:
        face = np.clip(face * 255.0, 0, 255).astype(np.uint8)

    result = DeepFace.analyze(
        img_path=face,
        actions=["emotion"],
        enforce_detection=False,
    )
    if isinstance(result, list):
        result = result[0]
    emo_dict = result.get("emotion", None)

    if not isinstance(emo_dict, dict) or len(emo_dict) == 0:
        show = st.stable_emo or "unknown"
        return (
            gr.update(visible=True),
            gr.update(value=_result_card_html(show, st.final_conf)),
            gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
            gr.update(value=_camera_hint_before_start()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    top_key = max(emo_dict, key=lambda k: float(emo_dict[k]))
    top_conf = float(emo_dict[top_key])

    if top_conf < CONF_TH:
        show = st.stable_emo or "unknown"
        return (
            gr.update(visible=True),
            gr.update(value=_result_card_html(show, st.final_conf)),
            gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
            gr.update(value=_camera_hint_before_start()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    keys = list(emo_dict.keys())
    vec = np.array([float(emo_dict[k]) for k in keys], dtype=np.float32)

    if st.ema_scores is None or st.emo_keys != keys:
        st.ema_scores = vec
        st.emo_keys = keys
    else:
        st.ema_scores = EMA_ALPHA * vec + (1.0 - EMA_ALPHA) * st.ema_scores

    stable_idx = int(np.argmax(st.ema_scores))
    stable_key = st.emo_keys[stable_idx]
    st.stable_emo = stable_key

    # 延遲確認
    if st.candidate_emo != stable_key:
        st.candidate_emo = stable_key
        st.candidate_since = now
    else:
        if st.candidate_since is None:
            st.candidate_since = now

    confirmed = (
        st.candidate_since is not None
        and (now - st.candidate_since) >= SWITCH_CONFIRM_SEC
    )

    if confirmed and st.final_emo is None:
        st.final_emo = stable_key
        st.final_conf = top_conf
        
        # ### [整合需求 新增] ###
        # 在辨識完成的瞬間，順便存入整數分數
        st.final_score = EMO_TO_SCORE.get(stable_key, 3)
        # ######################

        st.running = False
        st.finished = True

        return (
            gr.update(visible=False),
            gr.update(value=_result_card_html(stable_key, top_conf)),
            gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
            gr.update(value=_camera_hint_done()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    return (
        gr.update(visible=True),
        gr.update(value=_result_card_html(stable_key, top_conf)),
        gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
        gr.update(value=_camera_hint_before_start()),
        st,
        gr.update(value="重新攝影機辨識"),
    )


# =========================
# Mode / Buttons
# =========================

def on_restart(st: AppState):
    st = _reset_state_for_camera(st)
    return (
        gr.update(visible=True),                               # cam
        gr.update(value=_hint_html("等待影像…（尚未開始串流）")),  # result
        gr.update(value=_score_radio_value(3)),                # mood
        gr.update(value=_camera_hint_before_start()),          # hint
        st,
        gr.update(value="重新攝影機辨識"),                      # btn label
    )


def on_stop(st: AppState):
    st.running = False
    st.finished = True
    # ✅ 停止後也不要再顯示 Click/Record 提示，改成「已停止/可重新」
    if st.final_emo is None:
        return (
            gr.update(visible=False),
            gr.update(value=_hint_html("已停止辨識。")),
            gr.update(value=_score_radio_value(3)),
            gr.update(value=_camera_hint_stopped()),
            st,
            gr.update(value="重新攝影機辨識"),
        )

    return (
        gr.update(visible=False),
        gr.update(value=_result_card_html(st.final_emo, st.final_conf)),
        gr.update(value=_score_radio_value(EMO_TO_SCORE.get(st.final_emo, 3))),
        gr.update(value=_camera_hint_done()),
        st,
        gr.update(value="重新攝影機辨識"),
    )


# =========================
# Gradio UI
# =========================
css = """
#app_container { max-width: 960px; margin: 0 auto; }
.gradio-container { background: #f1f5f9 !important; }
"""

if __name__ == "__main__":
    with gr.Blocks(title="Emotion Detector") as demo:
        st = gr.State(AppState())

        gr.Markdown("## 情緒辨識")
        
        hint = gr.HTML(_camera_hint_before_start())

        cam = gr.Image(
            sources=["webcam"],
            streaming=True,
            type="numpy",
            label="攝影機",
            visible=True,
        )

        with gr.Row():
            btn_restart = gr.Button("重新攝影機辨識", variant="primary")
            btn_stop = gr.Button("停止", variant="secondary")

        # ✅ 你要的排序：結果在上、分數在下
        gr.Markdown("### 辨識結果")
        result = gr.HTML(_hint_html("等待影像…（尚未開始串流）"))

        gr.Markdown("### 心情分數（系統判定）")
        mood = gr.Radio(
            choices=SCORE_CHOICES,
            value=_score_radio_value(3),
            label="心情分數",
            interactive=False,
        )

        cam.stream(
            fn=on_stream,
            inputs=[cam, st],
            outputs=[cam, result, mood, hint, st, btn_restart],
            show_progress="minimal",
        )

        btn_restart.click(
            fn=on_restart,
            inputs=[st],
            outputs=[cam, result, mood, hint, st, btn_restart],
            show_progress="minimal",
        )

        btn_stop.click(
            fn=on_stop,
            inputs=[st],
            outputs=[cam, result, mood, hint, st, btn_restart],
            show_progress="minimal",
        )

    demo.launch(server_name="0.0.0.0", server_port=7860, share=False)