Update emotion.py

emotion.py

@@ -1,6 +1,7 @@
 import time
+import threading
 from dataclasses import dataclass, field
-from typing import Optional, Dict, Tuple, List
+from typing import Optional, List, Tuple, Dict
 
 import gradio as gr
 import numpy as np
@@ -17,16 +18,17 @@ from deepface import DeepFace
 # =========================
 # 參數（可微調）
 # =========================
-ANALYZE_EVERY_SEC = 0.8          # 情緒分析頻率（秒）
+ANALYZE_EVERY_SEC = 1.5          # ⭐ 建議 1.5~2.0：HF CPU 比較順
 EMA_ALPHA = 0.45
-CONF_TH = 40.0                  # DeepFace emotion confidence 門檻
-SWITCH_CONFIRM_SEC = 2.5        # 連續穩定多久才算完成
-ZOOM_FACTOR = 1.5               # 只用在「顯示」(你要保留 zoom 畫面)
+CONF_TH = 40.0                   # DeepFace emotion confidence 門檻
+SWITCH_CONFIRM_SEC = 2.5         # 連續穩定多久才算完成
+ZOOM_FACTOR = 1.5                # ✅ 保留：只用在「顯示」
+UI_TARGET_W = 720                # ✅ 回傳畫面縮小寬度，提升順暢度
 
 # MediaPipe face detector 門檻（太高很容易漏）
 MP_MIN_DET_CONF = 0.3
 
-# 模型路徑（請確認你的 repo 真的有這個檔案）
+# 模型路徑（確認 repo 中模型真的在這裡）
 MP_MODEL_PATH = "blaze_face_short_range.tflite"
 
 
@@ -163,16 +165,10 @@ def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
 
 
 # =========================
-# 影像處理（重點：保命轉換）
+# 影像處理（保命 + 顯示優化）
 # =========================
 def _to_rgb_uint8(frame: np.ndarray) -> np.ndarray:
-    """
-    MediaPipe Tasks 最穩格式：
-    - RGB
-    - uint8
-    - 0~255
-    - contiguous
-    """
+    """統一成 MediaPipe Tasks 最穩格式：RGB、uint8、0~255、contiguous。"""
     if frame is None:
         return frame
 
@@ -197,7 +193,6 @@ def _to_rgb_uint8(frame: np.ndarray) -> np.ndarray:
 def _zoom_center(frame: np.ndarray, zoom: float) -> np.ndarray:
     if frame is None or zoom <= 1.0:
         return frame
-
     h, w = frame.shape[:2]
     new_w = int(w / zoom)
     new_h = int(h / zoom)
@@ -206,6 +201,17 @@ def _zoom_center(frame: np.ndarray, zoom: float) -> np.ndarray:
     cropped = frame[start_y : start_y + new_h, start_x : start_x + new_w]
     return np.ascontiguousarray(cropped)
 
+def _resize_for_ui(frame_rgb: np.ndarray, target_w: int = UI_TARGET_W) -> np.ndarray:
+    """回傳給 Gradio 的畫面縮到固定寬度，減少傳輸/渲染負擔。"""
+    if frame_rgb is None:
+        return frame_rgb
+    h, w = frame_rgb.shape[:2]
+    if w <= target_w:
+        return frame_rgb
+    scale = target_w / w
+    new_h = max(1, int(h * scale))
+    return cv2.resize(frame_rgb, (target_w, new_h), interpolation=cv2.INTER_AREA)
+
 
 # =========================
 # MediaPipe Tasks Face Detector
@@ -239,7 +245,7 @@ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
         return frame_rgb, None
 
     best_det = None
-    best_area = -1
+    best_area = -1.0
     best_bbox = None
 
     # 最大臉：bbox 面積最大
@@ -280,6 +286,60 @@ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
     return drawn, (lx, ly, bw, bh)
 
 
+# =========================
+# 背景推論（避免卡頓）
+# =========================
+_ANALYZE_LOCK = threading.Lock()
+_ANALYZE_THREAD: Optional[threading.Thread] = None
+_PENDING_FACE: Optional[np.ndarray] = None
+_LAST_ANALYZE: Dict[str, Optional[float]] = {"ts": 0.0, "conf": None, "emo": "unknown"}
+
+def _analyze_worker():
+    """背景 thread：取最新 face_roi 做 DeepFace.analyze，更新 _LAST_ANALYZE。"""
+    global _PENDING_FACE, _ANALYZE_THREAD, _LAST_ANALYZE
+
+    while True:
+        with _ANALYZE_LOCK:
+            face = _PENDING_FACE
+            _PENDING_FACE = None
+
+        if face is None:
+            break
+
+        try:
+            r = DeepFace.analyze(
+                img_path=face,
+                actions=["emotion"],
+                enforce_detection=False,
+                detector_backend="skip",  # ⭐ 關鍵：你已經裁好 ROI
+            )
+            if isinstance(r, list):
+                r = r[0]
+            emo_dict = r.get("emotion", None)
+
+            if isinstance(emo_dict, dict) and len(emo_dict) > 0:
+                top_key = max(emo_dict, key=lambda k: float(emo_dict[k]))
+                top_conf = float(emo_dict[top_key])
+                _LAST_ANALYZE = {"ts": time.time(), "emo": top_key, "conf": top_conf}
+        except Exception:
+            # 失敗就保持上一次結果
+            pass
+
+    _ANALYZE_THREAD = None
+
+
+def _enqueue_analyze(face_roi: np.ndarray):
+    """把最新 ROI 丟給背景 thread，若 thread 沒在跑就啟動。"""
+    global _PENDING_FACE, _ANALYZE_THREAD
+
+    with _ANALYZE_LOCK:
+        _PENDING_FACE = face_roi
+
+    if _ANALYZE_THREAD is None:
+        _ANALYZE_THREAD = threading.Thread(target=_analyze_worker, daemon=True)
+        _ANALYZE_THREAD.start()
+
+
 # =========================
 # state helper
 # =========================
@@ -338,7 +398,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
             gr.update(value="重新攝影機辨識"),
         )
 
-    # 先整理格式（超重要：不然可能永遠偵測不到）
+    # 統一格式
     frame_rgb = _to_rgb_uint8(frame_rgb)
 
     # ✅ 偵測用原圖（穩）
@@ -346,6 +406,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
 
     # ✅ 顯示用 zoom（你要保留 zoom 畫面）
     frame_show = _zoom_center(frame_drawn if bbox_px is not None else frame_rgb, ZOOM_FACTOR)
+    frame_show = _resize_for_ui(frame_show, UI_TARGET_W)
 
     # 沒有人臉：不畫框/點 + 提示
     if bbox_px is None:
@@ -360,44 +421,28 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
 
     now = time.time()
 
-    # 還沒到分析時間：只更新畫面（框/點會一直在），不做情緒分析
-    if now - st.last_analyze_t < ANALYZE_EVERY_SEC:
-        show = st.stable_emo or "unknown"
-        return (
-            gr.update(value=frame_show, 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="重新攝影機辨識"),
-        )
-
-    # 到分析時間：用「原圖 bbox」裁 ROI 做情緒辨識（不要用 zoom 後的圖裁）
-    st.last_analyze_t = now
-
-    x, y, bw, bh = bbox_px
-    pad_x = int(bw * 0.10)
-    pad_y = int(bh * 0.10)
+    # 到分析時間：只排隊給背景 thread（不在 on_stream 裡跑重推論）
+    if now - st.last_analyze_t >= ANALYZE_EVERY_SEC:
+        st.last_analyze_t = now
 
-    H, W = frame_rgb.shape[:2]
-    x1 = max(0, x - pad_x)
-    y1 = max(0, y - pad_y)
-    x2 = min(W, x + bw + pad_x)
-    y2 = min(H, y + bh + pad_y)
+        x, y, bw, bh = bbox_px
+        pad_x = int(bw * 0.10)
+        pad_y = int(bh * 0.10)
+        H, W = frame_rgb.shape[:2]
+        x1 = max(0, x - pad_x)
+        y1 = max(0, y - pad_y)
+        x2 = min(W, x + bw + pad_x)
+        y2 = min(H, y + bh + pad_y)
+        face_roi = frame_rgb[y1:y2, x1:x2].astype(np.uint8)
 
-    face_roi = frame_rgb[y1:y2, x1:x2].astype(np.uint8)
+        _enqueue_analyze(face_roi)
 
-    # DeepFace.analyze
-    result = DeepFace.analyze(
-        img_path=face_roi,
-        actions=["emotion"],
-        enforce_detection=False,
-    )
-    if isinstance(result, list):
-        result = result[0]
+    # 取「最後一次背景推論結果」
+    top_key = _LAST_ANALYZE.get("emo", "unknown") or "unknown"
+    top_conf = _LAST_ANALYZE.get("conf", None)
 
-    emo_dict = result.get("emotion", None)
-    if not isinstance(emo_dict, dict) or len(emo_dict) == 0:
+    # 還沒有推論到結果：先用目前穩定值顯示
+    if top_conf is None:
         show = st.stable_emo or "unknown"
         return (
             gr.update(value=frame_show, visible=True),
@@ -408,9 +453,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
             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 (
@@ -422,18 +465,23 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
             gr.update(value="重新攝影機辨識"),
         )
 
-    # EMA smoothing
-    keys = list(emo_dict.keys())
-    vec = np.array([float(emo_dict[k]) for k in keys], dtype=np.float32)
+    # === EMA smoothing ===
+    # DeepFace emotion dict 的 keys 順序可能固定，但我們這裡用單一 top_conf/top_key 也能做穩定
+    # 仍然保留你原本的「看整個分佈」的 smoothing：用 top_key/top_conf 建一個簡化 dict
+    # 為了兼容你原本寫法，我們用 7 類別做一個向量（缺的填 0）
+    all_keys = ["angry", "disgust", "fear", "happy", "sad", "surprise", "neutral"]
+    vec = np.zeros((len(all_keys),), dtype=np.float32)
+    if top_key in all_keys:
+        vec[all_keys.index(top_key)] = float(top_conf)
 
-    if st.ema_scores is None or st.emo_keys != keys:
+    if st.ema_scores is None or st.emo_keys != all_keys:
         st.ema_scores = vec
-        st.emo_keys = keys
+        st.emo_keys = all_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]
+    stable_key = st.emo_keys[stable_idx] if st.emo_keys else "unknown"
     st.stable_emo = stable_key
 
     # 延遲確認，避免跳動
@@ -449,10 +497,10 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
         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_conf = float(top_conf) if top_conf is not None else None
         st.final_score = EMO_TO_SCORE.get(stable_key, 3)
 
         st.running = False
@@ -460,7 +508,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
 
         return (
             gr.update(value=None, visible=False),
-            gr.update(value=_result_card_html(stable_key, top_conf)),
+            gr.update(value=_result_card_html(stable_key, st.final_conf)),
             gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
             gr.update(value=_camera_hint_done()),
             st,
@@ -469,7 +517,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
 
     return (
         gr.update(value=frame_show, visible=True),
-        gr.update(value=_result_card_html(stable_key, top_conf)),
+        gr.update(value=_result_card_html(stable_key, float(top_conf))),
         gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
         gr.update(value=_camera_hint_before_start()),
         st,