Add local Web demo (FastAPI + OpenCV, no browser permission needed)

local_web/serve.py

@@ -0,0 +1,310 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+手势识别 · 本地 Web 版（FastAPI + OpenCV + ONNX Runtime）
+- 摄像头由后端直接打开（无需浏览器授权，零权限问题）
+- 浏览器只显示视频流和交互控制台
+- 实时手势识别 + 手势驱动页面交互（点赞/开关/相册/暂停/重置）
+
+用法：
+    pip install -r requirements.txt
+    python serve.py
+    # 浏览器打开 http://localhost:8000
+"""
+
+import argparse
+import json
+import os
+import sys
+import threading
+import time
+
+import cv2
+import numpy as np
+import onnxruntime as ort
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse, StreamingResponse
+from fastapi.staticfiles import StaticFiles
+from PIL import Image
+
+# 让 Windows 控制台能打印 emoji
+for _s in (sys.stdout, sys.stderr):
+    if hasattr(_s, "reconfigure"):
+        _s.reconfigure(encoding="utf-8", errors="replace")
+
+# ---------- 常量 ----------
+MEAN = np.array([0.5, 0.5, 0.5], dtype=np.float32)
+STD = np.array([0.5, 0.5, 0.5], dtype=np.float32)
+CONF_THRESHOLD = 0.3
+STABLE_FRAMES = 2
+
+ZH = {
+    "call": "✋ 招手 / 呼叫", "dislike": "👎 倒赞（拇指朝下）", "fist": "✊ 握拳",
+    "four": "🖖 四指张开（4）", "like": "👍 点赞（拇指朝上）", "mute": "🤫 噤声（食指贴唇）",
+    "ok": "👌 OK（拇指食指环）", "one": "☝️ 食指（1）", "palm": "🖐️ 手掌张开",
+    "peace": "✌️ 和平（V 字）", "peace_inverted": "✌️ 倒 V 和平", "rock": "🤘 摇滚（角）",
+    "stop": "🛑 停（手掌向前）", "stop_inverted": "🛑 倒停", "three": "🤟 三指（3）",
+    "three2": "🤟 三指变体（3）", "two_up": "✌️ 双指（2）", "two_up_inverted": "✌️ 倒双指（2）",
+}
+
+GALLERY_FILES = ["like.jpg", "ok.jpg", "peace.jpg", "fist.jpg", "rock.jpg", "palm.jpg"]
+GALLERY_LABELS = ["👍 点赞", "👌 OK", "✌️ 和平", "✊ 握拳", "🤘 摇滚", "🖐️ 手掌"]
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+
+
+# ---------- 应用主体 ----------
+class GestureApp:
+    def __init__(self, model_dir: str, examples_dir: str, camera_index: int, interval_ms: int):
+        # 模型
+        self.session = ort.InferenceSession(
+            os.path.join(model_dir, "model_quantized.onnx"),
+            providers=["CPUExecutionProvider"],
+        )
+        self.input_name = self.session.get_inputs()[0].name
+        cfg = json.load(open(os.path.join(model_dir, "config.json"), encoding="utf-8"))
+        self.labels = [cfg["id2label"][str(i)] for i in range(len(cfg["id2label"]))]
+        print(f"✅ 已加载模型: {model_dir}（{len(self.labels)} 类）")
+
+        # 摄像头
+        self.cap = cv2.VideoCapture(camera_index)
+        if not self.cap.isOpened():
+            print(f"⚠️  无法打开摄像头 index={camera_index}，将以无摄像头模式运行（仅显示示例与日志）")
+        else:
+            print(f"✅ 已打开摄像头 index={camera_index}（{int(self.cap.get(3))}x{int(self.cap.get(4))}）")
+
+        self.examples_dir = examples_dir
+        self.interval = interval_ms / 1000
+        self.lock = threading.Lock()
+        self.running = True
+
+        # 状态
+        self.gallery_idx = 0
+        self.like_count = 0
+        self.toggle_on = False
+        self.paused = False
+        self.stable_gesture = None
+        self.cand_gesture = None
+        self.cand_count = 0
+        self.cur_gesture = None
+        self.cur_score = 0.0
+        self.top5 = []
+        self.logs = []
+        self.latest_jpeg = None  # 镜像后的 jpeg bytes
+
+        self.thread = threading.Thread(target=self._loop, daemon=True)
+        self.thread.start()
+
+    @staticmethod
+    def softmax(x: np.ndarray) -> np.ndarray:
+        e = np.exp(x - x.max(-1, keepdims=True))
+        return e / e.sum(-1, keepdims=True)
+
+    def _preprocess(self, bgr: np.ndarray) -> np.ndarray:
+        rgb = cv2.cvtColor(bgr, cv2.COLOR_BGR2RGB)
+        img = Image.fromarray(rgb).resize((224, 224), Image.Resampling.BILINEAR)
+        a = np.asarray(img, dtype=np.float32) / 255.0
+        a = (a - MEAN) / STD
+        return a.transpose(2, 0, 1)[None]
+
+    def _log(self, msg: str):
+        t = time.strftime("%H:%M:%S")
+        with self.lock:
+            self.logs.insert(0, {"msg": msg, "t": t})
+            if len(self.logs) > 6:
+                self.logs.pop()
+        print(f"[{t}] {msg}")
+
+    def _trigger(self, g: str):
+        with self.lock:
+            if g == "like":
+                self.like_count += 1
+                self._log(f"👍 点赞 +1，当前 {self.like_count}")
+            elif g == "ok":
+                self.toggle_on = not self.toggle_on
+                self._log(f"👌 开关切换为「{'开' if self.toggle_on else '关'}」")
+            elif g == "fist":
+                self.gallery_idx = (self.gallery_idx - 1) % len(GALLERY_FILES)
+                self._log(f"✊ 相册上一张 → {GALLERY_LABELS[self.gallery_idx]}")
+            elif g == "palm":
+                self.gallery_idx = (self.gallery_idx + 1) % len(GALLERY_FILES)
+                self._log(f"🖐️ 相册下一张 → {GALLERY_LABELS[self.gallery_idx]}")
+            elif g == "peace":
+                self.paused = not self.paused
+                self._log(f"✌️ 交互已{'暂停' if self.paused else '恢复'}")
+            elif g == "rock":
+                self.like_count = 0
+                self.logs = []
+                self._log("🤘 已重置点赞与日志")
+
+    def _handle(self, label: str, score: float):
+        with self.lock:
+            self.cur_gesture, self.cur_score = label, score
+            if score < CONF_THRESHOLD:
+                self.cand_gesture = None
+                self.cand_count = 0
+                return
+            if label == self.cand_gesture:
+                self.cand_count += 1
+            else:
+                self.cand_gesture = label
+                self.cand_count = 1
+            if self.cand_count >= STABLE_FRAMES and label != self.stable_gesture:
+                self.stable_gesture = label
+                if self.paused and label != "peace":
+                    return
+                self._trigger(label)
+
+    def _loop(self):
+        while self.running:
+            frame = None
+            if self.cap.isOpened():
+                ok, frame = self.cap.read()
+                if not ok:
+                    frame = None
+            if frame is not None:
+                frame_m = cv2.flip(frame, 1)  # 镜像
+                with self.lock:
+                    paused = self.paused
+                if not paused:
+                    x = self._preprocess(frame_m)
+                    logits = self.session.run(None, {self.input_name: x})[0]
+                    probs = self.softmax(logits)[0]
+                    idx = probs.argsort()[::-1][:5]
+                    top = [(self.labels[i], float(probs[i])) for i in idx]
+                    with self.lock:
+                        self.top5 = top
+                    self._handle(top[0][0], top[0][1])
+
+                with self.lock:
+                    cg, cs = self.cur_gesture, self.cur_score
+                    paused = self.paused
+                if cg:
+                    label_text = f"{ZH.get(cg, cg)} {cs * 100:.0f}%"
+                    cv2.putText(frame_m, label_text, (16, 36),
+                                cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 255, 0), 2, cv2.LINE_AA)
+                if paused:
+                    cv2.putText(frame_m, "PAUSED", (frame_m.shape[1] - 200, 36),
+                                cv2.FONT_HERSHEY_SIMPLEX, 1.0, (0, 0, 255), 2, cv2.LINE_AA)
+
+                _, buf = cv2.imencode(".jpg", frame_m, [cv2.IMWRITE_JPEG_QUALITY, 80])
+                with self.lock:
+                    self.latest_jpeg = buf.tobytes()
+            time.sleep(self.interval)
+
+    # 公共方法
+    def get_jpeg(self):
+        with self.lock:
+            return self.latest_jpeg
+
+    def get_state(self):
+        with self.lock:
+            return {
+                "top": self.top5,
+                "curGesture": self.cur_gesture,
+                "curScore": self.cur_score,
+                "galleryIdx": self.gallery_idx,
+                "likeCount": self.like_count,
+                "toggleOn": self.toggle_on,
+                "paused": self.paused,
+                "logs": list(self.logs),
+            }
+
+    def prev_gallery(self):
+        with self.lock:
+            self.gallery_idx = (self.gallery_idx - 1) % len(GALLERY_FILES)
+            self._log(f"🖱️ 手动: 相册上一张 → {GALLERY_LABELS[self.gallery_idx]}")
+
+    def next_gallery(self):
+        with self.lock:
+            self.gallery_idx = (self.gallery_idx + 1) % len(GALLERY_FILES)
+            self._log(f"🖱️ 手动: 相册下一张 → {GALLERY_LABELS[self.gallery_idx]}")
+
+
+# ---------- FastAPI ----------
+app = FastAPI(title="本地手势识别")
+
+
+def create_app(model_dir, examples_dir, camera_index, interval_ms):
+    gesture = GestureApp(model_dir, examples_dir, camera_index, interval_ms)
+
+    # 相册图片静态服务
+    if os.path.isdir(examples_dir):
+        app.mount("/gallery", StaticFiles(directory=examples_dir), name="gallery")
+
+    @app.get("/", response_class=HTMLResponse)
+    def index():
+        with open(os.path.join(HERE, "index.html"), encoding="utf-8") as f:
+            return f.read()
+
+    @app.get("/video")
+    def video():
+        def gen():
+            boundary = b"--frame\r\n"
+            while True:
+                jpg = gesture.get_jpeg()
+                if jpg is None:
+                    time.sleep(0.1)
+                    continue
+                yield (boundary
+                       + b"Content-Type: image/jpeg\r\n"
+                       + b"Content-Length: " + str(len(jpg)).encode() + b"\r\n\r\n"
+                       + jpg + b"\r\n")
+                time.sleep(1 / 30)
+        return StreamingResponse(gen(), media_type="multipart/x-mixed-replace; boundary=frame")
+
+    @app.get("/state")
+    def state():
+        return gesture.get_state()
+
+    @app.post("/action/prev")
+    def action_prev():
+        gesture.prev_gallery()
+        return {"ok": True}
+
+    @app.post("/action/next")
+    def action_next():
+        gesture.next_gallery()
+        return {"ok": True}
+
+    return gesture
+
+
+def main():
+    ap = argparse.ArgumentParser(description="手势识别 · 本地 Web 服务")
+    ap.add_argument("--model-dir", default=os.path.dirname(os.path.abspath(__file__)),
+                    help="模型目录（含 model_quantized.onnx 与 config.json）")
+    ap.add_argument("--examples-dir", default=None, help="示例图目录（相册）")
+    ap.add_argument("--camera", type=int, default=0, help="摄像头 index")
+    ap.add_argument("--interval-ms", type=int, default=400, help="识别间隔（毫秒）")
+    ap.add_argument("--host", default="127.0.0.1", help="监听地址")
+    ap.add_argument("--port", type=int, default=8000, help="监听端口")
+    args = ap.parse_args()
+
+    # 自动寻找 examples 目录
+    if args.examples_dir is None:
+        for cand in [
+            os.path.join(args.model_dir, "examples"),
+            os.path.join(HERE, "examples"),
+            os.path.join(os.path.dirname(HERE), "static-space", "examples"),
+        ]:
+            if os.path.isdir(cand):
+                args.examples_dir = cand
+                break
+    if not args.examples_dir or not os.path.isdir(args.examples_dir):
+        print("⚠️  未找到 examples 目录，相册可能不可用")
+        args.examples_dir = HERE
+
+    print(f"📦 模型:    {args.model_dir}")
+    print(f"🖼️  示例图: {args.examples_dir}")
+    print(f"🎥 摄像头:  index={args.camera}")
+    print(f"🌐 打开浏览器访问: http://{args.host}:{args.port}")
+    print("（按 Ctrl+C 退出）\n")
+
+    create_app(args.model_dir, args.examples_dir, args.camera, args.interval_ms)
+    import uvicorn
+    uvicorn.run(app, host=args.host, port=args.port, log_level="warning")
+
+
+if __name__ == "__main__":
+    main()