feat(vision): YOLO 人脸/姿态识别与手势控制（后端推理）
- vision_yolo：yolov8n-face 人脸检测 + yolov8n-pose 姿态估计（torch/ultralytics，懒加载常驻）
- 单帧几何特征：举手（腕高过肩）与举拳（前臂上举收胸前），返回 raised/fists
- POST /api/vision/frame：前端抽帧 JPEG → 推理 → {faces, pose, raised, fists, latency_ms}，线程池执行
- POST /api/vision/event：状态上报，triggered 时经 MQTT 广播 alert
- 模型入库：backend/models/yolov8n-face.pt + yolov8n-pose.pt
- 前端 hook 改为网络版：抽帧→POST→状态机（面向10s问候/举手toggle对话），含亮度/耗时/链路诊断
This commit is contained in:
Pine
@@ -387,3 +387,42 @@ async def sse_events(request: Request):
|
||||
bus.unsubscribe(q)
|
||||
|
||||
return StreamingResponse(gen(), media_type="text/event-stream")
|
||||
|
||||
|
||||
# ==================== 视觉识别事件上报（前端摄像头实时识别 → 后端日志/MQTT） ====================
|
||||
class VisionEventBody(BaseModel):
|
||||
event: str # camera_on|camera_off|detecting|facing|triggered|silent|error|model_error
|
||||
faces: int = 0
|
||||
dwell_ms: int = 0
|
||||
detail: str = ""
|
||||
|
||||
@router.post("/api/vision/event")
|
||||
async def vision_event(body: VisionEventBody):
|
||||
"""前端摄像头识别状态/触发上报；后端记录详细日志，triggered 时广播 alert 到全屏。"""
|
||||
log.info(
|
||||
"vision event=%s faces=%d dwell_ms=%d detail=%s",
|
||||
body.event, body.faces, body.dwell_ms, body.detail,
|
||||
)
|
||||
if body.event == "triggered":
|
||||
hub.publish_command("alert", {"text": "有访客正对屏幕，语音助手已主动问候", "faces": body.faces})
|
||||
log.info("vision triggered -> alert 已广播（faces=%d）", body.faces)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
# ==================== YOLO 人脸检测（前端抽帧 → 后端推理） ====================
|
||||
class VisionFrameBody(BaseModel):
|
||||
image: str = "" # JPEG base64（不含 data: 前缀）
|
||||
conf: float = 0.0 # 可选：覆盖置信度阈值
|
||||
|
||||
@router.post("/api/vision/frame")
|
||||
def vision_frame(body: VisionFrameBody):
|
||||
"""接收前端抽帧 JPEG base64，后端 YOLO 推理返回人脸框（faces/boxes/latency_ms）。
|
||||
|
||||
普通 def 由 FastAPI 线程池执行（推理约 100-200ms），不阻塞事件循环。
|
||||
"""
|
||||
img_b64 = (body.image or "").strip()
|
||||
if not img_b64:
|
||||
log.warning("vision frame: 缺少 image")
|
||||
return {"ok": False, "error": "missing image"}
|
||||
from .vision_yolo import predict_base64
|
||||
return predict_base64(img_b64)
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""YOLO 人脸检测 + 姿态估计（后端推理）
|
||||
|
||||
链路：前端摄像头抽帧 → JPEG base64 → POST /api/vision/frame → 本模块推理 → 返回人脸/姿态/举手特征。
|
||||
模型：
|
||||
- backend/models/yolov8n-face.pt （lindevs/yolov8-face，WIDERFACE 训练，torch 6.3MB）
|
||||
- backend/models/yolov8n-pose.pt （ultralytics 官方，COCO 17 关键点，torch 6.8MB）
|
||||
推理：ultralytics 加载（torch 原生，自动 letterbox + NMS），CPU 设备。
|
||||
|
||||
手势判定分工：
|
||||
- 后端只做"单帧几何特征"（举手 = 手腕高于肩），无状态
|
||||
- 前端做时序判定（举手持续 1.5s → 开始对话；举手后手腕摆动 → 挥手结束对话）
|
||||
"""
|
||||
import base64
|
||||
import io
|
||||
import logging
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image
|
||||
|
||||
log = logging.getLogger("dpm.vision")
|
||||
|
||||
MODEL_DIR = Path(__file__).resolve().parent.parent / "models"
|
||||
FACE_MODEL_PATH = MODEL_DIR / "yolov8n-face.pt"
|
||||
POSE_MODEL_PATH = MODEL_DIR / "yolov8n-pose.pt"
|
||||
CONF_THRESHOLD = 0.25 # 远距离小脸可调低（如 0.15）
|
||||
IMGSZ = 640
|
||||
|
||||
# COCO 17 关键点索引
|
||||
KP_LEFT_SHOULDER = 5
|
||||
KP_RIGHT_SHOULDER = 6
|
||||
KP_LEFT_ELBOW = 7
|
||||
KP_RIGHT_ELBOW = 8
|
||||
KP_LEFT_WRIST = 9
|
||||
KP_RIGHT_WRIST = 10
|
||||
RAISE_LIFT = 0.12 # 举手：手腕高于对应肩 12% 画面高
|
||||
FIST_ELBOW_LIFT = 0.05 # 举拳：手腕高于肘 5% 画面高（前臂上举）
|
||||
FIST_SHOULDER_GAP = 0.15 # 举拳：拳不高于肩 15% 画面高（收在胸前）
|
||||
FIST_CHEST_DIST = 0.25 # 举拳：拳与肩水平距离 < 25% 画面宽（贴近躯干）
|
||||
KP_CONF_MIN = 0.3 # 关键点置信度过滤
|
||||
|
||||
_face_model = None
|
||||
_pose_model = None
|
||||
|
||||
|
||||
def get_face_model():
|
||||
global _face_model
|
||||
if _face_model is None:
|
||||
from ultralytics import YOLO
|
||||
log.info("vision: 加载 YOLO 人脸模型 %s（首次约 2-5s）", FACE_MODEL_PATH)
|
||||
_face_model = YOLO(str(FACE_MODEL_PATH), task="detect")
|
||||
log.info("vision: YOLO 人脸模型就绪")
|
||||
return _face_model
|
||||
|
||||
|
||||
def get_pose_model():
|
||||
global _pose_model
|
||||
if _pose_model is None:
|
||||
from ultralytics import YOLO
|
||||
log.info("vision: 加载 YOLO 姿态模型 %s（首次约 2-5s）", POSE_MODEL_PATH)
|
||||
_pose_model = YOLO(str(POSE_MODEL_PATH), task="pose")
|
||||
log.info("vision: YOLO 姿态模型就绪")
|
||||
return _pose_model
|
||||
|
||||
|
||||
def _parse_pose(results):
|
||||
"""解析姿态结果 → persons（17 关键点归一化 [x,y,conf]）+ raised（举手）+ fists（举拳）"""
|
||||
persons = []
|
||||
raised = []
|
||||
fists = []
|
||||
for r in results:
|
||||
if r.keypoints is None:
|
||||
continue
|
||||
kps = r.keypoints.data # [N,17,3] 归一化坐标
|
||||
for i in range(kps.shape[0]):
|
||||
kp = kps[i]
|
||||
person = [
|
||||
[round(float(kp[j][0]), 3), round(float(kp[j][1]), 3), round(float(kp[j][2]), 3)]
|
||||
for j in range(17)
|
||||
]
|
||||
persons.append(person)
|
||||
# 每只手臂：举手（腕明显高于肩）或 举拳（前臂上举、拳收胸前），两者互斥
|
||||
for side, wrist_i, elbow_i, shoulder_i in (
|
||||
("left", KP_LEFT_WRIST, KP_LEFT_ELBOW, KP_LEFT_SHOULDER),
|
||||
("right", KP_RIGHT_WRIST, KP_RIGHT_ELBOW, KP_RIGHT_SHOULDER),
|
||||
):
|
||||
w = kp[wrist_i]
|
||||
e = kp[elbow_i]
|
||||
s = kp[shoulder_i]
|
||||
if float(w[2]) < KP_CONF_MIN or float(s[2]) < KP_CONF_MIN or float(e[2]) < KP_CONF_MIN:
|
||||
continue
|
||||
wx, wy = float(w[0]), float(w[1])
|
||||
sx, sy = float(s[0]), float(s[1])
|
||||
ey = float(e[1])
|
||||
if wy < sy - RAISE_LIFT:
|
||||
raised.append({"side": side, "x": round(wx, 3), "y": round(wy, 3)})
|
||||
elif (
|
||||
wy < ey - FIST_ELBOW_LIFT # 前臂上举（腕高于肘）
|
||||
and wy >= sy - FIST_SHOULDER_GAP # 拳不高过肩太多（收在胸前）
|
||||
and abs(wx - sx) < FIST_CHEST_DIST # 拳贴近躯干中线
|
||||
):
|
||||
fists.append({"side": side, "x": round(wx, 3), "y": round(wy, 3)})
|
||||
return persons, raised, fists
|
||||
|
||||
|
||||
def predict_jpeg(jpeg_bytes: bytes):
|
||||
"""人脸检测 + 姿态估计 → {faces, boxes, pose, raised, latency_ms}"""
|
||||
t0 = time.time()
|
||||
img = Image.open(io.BytesIO(jpeg_bytes)).convert("RGB")
|
||||
|
||||
# 人脸
|
||||
face_res = get_face_model().predict(img, conf=CONF_THRESHOLD, imgsz=IMGSZ, verbose=False, device="cpu")
|
||||
boxes = []
|
||||
for r in face_res:
|
||||
if r.boxes is None:
|
||||
continue
|
||||
for b in r.boxes:
|
||||
xyxy = [round(float(v), 1) for v in b.xyxy[0].tolist()]
|
||||
conf = round(float(b.conf[0]), 3)
|
||||
boxes.append({"box": xyxy, "conf": conf})
|
||||
|
||||
# 姿态
|
||||
pose_res = get_pose_model().predict(img, conf=CONF_THRESHOLD, imgsz=IMGSZ, verbose=False, device="cpu")
|
||||
persons, raised, fists = _parse_pose(pose_res)
|
||||
|
||||
latency_ms = round((time.time() - t0) * 1000, 1)
|
||||
log.info(
|
||||
"vision: 人脸 %d 姿态 %d 举手 %d 举拳 %d · %.0fms",
|
||||
len(boxes), len(persons), len(raised), len(fists), latency_ms,
|
||||
)
|
||||
return {
|
||||
"faces": len(boxes),
|
||||
"boxes": boxes,
|
||||
"pose": persons,
|
||||
"raised": raised,
|
||||
"fists": fists,
|
||||
"latency_ms": latency_ms,
|
||||
}
|
||||
|
||||
|
||||
def predict_base64(b64: str):
|
||||
"""入口：base64 JPEG → 检测结果 dict（含 ok 标记，失败时带 error）"""
|
||||
try:
|
||||
jpeg = base64.b64decode(b64)
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.warning("vision: base64 解码失败 %s", e)
|
||||
return {"ok": False, "error": f"bad base64: {e}"}
|
||||
try:
|
||||
data = predict_jpeg(jpeg)
|
||||
data["ok"] = True
|
||||
return data
|
||||
except Exception as e: # noqa: BLE001
|
||||
log.error("vision: 推理失败 %s", e, exc_info=True)
|
||||
return {"ok": False, "error": str(e)}
|
||||
Binary file not shown.
Binary file not shown.
Reference in New Issue
Block a user