picture 1st从 0 到 1：基于 FastAPI + Vue + YOLOv8 打造 AI 智慧教室专注度监控系统（附完整源码与避坑指南）前言：晚自习老师巡课太累？学生偷偷走神、趴桌睡觉难以察觉？在线教育缺乏课堂反馈？今天，我将带你从零构建一个实时的 AI 学生专注度监控系统。本文将毫无保留地分享完整的架构设计、核心算法、性能优化（解决视频播完框还在动的痛点），以及整整一套运行环境的避坑指南！🎯 一、项目背景与痛点作为一名开发者，同时也是关心教育的人，我发现在晚自习或线上教学中，有两个难以解决的痛点：老师精力有限：无法同时兼顾几十个学生，难以实时发现学生的专注度下降或疲劳状态。传统监控不智能：普通的摄像头只能录像，不能实时分析画面中学生的行为和专注度。于是，我决定自己动手，开发了一套基于纯离线模型的 AI 专注度监控系统。系统核心亮点：🟢红绿框实时标注：专注显示绿框，涣散（低头、趴桌、站起）显示红框。🤖5 种离线模型一键切换：YOLOv8n-Person、YOLOv5s、YOLOX-Nano、OpenCV DNN SSD、MediaPipe Face。⏱️动态专注度评分：融合“年级因素（青春期）”和“晚自习时间衰减”算法。🗣️智能语音播报：专注度低于阈值，Windows 自动语音警告。🚀免费内网穿透：集成 Cloudflare Tunnel，随时随地远程访问。📊实时速度监控面板：对比视频播放速度与模型推理 FPS，解决“视频播完框还在动”的卡顿痛点。🛠️ 二、技术架构与选型后端：FastAPI（高性能异步框架）、OpenCV DNN（模型推理）、MediaPipe、pyttsx3（语音播报）、asyncio（异步与多线程）。前端：Vue 3、Element Plus、Canvas（透明叠加层）、WebSocket（实时通信）。穿透：Cloudflare Tunnel（cloudflared.exe）。环境：Python 3.11（强烈建议，3.12+ 有坑）。架构流程：前端视频抓帧➡️Base64 编码➡️WebSocket 发送➡️后端异步线程池推理➡️返回坐标JSON➡️前端 Canvas 透明绘制（不损原画质）。💻 三、核心代码解析（精华版）1. 后端：异步推理与防积压（解决卡顿）传统写法在await中直接跑模型会阻塞 WebSocket 接收新帧。我们利用run_in_executor将推理放入线程池，并配合前端isProcessing锁，丢帧保最新。python复制下载# FastAPI WebSocket 核心片段 executor = concurrent.futures.ThreadPoolExecutor(max_workers=2) @app.websocket("/ws/attention") async def websocket_endpoint(websocket: WebSocket): # ...省略连接与状态初始化... if payload.get("type") == "frame": try: image_data = base64.b64decode(payload["image"].split(",")[1]) frame = cv2.imdecode(np.frombuffer(image_data, np.uint8), cv2.IMREAD_COLOR) # 丢入线程池异步推理，避免阻塞 loop = asyncio.get_event_loop() result = await loop.run_in_executor( executor, calculate_attention, frame, state["grade"], current_time, state["model"] ) await websocket.send_text(json.dumps(result)) except Exception: continue2. 算法：智能截断与红绿框判定YOLO 在 COCO 数据集上训练，预测的人体框往往包含桌子下面的部分。我们加入了智能截断和比例判定，让框体完美贴合上半身。python复制下载# 坐标映射与截断核心 scale_x, scale_y = w / 640.0, h / 640.0 # 严格按比例映射回原图 x1 = max(0, int(nx * scale_x)) y1 = max(0, int(ny * scale_y)) x2 = min(w, int((nx + nw) * scale_x)) y2 = min(h, int((ny + nh) * scale_y)) box_w = x2 - x1 box_h = y2 - y1 # 如果框太高，强制截断下半身，避免框到地板 if box_h box_w * 1.5: y2 = min(h, y1 + int(box_w * 1.5)) # 基于形状的行为分析 box_h_ratio = (y2 - y1) / h if box_h_ratio 0.85: # 站起/走动 face_score -= 30; face_reasons.append("站起/走动") if box_h_ratio 0.08: # 趴在桌子上 face_score -= 30; face_reasons.append("疑似趴桌") # 红绿框逻辑 box_color = COLOR_GREEN if face_score = 80 else COLOR_RED label = "专注" if face_score = 80 else "涣散"3. 前端：Canvas 透明叠加（完美画质）摒弃了后端直接返回整张图片的做法，前端通过Canvas透明覆盖在video标签上，接收坐标数据后实时绘制，原视频画质零损失，带宽消耗极小。javascript复制下载// 前端 Canvas 绘制核心 const drawOverlay = (data) = { const canvas = overlayCanvas.value; const ctx = canvas.getContext('2d'); canvas.width = 640; canvas.height = 480; ctx.clearRect(0, 0, 640, 480); // 绘制检测框 data.boxes.forEach(b = { ctx.strokeStyle = b.color; // 绿或红 ctx.lineWidth = 3; ctx.strokeRect(b.x, b.y, b.w, b.h); // 绘制标签文字（中文完美支持） ctx.font = 'bold 16px "Microsoft YaHei", sans-serif'; ctx.fillStyle = b.color; ctx.fillText(b.label, b.x + 6, b.y - 6); }); // ...绘制右上角信息面板... };4. 部署：一键内网穿透集成了 Cloudflare，无需公网 IP，自动生成https://xxx.trycloudflare.com外网地址，完美解决外网访问和 HTTPS 下调用摄像头的问题。python复制下载def start_cloudflared(): cloudflared_path = os.path.join(BASE_DIR, "cloudflared.exe") # ...启动 cloudflared 并正则提取公网地址... match = re.search(r'https://[a-zA-Z0-9-]+\.trycloudflare\.com', line) if match: public_url = match.group(0)🚧 四、踩坑与优化指南（价值百万的避坑经验）这个项目开发过程中，我踩了无数坑，这里为你总结出最核心的 5 个坑：坑 1：Python 3.14 导致pyttsx3报错无效的类字符串问题：高版本 Python 中的comtypes无法调用 SAPI5。解决：降级到 Python 3.11（64位）。并删除旧的虚拟环境venv重新创建，使用清华源重新安装依赖。坑 2：mediapipe报错module has no attribute 'solutions'问题：新版 mediapipe (1.0.1) 移除了旧的Legacy SolutionsAPI。解决：降级安装pip install mediapipe==0.10.21。坑 3：视频播放完，画面定格但检测框还在不断变动问题：前端以 1秒/帧 的速度发送，后端推理慢，导致帧积压（缓冲队列堆积）。解决：后端：使用ThreadPoolExecutor异步推理。前端：增加isProcessing锁，模型还在处理上一帧时，直接跳过当前帧发送。同时检查ws.bufferedAmount。坑 4：WebSocket 频繁报错1001或10054问题：浏览器刷新页面时，Windows 系统强行切断 TCP 连接。解决：后端：精准拦截WebSocketDisconnect和ConnectionResetError，静默处理，不刷日志。前端：加入心跳机制（ping/pong），每 20 秒保活一次，并在beforeunload时主动ws.close()。坑 5：模型下载太慢或直接 404问题：GitHub 国内访问不稳定。解决：编写download_file_with_fallback函数，内置10~30 个国内镜像源（HuggingFace 镜像、GitHub 代理、Gitee 等），并校验Content-Length，防止下载残缺文件。📊 五、运行效果与监控面板系统启动后，你只需要在浏览器输入http://127.0.0.1:8000，选择“导入视频文件”或“实时摄像头”。左侧视频区：画面中所有学生被红框或绿框精准框选。右上角面板：实时显示综合专注度、检测人数、当前模型、主要原因。右下角速度面板：实时绘制 FPS 折线图，绿色线为模型推理速度，红色虚线为视频播放速度。一旦绿线低于红线，系统会预警，提示你需要切换到YOLOX-Nano等轻量模型。🎁 六、总结与完整源码获取这个项目涵盖了计算机视觉、异步 Web 开发、前端可视化、内网穿透、系统级语音播报，无论你是做毕业设计、参加比赛，还是想给自己的辅导班/自习室部署一套，都非常有参考价值。后续优化方向：引入DeepSORT或ByteTrack进行目标追踪，进一步减少画面抖动。增加“头部姿态估计”模型，精准识别“眼神飘忽”。将历史专注度数据存入 SQLite，生成周报/月报。源码已整理完毕！如果这篇文章对你有帮助，请点赞、收藏、关注三连支持一下！你的支持是我持续输出硬核技术的最大动力！💬互动话题：你觉得 AI 监控学生专注度，是教育的进步还是隐私的越界？欢迎在评论区留下你的看法！main,py import cv2 import numpy as np import mediapipe as mp import pyttsx3 import datetime import threading import base64 import json import os import urllib.request import socket import time import subprocess import re import asyncio import concurrent.futures from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse import uvicorn socket.setdefaulttimeout(15) app = FastAPI(title="AI 智能学生专注度系统") BASE_DIR = os.path.dirname(os.path.abspath(__file__)) STATIC_DIR = os.path.join(BASE_DIR, "static") app.mount("/static", StaticFiles(directory=STATIC_DIR),) # 全局线程池，用于后端并发推理 executor = concurrent.futures.ThreadPoolExecutor(max_workers=2) # ================= 语音播报 ================= def speak_warning(message: str): def run(): try: engine = pyttsx3.init('sapi5') engine.setProperty('rate', 180) engine.setProperty('volume', 1.0) engine.say(message) engine.runAndWait() engine.stop() except Exception as e: print(f"[语音播报失败] {e}") threading.Thread(target=run, daemon=True).start() # ================= 智能下载 ================= def download_file_with_fallback(urls, path): if os.path.exists(path) and os.path.getsize(path) 1 * 1024: print(f"✅ {os.path.basename(path)} 已存在，跳过下载。") return True print(f"正在尝试下载: {os.path.basename(path)} ...") for idx, url in enumerate(urls): try: print(f" [{idx+1}/{len(urls)}] 尝试链接: {url}") req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}) with urllib.request.urlopen(req, timeout=15) as response, open(path, 'wb') as out_file: total_size = int(response.info().get('Content-Length', 0)) downloaded = 0 chunk_size = 1024 * 256 start_time = time.time() last_print_time = start_time while True: chunk = response.read(chunk_size) if not chunk: break out_file.write(chunk) downloaded += len(chunk) current_time = time.time() if current_time - last_print_time 0.5 or (total_size 0 and downloaded = total_size): speed = downloaded / (current_time - start_time) if (current_time - start_time) 0 else 0 if total_size 0: percent = min(100, int((downloaded / total_size) * 100)) print(f"\r 进度: {percent}% | {downloaded/(1024*1024):.2f}MB / {total_size/(1024*1024):.2f}MB | 速度: {speed/1024:.1f} KB/s", end='', flush=True) last_print_time = current_time print() if total_size 0 and downloaded total_size: raise Exception(f"连接提前断开") if downloaded 1024: raise Exception(f"下载内容异常，文件太小") print(f" ✅ 下载成功，大小: {downloaded//1024} KB") return True except Exception as e: print(f"\n ❌ 失败: {e}") if os.path.exists(path): try: os.remove(path) except: pass return False # ================= 模型加载 ================= YOLO_ONNX = os.path.join(BA 从 0 到 1：基于 FastAPI + Vue + YOLOv8 打造 AI 智慧教室专注度监控系统（附完整源码与避坑指南）
picture 1st从 0 到 1：基于 FastAPI + Vue + YOLOv8 打造 AI 智慧教室专注度监控系统（附完整源码与避坑指南）前言：晚自习老师巡课太累？学生偷偷走神、趴桌睡觉难以察觉？在线教育缺乏课堂反馈？今天，我将带你从零构建一个实时的 AI 学生专注度监控系统。本文将毫无保留地分享完整的架构设计、核心算法、性能优化（解决视频播完框还在动的痛点），以及整整一套运行环境的避坑指南！🎯 一、项目背景与痛点作为一名开发者，同时也是关心教育的人，我发现在晚自习或线上教学中，有两个难以解决的痛点：老师精力有限：无法同时兼顾几十个学生，难以实时发现学生的专注度下降或疲劳状态。传统监控不智能：普通的摄像头只能录像，不能实时分析画面中学生的行为和专注度。于是，我决定自己动手，开发了一套基于纯离线模型的 AI 专注度监控系统。系统核心亮点：🟢红绿框实时标注：专注显示绿框，涣散（低头、趴桌、站起）显示红框。🤖5 种离线模型一键切换：YOLOv8n-Person、YOLOv5s、YOLOX-Nano、OpenCV DNN SSD、MediaPipe Face。⏱️动态专注度评分：融合“年级因素（青春期）”和“晚自习时间衰减”算法。🗣️智能语音播报：专注度低于阈值，Windows 自动语音警告。🚀免费内网穿透：集成 Cloudflare Tunnel，随时随地远程访问。📊实时速度监控面板：对比视频播放速度与模型推理 FPS，解决“视频播完框还在动”的卡顿痛点。🛠️ 二、技术架构与选型后端：FastAPI（高性能异步框架）、OpenCV DNN（模型推理）、MediaPipe、pyttsx3（语音播报）、asyncio（异步与多线程）。前端：Vue 3、Element Plus、Canvas（透明叠加层）、WebSocket（实时通信）。穿透：Cloudflare Tunnel（cloudflared.exe）。环境：Python 3.11（强烈建议，3.12+ 有坑）。架构流程：前端视频抓帧➡️Base64 编码➡️WebSocket 发送➡️后端异步线程池推理➡️返回坐标JSON➡️前端 Canvas 透明绘制（不损原画质）。💻 三、核心代码解析（精华版）1. 后端：异步推理与防积压（解决卡顿）传统写法在await中直接跑模型会阻塞 WebSocket 接收新帧。我们利用run_in_executor将推理放入线程池，并配合前端isProcessing锁，丢帧保最新。python复制下载# FastAPI WebSocket 核心片段 executor = concurrent.futures.ThreadPoolExecutor(max_workers=2) @app.websocket("/ws/attention") async def websocket_endpoint(websocket: WebSocket): # ...省略连接与状态初始化... if payload.get("type") == "frame": try: image_data = base64.b64decode(payload["image"].split(",")[1]) frame = cv2.imdecode(np.frombuffer(image_data, np.uint8), cv2.IMREAD_COLOR) # 丢入线程池异步推理，避免阻塞 loop = asyncio.get_event_loop() result = await loop.run_in_executor( executor, calculate_attention, frame, state["grade"], current_time, state["model"] ) await websocket.send_text(json.dumps(result)) except Exception: continue2. 算法：智能截断与红绿框判定YOLO 在 COCO 数据集上训练，预测的人体框往往包含桌子下面的部分。我们加入了智能截断和比例判定，让框体完美贴合上半身。python复制下载# 坐标映射与截断核心 scale_x, scale_y = w / 640.0, h / 640.0 # 严格按比例映射回原图 x1 = max(0, int(nx * scale_x)) y1 = max(0, int(ny * scale_y)) x2 = min(w, int((nx + nw) * scale_x)) y2 = min(h, int((ny + nh) * scale_y)) box_w = x2 - x1 box_h = y2 - y1 # 如果框太高，强制截断下半身，避免框到地板 if box_h box_w * 1.5: y2 = min(h, y1 + int(box_w * 1.5)) # 基于形状的行为分析 box_h_ratio = (y2 - y1) / h if box_h_ratio 0.85: # 站起/走动 face_score -= 30; face_reasons.append("站起/走动") if box_h_ratio 0.08: # 趴在桌子上 face_score -= 30; face_reasons.append("疑似趴桌") # 红绿框逻辑 box_color = COLOR_GREEN if face_score = 80 else COLOR_RED label = "专注" if face_score = 80 else "涣散"3. 前端：Canvas 透明叠加（完美画质）摒弃了后端直接返回整张图片的做法，前端通过Canvas透明覆盖在video标签上，接收坐标数据后实时绘制，原视频画质零损失，带宽消耗极小。javascript复制下载// 前端 Canvas 绘制核心 const drawOverlay = (data) = { const canvas = overlayCanvas.value; const ctx = canvas.getContext('2d'); canvas.width = 640; canvas.height = 480; ctx.clearRect(0, 0, 640, 480); // 绘制检测框 data.boxes.forEach(b = { ctx.strokeStyle = b.color; // 绿或红 ctx.lineWidth = 3; ctx.strokeRect(b.x, b.y, b.w, b.h); // 绘制标签文字（中文完美支持） ctx.font = 'bold 16px "Microsoft YaHei", sans-serif'; ctx.fillStyle = b.color; ctx.fillText(b.label, b.x + 6, b.y - 6); }); // ...绘制右上角信息面板... };4. 部署：一键内网穿透集成了 Cloudflare，无需公网 IP，自动生成https://xxx.trycloudflare.com外网地址，完美解决外网访问和 HTTPS 下调用摄像头的问题。python复制下载def start_cloudflared(): cloudflared_path = os.path.join(BASE_DIR, "cloudflared.exe") # ...启动 cloudflared 并正则提取公网地址... match = re.search(r'https://[a-zA-Z0-9-]+\.trycloudflare\.com', line) if match: public_url = match.group(0)🚧 四、踩坑与优化指南（价值百万的避坑经验）这个项目开发过程中，我踩了无数坑，这里为你总结出最核心的 5 个坑：坑 1：Python 3.14 导致pyttsx3报错无效的类字符串问题：高版本 Python 中的comtypes无法调用 SAPI5。解决：降级到 Python 3.11（64位）。并删除旧的虚拟环境venv重新创建，使用清华源重新安装依赖。坑 2：mediapipe报错module has no attribute 'solutions'问题：新版 mediapipe (1.0.1) 移除了旧的Legacy SolutionsAPI。解决：降级安装pip install mediapipe==0.10.21。坑 3：视频播放完，画面定格但检测框还在不断变动问题：前端以 1秒/帧 的速度发送，后端推理慢，导致帧积压（缓冲队列堆积）。解决：后端：使用ThreadPoolExecutor异步推理。前端：增加isProcessing锁，模型还在处理上一帧时，直接跳过当前帧发送。同时检查ws.bufferedAmount。坑 4：WebSocket 频繁报错1001或10054问题：浏览器刷新页面时，Windows 系统强行切断 TCP 连接。解决：后端：精准拦截WebSocketDisconnect和ConnectionResetError，静默处理，不刷日志。前端：加入心跳机制（ping/pong），每 20 秒保活一次，并在beforeunload时主动ws.close()。坑 5：模型下载太慢或直接 404问题：GitHub 国内访问不稳定。解决：编写download_file_with_fallback函数，内置10~30 个国内镜像源（HuggingFace 镜像、GitHub 代理、Gitee 等），并校验Content-Length，防止下载残缺文件。📊 五、运行效果与监控面板系统启动后，你只需要在浏览器输入http://127.0.0.1:8000，选择“导入视频文件”或“实时摄像头”。左侧视频区：画面中所有学生被红框或绿框精准框选。右上角面板：实时显示综合专注度、检测人数、当前模型、主要原因。右下角速度面板：实时绘制 FPS 折线图，绿色线为模型推理速度，红色虚线为视频播放速度。一旦绿线低于红线，系统会预警，提示你需要切换到YOLOX-Nano等轻量模型。🎁 六、总结与完整源码获取这个项目涵盖了计算机视觉、异步 Web 开发、前端可视化、内网穿透、系统级语音播报，无论你是做毕业设计、参加比赛，还是想给自己的辅导班/自习室部署一套，都非常有参考价值。后续优化方向：引入DeepSORT或ByteTrack进行目标追踪，进一步减少画面抖动。增加“头部姿态估计”模型，精准识别“眼神飘忽”。将历史专注度数据存入 SQLite，生成周报/月报。源码已整理完毕！如果这篇文章对你有帮助，请点赞、收藏、关注三连支持一下！你的支持是我持续输出硬核技术的最大动力！💬互动话题：你觉得 AI 监控学生专注度，是教育的进步还是隐私的越界？欢迎在评论区留下你的看法！main,py import cv2 import numpy as np import mediapipe as mp import pyttsx3 import datetime import threading import base64 import json import os import urllib.request import socket import time import subprocess import re import asyncio import concurrent.futures from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse import uvicorn socket.setdefaulttimeout(15) app = FastAPI(title="AI 智能学生专注度系统") BASE_DIR = os.path.dirname(os.path.abspath(__file__)) STATIC_DIR = os.path.join(BASE_DIR, "static") app.mount("/static", StaticFiles(directory=STATIC_DIR),) # 全局线程池，用于后端并发推理 executor = concurrent.futures.ThreadPoolExecutor(max_workers=2) # ================= 语音播报 ================= def speak_warning(message: str): def run(): try: engine = pyttsx3.init('sapi5') engine.setProperty('rate', 180) engine.setProperty('volume', 1.0) engine.say(message) engine.runAndWait() engine.stop() except Exception as e: print(f"[语音播报失败] {e}") threading.Thread(target=run, daemon=True).start() # ================= 智能下载 ================= def download_file_with_fallback(urls, path): if os.path.exists(path) and os.path.getsize(path) 1 * 1024: print(f"✅ {os.path.basename(path)} 已存在，跳过下载。") return True print(f"正在尝试下载: {os.path.basename(path)} ...") for idx, url in enumerate(urls): try: print(f" [{idx+1}/{len(urls)}] 尝试链接: {url}") req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'}) with urllib.request.urlopen(req, timeout=15) as response, open(path, 'wb') as out_file: total_size = int(response.info().get('Content-Length', 0)) downloaded = 0 chunk_size = 1024 * 256 start_time = time.time() last_print_time = start_time while True: chunk = response.read(chunk_size) if not chunk: break out_file.write(chunk) downloaded += len(chunk) current_time = time.time() if current_time - last_print_time 0.5 or (total_size 0 and downloaded = total_size): speed = downloaded / (current_time - start_time) if (current_time - start_time) 0 else 0 if total_size 0: percent = min(100, int((downloaded / total_size) * 100)) print(f"\r 进度: {percent}% | {downloaded/(1024*1024):.2f}MB / {total_size/(1024*1024):.2f}MB | 速度: {speed/1024:.1f} KB/s", end='', flush=True) last_print_time = current_time print() if total_size 0 and downloaded total_size: raise Exception(f"连接提前断开") if downloaded 1024: raise Exception(f"下载内容异常，文件太小") print(f" ✅ 下载成功，大小: {downloaded//1024} KB") return True except Exception as e: print(f"\n ❌ 失败: {e}") if os.path.exists(path): try: os.remove(path) except: pass return False # ================= 模型加载 ================= YOLO_ONNX = os.path.join(BA