PhilipL commited on
Commit
260a447
·
verified ·
1 Parent(s): c8b2595

Upload emotion.py

Browse files
Files changed (1) hide show
  1. emotion.py +274 -180
emotion.py CHANGED
@@ -1,10 +1,15 @@
1
  import time
 
2
  from dataclasses import dataclass, field
3
- from typing import Optional, Dict, Tuple, List
4
 
5
  import gradio as gr
6
  import numpy as np
7
- from PIL import Image
 
 
 
 
8
 
9
  from deepface import DeepFace
10
 
@@ -12,16 +17,18 @@ from deepface import DeepFace
12
  # =========================
13
  # 參數(可微調)
14
  # =========================
15
- ANALYZE_EVERY_SEC = 0.8
16
- DOWNSAMPLE_W = 720
17
  EMA_ALPHA = 0.45
18
- CONF_TH = 40.0
19
- SWITCH_CONFIRM_SEC = 2.5
20
- IOU_TH = 0.25
21
- ZOOM_FACTOR = 1.5
 
 
 
22
 
23
- DETECTOR_BACKEND = "mtcnn" # 想快一點可改 "opencv"
24
- ALIGN_FACE = True
25
 
26
 
27
  # =========================
@@ -38,7 +45,6 @@ EMO_ZH = {
38
  "unknown": "未知",
39
  }
40
 
41
- # 你要的:把 7 種情緒塞進 5 分心情分數
42
  EMO_TO_SCORE = {
43
  "angry": 1,
44
  "disgust": 1,
@@ -50,9 +56,6 @@ EMO_TO_SCORE = {
50
  "unknown": 3,
51
  }
52
 
53
- def get_emotion_score(emo: str) -> int:
54
- return EMO_TO_SCORE.get(emo, 3)
55
-
56
  SCORE_LABELS = {
57
  1: "1(心情差)",
58
  2: "2(不太好)",
@@ -68,15 +71,12 @@ SCORE_CHOICES = [SCORE_LABELS[i] for i in [1, 2, 3, 4, 5]]
68
  # =========================
69
  @dataclass
70
  class AppState:
71
- mode: str = "攝影機辨識"
72
 
73
  running: bool = True
74
  finished: bool = False
75
  last_analyze_t: float = 0.0
76
 
77
- # tracking
78
- track_bbox: Optional[Tuple[int, int, int, int]] = None
79
-
80
  # smoothing
81
  ema_scores: Optional[np.ndarray] = None
82
  emo_keys: List[str] = field(default_factory=list)
@@ -89,36 +89,40 @@ class AppState:
89
  # final
90
  final_emo: Optional[str] = None
91
  final_conf: Optional[float] = None
92
-
93
- final_score: int = 3
94
 
95
 
96
  # =========================
97
- # UI HTML (強制黑色文字)
98
  # =========================
99
  def _hint_html(msg: str) -> str:
100
- # 加入 color:#000000 強制黑色
101
  return f"""
102
  <div style="border-radius:14px;padding:14px;border:1px dashed #DADADA;background:#FAFAFA;color:#000000;">
103
  {msg}
104
  </div>
105
  """
106
 
107
-
108
  def _camera_hint_before_start() -> str:
109
  return _hint_html(
110
  "請先按「Click to access webcam / 允許」開啟攝影機;"
111
  "接著,請按「錄製(Record)」開始。"
112
  )
113
 
 
 
 
 
114
 
115
  def _camera_hint_done() -> str:
116
  return _hint_html("已完成辨識,可按「重新攝影機辨識」再次進行辨識。")
117
 
118
-
119
  def _camera_hint_stopped() -> str:
120
  return _hint_html("已停止辨識。可按「重新攝影機辨識」再試一次。")
121
 
 
 
 
 
122
 
123
  def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
124
  zh = EMO_ZH.get(emo_key, emo_key)
@@ -126,14 +130,13 @@ def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
126
  conf_txt = "" if conf is None else f"{conf:.1f}%"
127
  score_desc = SCORE_LABELS[score].split("(")[1].rstrip(")")
128
 
129
- # ✅ 強制設定 color: #000000
130
  return f"""
131
  <div style="
132
  border-radius:16px;
133
  padding:20px;
134
  border:1px solid #E6E6E6;
135
  background:#FFFFFF;
136
- color:#000000;
137
  ">
138
  <div style="font-size:14px;color:#000000;margin-bottom:8px;">
139
  辨識結果
@@ -160,118 +163,206 @@ def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
160
  """
161
 
162
 
163
- def _score_radio_value(score: int) -> str:
164
- score = int(score)
165
- score = min(5, max(1, score))
166
- return SCORE_LABELS[score]
167
-
168
-
169
- def _downsample_rgb(frame: np.ndarray, target_w: int) -> np.ndarray:
170
  if frame is None:
171
- return None
172
- h, w = frame.shape[:2]
173
- if w <= target_w:
174
  return frame
175
- scale = target_w / w
176
- new_h = max(1, int(h * scale))
177
- img = Image.fromarray(frame)
178
- img = img.resize((target_w, new_h))
179
- return np.array(img)
 
 
 
 
 
 
 
 
 
 
 
 
 
180
 
181
  def _zoom_center(frame: np.ndarray, zoom: float) -> np.ndarray:
182
- """
183
- 裁切畫面中心以模擬數位變焦。
184
- zoom: 放大倍率 (例如 1.5)
185
- """
186
  if frame is None or zoom <= 1.0:
187
  return frame
188
-
189
  h, w = frame.shape[:2]
190
-
191
- # 計算新的寬高
192
  new_w = int(w / zoom)
193
  new_h = int(h / zoom)
194
-
195
- # 計算起始點 (讓裁切框置中)
196
  start_x = (w - new_w) // 2
197
  start_y = (h - new_h) // 2
198
-
199
- # 進行裁切
200
  cropped = frame[start_y : start_y + new_h, start_x : start_x + new_w]
201
- return cropped
202
-
203
-
204
- def _iou(a: Tuple[int, int, int, int], b: Tuple[int, int, int, int]) -> float:
205
- ax, ay, aw, ah = a
206
- bx, by, bw, bh = b
207
- ax2, ay2 = ax + aw, ay + ah
208
- bx2, by2 = bx + bw, by + bh
209
-
210
- ix1, iy1 = max(ax, bx), max(ay, by)
211
- ix2, iy2 = min(ax2, bx2), min(ay2, by2)
212
- iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1)
213
- inter = iw * ih
214
- if inter <= 0:
215
- return 0.0
216
- union = aw * ah + bw * bh - inter
217
- return inter / max(1e-6, union)
218
-
219
-
220
- def _pick_face_by_tracking(
221
- faces: List[Dict],
222
- prev_bbox: Optional[Tuple[int, int, int, int]]
223
- ) -> Tuple[Optional[Dict], Optional[Tuple[int, int, int, int]]]:
224
- if not faces:
225
- return None, None
226
-
227
- cand = []
228
- for f in faces:
229
- fa = f.get("facial_area", {}) or {}
230
- x = int(fa.get("x", 0) or 0)
231
- y = int(fa.get("y", 0) or 0)
232
- w = int(fa.get("w", 0) or 0)
233
- h = int(fa.get("h", 0) or 0)
234
- if w > 0 and h > 0 and f.get("face") is not None:
235
- cand.append((f, (x, y, w, h), w * h))
236
-
237
- if not cand:
238
- return None, None
239
-
240
- if prev_bbox is not None:
241
- best = None
242
- best_i = 0.0
243
- for f, bb, area in cand:
244
- i = _iou(prev_bbox, bb)
245
- if i > best_i:
246
- best_i = i
247
- best = (f, bb)
248
- if best is not None and best_i >= IOU_TH:
249
- return best[0], best[1]
250
-
251
- cand.sort(key=lambda x: x[2], reverse=True)
252
- return cand[0][0], cand[0][1]
253
 
 
 
254
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
255
  def _reset_state_for_camera(st: AppState) -> AppState:
256
  st.running = True
257
  st.finished = False
258
  st.last_analyze_t = 0.0
259
- st.track_bbox = None
260
  st.ema_scores = None
261
  st.emo_keys = []
 
262
  st.stable_emo = "unknown"
263
  st.candidate_emo = None
264
  st.candidate_since = None
 
265
  st.final_emo = None
266
  st.final_conf = None
267
  st.final_score = 3
 
 
268
  return st
269
 
270
 
271
  # =========================
272
- # Camera streaming
273
  # =========================
274
  def on_stream(frame_rgb: np.ndarray, st: AppState):
 
275
  if st.finished or (not st.running):
276
  cam_visible = False
277
 
@@ -285,7 +376,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
285
  hint_html = _camera_hint_stopped()
286
 
287
  return (
288
- gr.update(visible=cam_visible),
289
  gr.update(value=result_html),
290
  gr.update(value=mood_val),
291
  gr.update(value=hint_html),
@@ -293,105 +384,108 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
293
  gr.update(value="重新攝影機辨識"),
294
  )
295
 
 
296
  if frame_rgb is None:
297
  return (
298
- gr.update(visible=True),
299
  gr.update(value=_hint_html("等待影像…(尚未開始串流)")),
300
  gr.update(value=_score_radio_value(3)),
301
  gr.update(value=_camera_hint_before_start()),
302
  st,
303
  gr.update(value="重新攝影機辨識"),
304
  )
305
-
306
- frame_rgb = _zoom_center(frame_rgb, ZOOM_FACTOR)
307
 
308
- now = time.time()
309
- if now - st.last_analyze_t < ANALYZE_EVERY_SEC:
310
- show = st.stable_emo or "unknown"
311
- return (
312
- gr.update(visible=True),
313
- gr.update(value=_result_card_html(show, st.final_conf)),
314
- gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
315
- gr.update(value=_camera_hint_before_start()),
316
- st,
317
- gr.update(value="重新攝影機辨識"),
318
- )
319
 
320
- st.last_analyze_t = now
 
321
 
322
- small = _downsample_rgb(frame_rgb, DOWNSAMPLE_W)
 
 
 
 
 
 
 
 
 
 
 
 
323
 
324
- faces = DeepFace.extract_faces(
325
- img_path=small,
326
- detector_backend=DETECTOR_BACKEND,
327
- enforce_detection=False,
328
- align=ALIGN_FACE,
329
- )
330
- pick, bbox = _pick_face_by_tracking(faces, st.track_bbox)
331
- if pick is None or bbox is None or pick.get("face") is None:
332
- show = st.stable_emo or "unknown"
333
  return (
334
- gr.update(visible=True),
335
- gr.update(value=_result_card_html(show, st.final_conf)),
336
- gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
337
- gr.update(value=_camera_hint_before_start()),
338
  st,
339
  gr.update(value="重新攝影機辨識"),
340
  )
341
 
342
- st.track_bbox = bbox
343
- face = pick["face"]
344
- if face.dtype != np.uint8:
345
- face = np.clip(face * 255.0, 0, 255).astype(np.uint8)
346
 
347
- result = DeepFace.analyze(
348
- img_path=face,
349
- actions=["emotion"],
350
- enforce_detection=False,
351
- )
352
- if isinstance(result, list):
353
- result = result[0]
354
- emo_dict = result.get("emotion", None)
 
 
 
 
 
355
 
356
- if not isinstance(emo_dict, dict) or len(emo_dict) == 0:
 
 
 
 
 
 
 
357
  show = st.stable_emo or "unknown"
358
  return (
359
- gr.update(visible=True),
360
- gr.update(value=_result_card_html(show, st.final_conf)),
361
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
362
  gr.update(value=_camera_hint_before_start()),
363
  st,
364
  gr.update(value="重新攝影機辨識"),
365
  )
366
 
367
- top_key = max(emo_dict, key=lambda k: float(emo_dict[k]))
368
- top_conf = float(emo_dict[top_key])
369
-
370
  if top_conf < CONF_TH:
371
  show = st.stable_emo or "unknown"
372
  return (
373
- gr.update(visible=True),
374
- gr.update(value=_result_card_html(show, st.final_conf)),
375
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
376
  gr.update(value=_camera_hint_before_start()),
377
  st,
378
  gr.update(value="重新攝影機辨識"),
379
  )
380
 
381
- keys = list(emo_dict.keys())
382
- vec = np.array([float(emo_dict[k]) for k in keys], dtype=np.float32)
 
 
 
383
 
384
- if st.ema_scores is None or st.emo_keys != keys:
385
  st.ema_scores = vec
386
- st.emo_keys = keys
387
  else:
388
  st.ema_scores = EMA_ALPHA * vec + (1.0 - EMA_ALPHA) * st.ema_scores
389
 
390
  stable_idx = int(np.argmax(st.ema_scores))
391
- stable_key = st.emo_keys[stable_idx]
392
  st.stable_emo = stable_key
393
 
394
- # 延遲確認
395
  if st.candidate_emo != stable_key:
396
  st.candidate_emo = stable_key
397
  st.candidate_since = now
@@ -404,17 +498,18 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
404
  and (now - st.candidate_since) >= SWITCH_CONFIRM_SEC
405
  )
406
 
 
407
  if confirmed and st.final_emo is None:
408
  st.final_emo = stable_key
409
- st.final_conf = top_conf
410
  st.final_score = EMO_TO_SCORE.get(stable_key, 3)
411
 
412
  st.running = False
413
  st.finished = True
414
 
415
  return (
416
- gr.update(visible=False),
417
- gr.update(value=_result_card_html(stable_key, top_conf)),
418
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
419
  gr.update(value=_camera_hint_done()),
420
  st,
@@ -422,8 +517,8 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
422
  )
423
 
424
  return (
425
- gr.update(visible=True),
426
- gr.update(value=_result_card_html(stable_key, top_conf)),
427
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
428
  gr.update(value=_camera_hint_before_start()),
429
  st,
@@ -432,27 +527,26 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
432
 
433
 
434
  # =========================
435
- # Mode / Buttons
436
  # =========================
437
-
438
  def on_restart(st: AppState):
439
  st = _reset_state_for_camera(st)
440
  return (
441
- gr.update(visible=True),
442
- gr.update(value=_hint_html("等待影像…(尚未開始串流)")),
443
- gr.update(value=_score_radio_value(3)),
444
- gr.update(value=_camera_hint_before_start()),
445
  st,
446
- gr.update(value="重新攝影機辨識"),
447
  )
448
 
449
-
450
  def on_stop(st: AppState):
451
  st.running = False
452
  st.finished = True
 
453
  if st.final_emo is None:
454
  return (
455
- gr.update(visible=False),
456
  gr.update(value=_hint_html("已停止辨識。")),
457
  gr.update(value=_score_radio_value(3)),
458
  gr.update(value=_camera_hint_stopped()),
@@ -461,7 +555,7 @@ def on_stop(st: AppState):
461
  )
462
 
463
  return (
464
- gr.update(visible=False),
465
  gr.update(value=_result_card_html(st.final_emo, st.final_conf)),
466
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(st.final_emo, 3))),
467
  gr.update(value=_camera_hint_done()),
@@ -473,20 +567,20 @@ def on_stop(st: AppState):
473
  # =========================
474
  # Gradio UI
475
  # =========================
 
476
 
477
  if __name__ == "__main__":
478
  with gr.Blocks(title="Emotion Detector", css=css) as demo:
479
  st = gr.State(AppState())
480
 
481
  gr.Markdown("## 情緒辨識")
482
-
483
  hint = gr.HTML(_camera_hint_before_start())
484
 
485
  cam = gr.Image(
486
  sources=["webcam"],
487
  streaming=True,
488
  type="numpy",
489
- label="攝影機",
490
  visible=True,
491
  )
492
 
@@ -526,4 +620,4 @@ if __name__ == "__main__":
526
  show_progress="minimal",
527
  )
528
 
529
- demo.launch(server_name="0.0.0.0", server_port=7860, share=False)
 
1
  import time
2
+ import threading
3
  from dataclasses import dataclass, field
4
+ from typing import Optional, List, Tuple, Dict
5
 
6
  import gradio as gr
7
  import numpy as np
8
+ import cv2
9
+
10
+ import mediapipe as mp
11
+ from mediapipe.tasks import python
12
+ from mediapipe.tasks.python import vision
13
 
14
  from deepface import DeepFace
15
 
 
17
  # =========================
18
  # 參數(可微調)
19
  # =========================
20
+ ANALYZE_EVERY_SEC = 1.5 # ⭐ 建議 1.5~2.0:HF CPU 比較順
 
21
  EMA_ALPHA = 0.45
22
+ CONF_TH = 40.0 # DeepFace emotion confidence 門檻
23
+ SWITCH_CONFIRM_SEC = 2.5 # 連續穩定多久才算完成
24
+ ZOOM_FACTOR = 1.5 # ✅ 保留:只用在「顯示」
25
+ UI_TARGET_W = 720 # ✅ 回傳畫面縮小寬度,提升順暢度
26
+
27
+ # MediaPipe face detector 門檻(太高很容易漏)
28
+ MP_MIN_DET_CONF = 0.3
29
 
30
+ # 模型路徑(確認 repo 中模型真的在這裡)
31
+ MP_MODEL_PATH = "blaze_face_short_range.tflite"
32
 
33
 
34
  # =========================
 
45
  "unknown": "未知",
46
  }
47
 
 
48
  EMO_TO_SCORE = {
49
  "angry": 1,
50
  "disgust": 1,
 
56
  "unknown": 3,
57
  }
58
 
 
 
 
59
  SCORE_LABELS = {
60
  1: "1(心情差)",
61
  2: "2(不太好)",
 
71
  # =========================
72
  @dataclass
73
  class AppState:
74
+ mode: str = "攝影機辨識"
75
 
76
  running: bool = True
77
  finished: bool = False
78
  last_analyze_t: float = 0.0
79
 
 
 
 
80
  # smoothing
81
  ema_scores: Optional[np.ndarray] = None
82
  emo_keys: List[str] = field(default_factory=list)
 
89
  # final
90
  final_emo: Optional[str] = None
91
  final_conf: Optional[float] = None
92
+ final_score: int = 3
 
93
 
94
 
95
  # =========================
96
+ # UI HTML
97
  # =========================
98
  def _hint_html(msg: str) -> str:
 
99
  return f"""
100
  <div style="border-radius:14px;padding:14px;border:1px dashed #DADADA;background:#FAFAFA;color:#000000;">
101
  {msg}
102
  </div>
103
  """
104
 
 
105
  def _camera_hint_before_start() -> str:
106
  return _hint_html(
107
  "請先按「Click to access webcam / 允許」開啟攝影機;"
108
  "接著,請按「錄製(Record)」開始。"
109
  )
110
 
111
+ def _camera_hint_need_face() -> str:
112
+ return _hint_html(
113
+ "⚠️ 目前未偵測到人臉,請將臉部移至畫面中央、靠近鏡頭,並避免瀏海遮住眉眼,才可進行情緒辨識。"
114
+ )
115
 
116
  def _camera_hint_done() -> str:
117
  return _hint_html("已完成辨識,可按「重新攝影機辨識」再次進行辨識。")
118
 
 
119
  def _camera_hint_stopped() -> str:
120
  return _hint_html("已停止辨識。可按「重新攝影機辨識」再試一次。")
121
 
122
+ def _score_radio_value(score: int) -> str:
123
+ score = int(score)
124
+ score = min(5, max(1, score))
125
+ return SCORE_LABELS[score]
126
 
127
  def _result_card_html(emo_key: str, conf: Optional[float]) -> str:
128
  zh = EMO_ZH.get(emo_key, emo_key)
 
130
  conf_txt = "" if conf is None else f"{conf:.1f}%"
131
  score_desc = SCORE_LABELS[score].split("(")[1].rstrip(")")
132
 
 
133
  return f"""
134
  <div style="
135
  border-radius:16px;
136
  padding:20px;
137
  border:1px solid #E6E6E6;
138
  background:#FFFFFF;
139
+ color:#000000;
140
  ">
141
  <div style="font-size:14px;color:#000000;margin-bottom:8px;">
142
  辨識結果
 
163
  """
164
 
165
 
166
+ # =========================
167
+ # 影像處理(保命 + 顯示優化)
168
+ # =========================
169
+ def _to_rgb_uint8(frame: np.ndarray) -> np.ndarray:
170
+ """統一成 MediaPipe Tasks 最穩格式:RGB、uint8、0~255、contiguous。"""
 
 
171
  if frame is None:
 
 
 
172
  return frame
173
+
174
+ # RGBA -> RGB
175
+ if frame.ndim == 3 and frame.shape[2] == 4:
176
+ frame = frame[:, :, :3]
177
+
178
+ # 灰階 -> RGB
179
+ if frame.ndim == 2:
180
+ frame = np.stack([frame, frame, frame], axis=-1)
181
+
182
+ # dtype / range 修正
183
+ if frame.dtype != np.uint8:
184
+ mx = float(np.max(frame)) if frame.size else 0.0
185
+ if mx <= 1.5: # 多半是 0~1
186
+ frame = (frame * 255.0).clip(0, 255).astype(np.uint8)
187
+ else: # 多半是 0~255 但 float
188
+ frame = np.clip(frame, 0, 255).astype(np.uint8)
189
+
190
+ return np.ascontiguousarray(frame)
191
 
192
  def _zoom_center(frame: np.ndarray, zoom: float) -> np.ndarray:
 
 
 
 
193
  if frame is None or zoom <= 1.0:
194
  return frame
 
195
  h, w = frame.shape[:2]
 
 
196
  new_w = int(w / zoom)
197
  new_h = int(h / zoom)
 
 
198
  start_x = (w - new_w) // 2
199
  start_y = (h - new_h) // 2
 
 
200
  cropped = frame[start_y : start_y + new_h, start_x : start_x + new_w]
201
+ return np.ascontiguousarray(cropped)
202
+
203
+ def _resize_for_ui(frame_rgb: np.ndarray, target_w: int = UI_TARGET_W) -> np.ndarray:
204
+ """回傳給 Gradio 的畫面縮到固定寬度,減少傳輸/渲染負擔。"""
205
+ if frame_rgb is None:
206
+ return frame_rgb
207
+ h, w = frame_rgb.shape[:2]
208
+ if w <= target_w:
209
+ return frame_rgb
210
+ scale = target_w / w
211
+ new_h = max(1, int(h * scale))
212
+ return cv2.resize(frame_rgb, (target_w, new_h), interpolation=cv2.INTER_AREA)
213
+
214
+
215
+ # =========================
216
+ # MediaPipe Tasks Face Detector
217
+ # =========================
218
+ BaseOptions = python.BaseOptions
219
+ FaceDetector = vision.FaceDetector
220
+ FaceDetectorOptions = vision.FaceDetectorOptions
221
+ VisionRunningMode = vision.RunningMode
222
+
223
+ _face_detector = FaceDetector.create_from_options(
224
+ FaceDetectorOptions(
225
+ base_options=BaseOptions(model_asset_path=MP_MODEL_PATH),
226
+ running_mode=VisionRunningMode.IMAGE,
227
+ min_detection_confidence=MP_MIN_DET_CONF,
228
+ )
229
+ )
230
+
231
+ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
232
+ """
233
+ - 最大臉策略(bbox 面積最大)
234
+ - 有臉:畫白框 + 紅點 keypoints
235
+ - 回 bbox_px (x, y, w, h) 供 ROI 裁切
236
+ """
237
+ frame_rgb = _to_rgb_uint8(frame_rgb)
238
+ h, w = frame_rgb.shape[:2]
239
+
240
+ mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=frame_rgb)
241
+ result = _face_detector.detect(mp_image)
 
 
 
 
 
 
 
 
 
 
 
242
 
243
+ if not result.detections:
244
+ return frame_rgb, None
245
 
246
+ best_det = None
247
+ best_area = -1.0
248
+ best_bbox = None
249
+
250
+ for det in result.detections:
251
+ bbox = det.bounding_box
252
+ area = float(bbox.width) * float(bbox.height)
253
+ if area > best_area:
254
+ best_area = area
255
+ best_det = det
256
+ best_bbox = bbox
257
+
258
+ if best_det is None or best_bbox is None:
259
+ return frame_rgb, None
260
+
261
+ lx = int(best_bbox.origin_x)
262
+ ly = int(best_bbox.origin_y)
263
+ bw = int(best_bbox.width)
264
+ bh = int(best_bbox.height)
265
+
266
+ lx = max(0, min(w - 1, lx))
267
+ ly = max(0, min(h - 1, ly))
268
+ bw = max(1, min(w - lx, bw))
269
+ bh = max(1, min(h - ly, bh))
270
+
271
+ bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)
272
+ cv2.rectangle(bgr, (lx, ly), (lx + bw, ly + bh), (255, 255, 255), 3)
273
+
274
+ for kp in best_det.keypoints:
275
+ cx = int(kp.x * w)
276
+ cy = int(kp.y * h)
277
+ cv2.circle(bgr, (cx, cy), 6, (0, 0, 255), -1)
278
+
279
+ drawn = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
280
+ return drawn, (lx, ly, bw, bh)
281
+
282
+
283
+ # =========================
284
+ # 背景推論(避免卡頓) + ✅ 沒臉時清空
285
+ # =========================
286
+ _ANALYZE_LOCK = threading.Lock()
287
+ _ANALYZE_THREAD: Optional[threading.Thread] = None
288
+ _PENDING_FACE: Optional[np.ndarray] = None
289
+ _LAST_ANALYZE: Dict[str, Optional[float]] = {"ts": 0.0, "conf": None, "emo": "unknown"}
290
+
291
+ def _clear_last_analyze():
292
+ """✅ 沒有人臉時,清空上一筆情緒,避免殘留顯示(你影片遇到的 bug)"""
293
+ global _LAST_ANALYZE
294
+ _LAST_ANALYZE = {"ts": 0.0, "conf": None, "emo": "unknown"}
295
+
296
+ def _analyze_worker():
297
+ global _PENDING_FACE, _ANALYZE_THREAD, _LAST_ANALYZE
298
+
299
+ while True:
300
+ with _ANALYZE_LOCK:
301
+ face = _PENDING_FACE
302
+ _PENDING_FACE = None
303
+
304
+ if face is None:
305
+ break
306
+
307
+ try:
308
+ r = DeepFace.analyze(
309
+ img_path=face,
310
+ actions=["emotion"],
311
+ enforce_detection=False,
312
+ detector_backend="skip", # ⭐ 你已裁 ROI,別再偵測一次
313
+ )
314
+ if isinstance(r, list):
315
+ r = r[0]
316
+ emo_dict = r.get("emotion", None)
317
+
318
+ if isinstance(emo_dict, dict) and len(emo_dict) > 0:
319
+ top_key = max(emo_dict, key=lambda k: float(emo_dict[k]))
320
+ top_conf = float(emo_dict[top_key])
321
+ _LAST_ANALYZE = {"ts": time.time(), "emo": top_key, "conf": top_conf}
322
+ except Exception:
323
+ pass
324
+
325
+ _ANALYZE_THREAD = None
326
+
327
+ def _enqueue_analyze(face_roi: np.ndarray):
328
+ global _PENDING_FACE, _ANALYZE_THREAD
329
+
330
+ with _ANALYZE_LOCK:
331
+ _PENDING_FACE = face_roi
332
+
333
+ if _ANALYZE_THREAD is None:
334
+ _ANALYZE_THREAD = threading.Thread(target=_analyze_worker, daemon=True)
335
+ _ANALYZE_THREAD.start()
336
+
337
+
338
+ # =========================
339
+ # state helper
340
+ # =========================
341
  def _reset_state_for_camera(st: AppState) -> AppState:
342
  st.running = True
343
  st.finished = False
344
  st.last_analyze_t = 0.0
345
+
346
  st.ema_scores = None
347
  st.emo_keys = []
348
+
349
  st.stable_emo = "unknown"
350
  st.candidate_emo = None
351
  st.candidate_since = None
352
+
353
  st.final_emo = None
354
  st.final_conf = None
355
  st.final_score = 3
356
+
357
+ _clear_last_analyze()
358
  return st
359
 
360
 
361
  # =========================
362
+ # streaming callback
363
  # =========================
364
  def on_stream(frame_rgb: np.ndarray, st: AppState):
365
+ # stopped / finished
366
  if st.finished or (not st.running):
367
  cam_visible = False
368
 
 
376
  hint_html = _camera_hint_stopped()
377
 
378
  return (
379
+ gr.update(value=None, visible=cam_visible),
380
  gr.update(value=result_html),
381
  gr.update(value=mood_val),
382
  gr.update(value=hint_html),
 
384
  gr.update(value="重新攝影機辨識"),
385
  )
386
 
387
+ # no frame yet
388
  if frame_rgb is None:
389
  return (
390
+ gr.update(value=None, visible=True),
391
  gr.update(value=_hint_html("等待影像…(尚未開始串流)")),
392
  gr.update(value=_score_radio_value(3)),
393
  gr.update(value=_camera_hint_before_start()),
394
  st,
395
  gr.update(value="重新攝影機辨識"),
396
  )
 
 
397
 
398
+ frame_rgb = _to_rgb_uint8(frame_rgb)
 
 
 
 
 
 
 
 
 
 
399
 
400
+ # ✅ 偵測用原圖
401
+ frame_drawn, bbox_px = mp_tasks_detect_and_draw(frame_rgb)
402
 
403
+ # ✅ 顯示用 zoom
404
+ frame_show = _zoom_center(frame_drawn if bbox_px is not None else frame_rgb, ZOOM_FACTOR)
405
+ frame_show = _resize_for_ui(frame_show, UI_TARGET_W)
406
+
407
+ # ✅ 沒有臉:清空殘留情緒 + 重置候選狀態
408
+ if bbox_px is None:
409
+ _clear_last_analyze()
410
+
411
+ st.stable_emo = "unknown"
412
+ st.candidate_emo = None
413
+ st.candidate_since = None
414
+ st.ema_scores = None
415
+ st.emo_keys = []
416
 
 
 
 
 
 
 
 
 
 
417
  return (
418
+ gr.update(value=frame_show, visible=True),
419
+ gr.update(value=_hint_html("尚未偵測到人臉")),
420
+ gr.update(value=_score_radio_value(3)),
421
+ gr.update(value=_camera_hint_need_face()),
422
  st,
423
  gr.update(value="重新攝影機辨識"),
424
  )
425
 
426
+ now = time.time()
 
 
 
427
 
428
+ # 到分析時間:只排隊給背景 thread
429
+ if now - st.last_analyze_t >= ANALYZE_EVERY_SEC:
430
+ st.last_analyze_t = now
431
+
432
+ x, y, bw, bh = bbox_px
433
+ pad_x = int(bw * 0.10)
434
+ pad_y = int(bh * 0.10)
435
+ H, W = frame_rgb.shape[:2]
436
+ x1 = max(0, x - pad_x)
437
+ y1 = max(0, y - pad_y)
438
+ x2 = min(W, x + bw + pad_x)
439
+ y2 = min(H, y + bh + pad_y)
440
+ face_roi = frame_rgb[y1:y2, x1:x2].astype(np.uint8)
441
 
442
+ _enqueue_analyze(face_roi)
443
+
444
+ # ✅ Gate:只有「有臉」才讀取 _LAST_ANALYZE(避免殘留)
445
+ top_key = _LAST_ANALYZE.get("emo", "unknown") or "unknown"
446
+ top_conf = _LAST_ANALYZE.get("conf", None)
447
+
448
+ # 尚未推論到結果:先顯示 unknown
449
+ if top_conf is None:
450
  show = st.stable_emo or "unknown"
451
  return (
452
+ gr.update(value=frame_show, visible=True),
453
+ gr.update(value=_result_card_html(show, None)),
454
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
455
  gr.update(value=_camera_hint_before_start()),
456
  st,
457
  gr.update(value="重新攝影機辨識"),
458
  )
459
 
460
+ # 低信心就不切換
 
 
461
  if top_conf < CONF_TH:
462
  show = st.stable_emo or "unknown"
463
  return (
464
+ gr.update(value=frame_show, visible=True),
465
+ gr.update(value=_result_card_html(show, None)),
466
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
467
  gr.update(value=_camera_hint_before_start()),
468
  st,
469
  gr.update(value="重新攝影機辨識"),
470
  )
471
 
472
+ # === EMA smoothing(用 7 類別向量)===
473
+ all_keys = ["angry", "disgust", "fear", "happy", "sad", "surprise", "neutral"]
474
+ vec = np.zeros((len(all_keys),), dtype=np.float32)
475
+ if top_key in all_keys:
476
+ vec[all_keys.index(top_key)] = float(top_conf)
477
 
478
+ if st.ema_scores is None or st.emo_keys != all_keys:
479
  st.ema_scores = vec
480
+ st.emo_keys = all_keys
481
  else:
482
  st.ema_scores = EMA_ALPHA * vec + (1.0 - EMA_ALPHA) * st.ema_scores
483
 
484
  stable_idx = int(np.argmax(st.ema_scores))
485
+ stable_key = st.emo_keys[stable_idx] if st.emo_keys else "unknown"
486
  st.stable_emo = stable_key
487
 
488
+ # 延遲確認,避免跳動
489
  if st.candidate_emo != stable_key:
490
  st.candidate_emo = stable_key
491
  st.candidate_since = now
 
498
  and (now - st.candidate_since) >= SWITCH_CONFIRM_SEC
499
  )
500
 
501
+ # ✅ 完成一次辨識後停止
502
  if confirmed and st.final_emo is None:
503
  st.final_emo = stable_key
504
+ st.final_conf = float(top_conf) if top_conf is not None else None
505
  st.final_score = EMO_TO_SCORE.get(stable_key, 3)
506
 
507
  st.running = False
508
  st.finished = True
509
 
510
  return (
511
+ gr.update(value=None, visible=False),
512
+ gr.update(value=_result_card_html(stable_key, st.final_conf)),
513
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
514
  gr.update(value=_camera_hint_done()),
515
  st,
 
517
  )
518
 
519
  return (
520
+ gr.update(value=frame_show, visible=True),
521
+ gr.update(value=_result_card_html(stable_key, float(top_conf))),
522
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(stable_key, 3))),
523
  gr.update(value=_camera_hint_before_start()),
524
  st,
 
527
 
528
 
529
  # =========================
530
+ # Buttons
531
  # =========================
 
532
  def on_restart(st: AppState):
533
  st = _reset_state_for_camera(st)
534
  return (
535
+ gr.update(value=None, visible=True),
536
+ gr.update(value=_hint_html("等待影像…(尚未開始串流)")),
537
+ gr.update(value=_score_radio_value(3)),
538
+ gr.update(value=_camera_hint_before_start()),
539
  st,
540
+ gr.update(value="重新攝影機辨識"),
541
  )
542
 
 
543
  def on_stop(st: AppState):
544
  st.running = False
545
  st.finished = True
546
+
547
  if st.final_emo is None:
548
  return (
549
+ gr.update(value=None, visible=False),
550
  gr.update(value=_hint_html("已停止辨識。")),
551
  gr.update(value=_score_radio_value(3)),
552
  gr.update(value=_camera_hint_stopped()),
 
555
  )
556
 
557
  return (
558
+ gr.update(value=None, visible=False),
559
  gr.update(value=_result_card_html(st.final_emo, st.final_conf)),
560
  gr.update(value=_score_radio_value(EMO_TO_SCORE.get(st.final_emo, 3))),
561
  gr.update(value=_camera_hint_done()),
 
567
  # =========================
568
  # Gradio UI
569
  # =========================
570
+ css = ""
571
 
572
  if __name__ == "__main__":
573
  with gr.Blocks(title="Emotion Detector", css=css) as demo:
574
  st = gr.State(AppState())
575
 
576
  gr.Markdown("## 情緒辨識")
 
577
  hint = gr.HTML(_camera_hint_before_start())
578
 
579
  cam = gr.Image(
580
  sources=["webcam"],
581
  streaming=True,
582
  type="numpy",
583
+ label="攝影機畫面",
584
  visible=True,
585
  )
586
 
 
620
  show_progress="minimal",
621
  )
622
 
623
+ demo.launch(server_name="0.0.0.0", server_port=7860, share=False)