Update emotion.py
Browse files- emotion.py +32 -59
emotion.py
CHANGED
|
@@ -5,9 +5,8 @@ from typing import Optional, List, Tuple, Dict
|
|
| 5 |
|
| 6 |
import gradio as gr
|
| 7 |
import numpy as np
|
| 8 |
-
from PIL import Image
|
| 9 |
-
|
| 10 |
import cv2
|
|
|
|
| 11 |
import mediapipe as mp
|
| 12 |
from mediapipe.tasks import python
|
| 13 |
from mediapipe.tasks.python import vision
|
|
@@ -18,7 +17,7 @@ from deepface import DeepFace
|
|
| 18 |
# =========================
|
| 19 |
# 參數(可微調)
|
| 20 |
# =========================
|
| 21 |
-
ANALYZE_EVERY_SEC =
|
| 22 |
EMA_ALPHA = 0.45
|
| 23 |
CONF_TH = 40.0 # DeepFace emotion confidence 門檻
|
| 24 |
SWITCH_CONFIRM_SEC = 2.5 # 連續穩定多久才算完成
|
|
@@ -29,7 +28,7 @@ UI_TARGET_W = 720 # ✅ 回傳畫面縮小寬度,提升順暢
|
|
| 29 |
MP_MIN_DET_CONF = 0.3
|
| 30 |
|
| 31 |
# 模型路徑(確認 repo 中模型真的在這裡)
|
| 32 |
-
MP_MODEL_PATH = "blaze_face_short_range.tflite"
|
| 33 |
|
| 34 |
|
| 35 |
# =========================
|
|
@@ -248,7 +247,6 @@ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
|
|
| 248 |
best_area = -1.0
|
| 249 |
best_bbox = None
|
| 250 |
|
| 251 |
-
# 最大臉:bbox 面積最大
|
| 252 |
for det in result.detections:
|
| 253 |
bbox = det.bounding_box
|
| 254 |
area = float(bbox.width) * float(bbox.height)
|
|
@@ -265,18 +263,14 @@ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
|
|
| 265 |
bw = int(best_bbox.width)
|
| 266 |
bh = int(best_bbox.height)
|
| 267 |
|
| 268 |
-
# clamp
|
| 269 |
lx = max(0, min(w - 1, lx))
|
| 270 |
ly = max(0, min(h - 1, ly))
|
| 271 |
bw = max(1, min(w - lx, bw))
|
| 272 |
bh = max(1, min(h - ly, bh))
|
| 273 |
|
| 274 |
bgr = cv2.cvtColor(frame_rgb, cv2.COLOR_RGB2BGR)
|
| 275 |
-
|
| 276 |
-
# 白框
|
| 277 |
cv2.rectangle(bgr, (lx, ly), (lx + bw, ly + bh), (255, 255, 255), 3)
|
| 278 |
|
| 279 |
-
# 紅點:五官特徵點(相對座標)
|
| 280 |
for kp in best_det.keypoints:
|
| 281 |
cx = int(kp.x * w)
|
| 282 |
cy = int(kp.y * h)
|
|
@@ -287,15 +281,19 @@ def mp_tasks_detect_and_draw(frame_rgb: np.ndarray):
|
|
| 287 |
|
| 288 |
|
| 289 |
# =========================
|
| 290 |
-
# 背景推論(避免卡頓)
|
| 291 |
# =========================
|
| 292 |
_ANALYZE_LOCK = threading.Lock()
|
| 293 |
_ANALYZE_THREAD: Optional[threading.Thread] = None
|
| 294 |
_PENDING_FACE: Optional[np.ndarray] = None
|
| 295 |
_LAST_ANALYZE: Dict[str, Optional[float]] = {"ts": 0.0, "conf": None, "emo": "unknown"}
|
| 296 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 297 |
def _analyze_worker():
|
| 298 |
-
"""背景 thread:取最新 face_roi 做 DeepFace.analyze,更新 _LAST_ANALYZE。"""
|
| 299 |
global _PENDING_FACE, _ANALYZE_THREAD, _LAST_ANALYZE
|
| 300 |
|
| 301 |
while True:
|
|
@@ -311,7 +309,7 @@ def _analyze_worker():
|
|
| 311 |
img_path=face,
|
| 312 |
actions=["emotion"],
|
| 313 |
enforce_detection=False,
|
| 314 |
-
detector_backend="skip", # ⭐
|
| 315 |
)
|
| 316 |
if isinstance(r, list):
|
| 317 |
r = r[0]
|
|
@@ -322,52 +320,21 @@ def _analyze_worker():
|
|
| 322 |
top_conf = float(emo_dict[top_key])
|
| 323 |
_LAST_ANALYZE = {"ts": time.time(), "emo": top_key, "conf": top_conf}
|
| 324 |
except Exception:
|
| 325 |
-
# 失敗就保持上一次結果
|
| 326 |
pass
|
| 327 |
|
| 328 |
_ANALYZE_THREAD = None
|
| 329 |
|
| 330 |
-
|
| 331 |
-
def _resize_letterbox_rgb(img: np.ndarray, size: int = 224) -> np.ndarray:
|
| 332 |
-
"""
|
| 333 |
-
把 ROI 變成 size×size,保持比例,用黑邊補滿(letterbox)。
|
| 334 |
-
好處:不會把臉拉伸變形,對情緒分類通常更穩。
|
| 335 |
-
"""
|
| 336 |
-
img = _to_rgb_uint8(img)
|
| 337 |
-
h, w = img.shape[:2]
|
| 338 |
-
if h <= 0 or w <= 0:
|
| 339 |
-
return img
|
| 340 |
-
|
| 341 |
-
scale = size / max(h, w)
|
| 342 |
-
nh, nw = int(h * scale), int(w * scale)
|
| 343 |
-
nh = max(1, nh)
|
| 344 |
-
nw = max(1, nw)
|
| 345 |
-
|
| 346 |
-
resized = cv2.resize(img, (nw, nh), interpolation=cv2.INTER_AREA)
|
| 347 |
-
|
| 348 |
-
out = np.zeros((size, size, 3), dtype=np.uint8)
|
| 349 |
-
y0 = (size - nh) // 2
|
| 350 |
-
x0 = (size - nw) // 2
|
| 351 |
-
out[y0:y0+nh, x0:x0+nw] = resized
|
| 352 |
-
return out
|
| 353 |
-
|
| 354 |
-
|
| 355 |
def _enqueue_analyze(face_roi: np.ndarray):
|
| 356 |
-
"""把最新 ROI 丟給背景 thread,若 thread 沒在跑就啟動。"""
|
| 357 |
global _PENDING_FACE, _ANALYZE_THREAD
|
| 358 |
|
| 359 |
-
# ✅ 這行是你要的:先把 ROI 縮到固定大小(224×224)
|
| 360 |
-
face_small = _resize_letterbox_rgb(face_roi, size=224)
|
| 361 |
-
|
| 362 |
with _ANALYZE_LOCK:
|
| 363 |
-
_PENDING_FACE =
|
| 364 |
|
| 365 |
if _ANALYZE_THREAD is None:
|
| 366 |
_ANALYZE_THREAD = threading.Thread(target=_analyze_worker, daemon=True)
|
| 367 |
_ANALYZE_THREAD.start()
|
| 368 |
|
| 369 |
|
| 370 |
-
|
| 371 |
# =========================
|
| 372 |
# state helper
|
| 373 |
# =========================
|
|
@@ -386,6 +353,8 @@ def _reset_state_for_camera(st: AppState) -> AppState:
|
|
| 386 |
st.final_emo = None
|
| 387 |
st.final_conf = None
|
| 388 |
st.final_score = 3
|
|
|
|
|
|
|
| 389 |
return st
|
| 390 |
|
| 391 |
|
|
@@ -426,18 +395,25 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
|
|
| 426 |
gr.update(value="重新攝影機辨識"),
|
| 427 |
)
|
| 428 |
|
| 429 |
-
# 統���格式
|
| 430 |
frame_rgb = _to_rgb_uint8(frame_rgb)
|
| 431 |
|
| 432 |
-
# ✅ 偵測用原圖
|
| 433 |
frame_drawn, bbox_px = mp_tasks_detect_and_draw(frame_rgb)
|
| 434 |
|
| 435 |
-
# ✅ 顯示用 zoom
|
| 436 |
frame_show = _zoom_center(frame_drawn if bbox_px is not None else frame_rgb, ZOOM_FACTOR)
|
| 437 |
frame_show = _resize_for_ui(frame_show, UI_TARGET_W)
|
| 438 |
|
| 439 |
-
# 沒有
|
| 440 |
if bbox_px is None:
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 441 |
return (
|
| 442 |
gr.update(value=frame_show, visible=True),
|
| 443 |
gr.update(value=_hint_html("尚未偵測到人臉")),
|
|
@@ -449,7 +425,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
|
|
| 449 |
|
| 450 |
now = time.time()
|
| 451 |
|
| 452 |
-
# 到分析時間:只排隊給背景 thread
|
| 453 |
if now - st.last_analyze_t >= ANALYZE_EVERY_SEC:
|
| 454 |
st.last_analyze_t = now
|
| 455 |
|
|
@@ -465,38 +441,35 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
|
|
| 465 |
|
| 466 |
_enqueue_analyze(face_roi)
|
| 467 |
|
| 468 |
-
#
|
| 469 |
top_key = _LAST_ANALYZE.get("emo", "unknown") or "unknown"
|
| 470 |
top_conf = _LAST_ANALYZE.get("conf", None)
|
| 471 |
|
| 472 |
-
#
|
| 473 |
if top_conf is None:
|
| 474 |
show = st.stable_emo or "unknown"
|
| 475 |
return (
|
| 476 |
gr.update(value=frame_show, visible=True),
|
| 477 |
-
gr.update(value=_result_card_html(show,
|
| 478 |
gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
|
| 479 |
gr.update(value=_camera_hint_before_start()),
|
| 480 |
st,
|
| 481 |
gr.update(value="重新攝影機辨識"),
|
| 482 |
)
|
| 483 |
|
| 484 |
-
# 低信心就不切換
|
| 485 |
if top_conf < CONF_TH:
|
| 486 |
show = st.stable_emo or "unknown"
|
| 487 |
return (
|
| 488 |
gr.update(value=frame_show, visible=True),
|
| 489 |
-
gr.update(value=_result_card_html(show,
|
| 490 |
gr.update(value=_score_radio_value(EMO_TO_SCORE.get(show, 3))),
|
| 491 |
gr.update(value=_camera_hint_before_start()),
|
| 492 |
st,
|
| 493 |
gr.update(value="重新攝影機辨識"),
|
| 494 |
)
|
| 495 |
|
| 496 |
-
# === EMA smoothing ===
|
| 497 |
-
# DeepFace emotion dict 的 keys 順序可能固定,但我們這裡用單一 top_conf/top_key 也能做穩定
|
| 498 |
-
# 仍然保留你原本的「看整個分佈」的 smoothing:用 top_key/top_conf 建一個簡化 dict
|
| 499 |
-
# 為了兼容你原本寫法,我們用 7 類別做一個向量(缺的填 0)
|
| 500 |
all_keys = ["angry", "disgust", "fear", "happy", "sad", "surprise", "neutral"]
|
| 501 |
vec = np.zeros((len(all_keys),), dtype=np.float32)
|
| 502 |
if top_key in all_keys:
|
|
@@ -525,7 +498,7 @@ def on_stream(frame_rgb: np.ndarray, st: AppState):
|
|
| 525 |
and (now - st.candidate_since) >= SWITCH_CONFIRM_SEC
|
| 526 |
)
|
| 527 |
|
| 528 |
-
# ✅
|
| 529 |
if confirmed and st.final_emo is None:
|
| 530 |
st.final_emo = stable_key
|
| 531 |
st.final_conf = float(top_conf) if top_conf is not None else None
|
|
|
|
| 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
|
|
|
|
| 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 # 連續穩定多久才算完成
|
|
|
|
| 28 |
MP_MIN_DET_CONF = 0.3
|
| 29 |
|
| 30 |
# 模型路徑(確認 repo 中模型真的在這裡)
|
| 31 |
+
MP_MODEL_PATH = "model/blaze_face_short_range.tflite"
|
| 32 |
|
| 33 |
|
| 34 |
# =========================
|
|
|
|
| 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)
|
|
|
|
| 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)
|
|
|
|
| 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:
|
|
|
|
| 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]
|
|
|
|
| 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 |
# =========================
|
|
|
|
| 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 |
|
|
|
|
| 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("尚未偵測到人臉")),
|
|
|
|
| 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 |
|
|
|
|
| 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:
|
|
|
|
| 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
|