【全栈AI开发1.0】基于 FastAPI + WebSocket + YOLOv8 的实时视频检测与统计系统
一、引言
在智能安防、新零售、工业质检等领域，实时目标检测与数据记录是不可或缺的基础能力。然而，传统方案往往存在开发成本高、部署复杂、数据难以可视化等问题。为此，我设计并实现了一个轻量级、开箱即用的实时视频检测系统，具备以下特点：
🎥实时检测与标注：基于 YOLOv8，对摄像头每一帧进行推理，并用彩色方框标出物体类别与置信度。
📊动态统计面板：网页右侧实时显示当前画面中各类目标的数量。
💾Excel 自动记录：每隔 1 秒将检测统计数据写入 Excel，且支持动态添加新类别列（如画面首次出现“car”，表格自动新增该列）。
🌐WebSocket 低延迟推送：前端通过 WebSocket 接收 base64 编码的视频帧，实现流畅播放。
🚀一键启动与优雅退出：运行脚本自动打开 Chrome 浏览器；关闭所有网页后，后端自动停止服务并释放资源。
🧩模块化前端：HTML、CSS、JavaScript 分离，便于维护和定制。
本文将从系统架构、关键代码实现、运行效果、部署方式等方面进行详细分享，并提供完整的项目源码。
二、技术栈
| 层级 | 技术 |
| 后端框架 | FastAPI + Uvicorn (ASGI 服务器) |
| 目标检测 | Ultralytics YOLOv8 (yolov8n.pt) |
| 图像处理 | OpenCV |
| 实时通信 | WebSocket |
| 数据存储 | OpenPyXL（Excel 写入，支持动态列扩展） |
| 前端 | HTML/CSS/JavaScript (原生) |
| 浏览器控制 | subprocess + threading (自动打开) |
三、系统架构
下图展示了整个系统的数据流与模块关系：
后端：FastAPI 提供 REST API 用于启动/停止检测，同时提供 WebSocket 端点。
前端：连接 WebSocket，接收 JSON 消息并更新
<img>标签的src属性，同时渲染统计列表。数据持久化：全局变量
latest_stats保存最新一帧的检测结果，log_stats_periodically每 1 秒将其写入 Excel，并自动处理新出现的类别
四、关键代码实现（核心片段）
4.1 摄像头采集（camera.py）
import cv2 class Camera: def __init__(self, source=0): self.source = source # 0 代表本地摄像头，也可以是 RTSP 地址 self.cap = None def open(self): self.cap = cv2.VideoCapture(self.source) return self.cap.isOpened() def read_frame(self): if self.cap and self.cap.isOpened(): ret, frame = self.cap.read() return frame if ret else None def release(self): if self.cap: self.cap.release()4.2 YOLO 检测器（detector.py）
from ultralytics import YOLO from collections import Counter class YOLODetector: def __init__(self, model_path='./model/yolov8n.pt', conf=0.5): self.model = YOLO(model_path) self.conf = conf def detect(self, frame): if frame is None: return None, {} results = self.model(frame, conf=self.conf) if results[0].boxes is not None: cls_ids = results[0].boxes.cls.cpu().numpy().astype(int) names = [results[0].names[i] for i in cls_ids] stats = dict(Counter(names)) else: stats = {} annotated_frame = results[0].plot() # 自动绘制边界框和标签 return annotated_frame, stats4.3 WebSocket 连接管理器（websocket_manager.py）
class ConnectionManager: def __init__(self): self.active_connections = set() self.on_empty = None # 回调：所有连接断开时触发 async def connect(self, websocket): await websocket.accept() self.active_connections.add(websocket) def disconnect(self, websocket): self.active_connections.discard(websocket) if not self.active_connections and self.on_empty: self.on_empty() # 触发关闭服务器 async def broadcast(self, message): for conn in self.active_connections.copy(): try: await conn.send_text(message) except: self.disconnect(conn)4.4 视频处理主循环（app.py 核心）
async def video_processing_loop(): global is_running, latest_stats while is_running: frame = camera.read_frame() if frame is None: await asyncio.sleep(0.1) continue annotated_frame, stats = detector.detect(frame) async with stats_lock: latest_stats = stats # 供定时写入任务使用 _, buffer = cv2.imencode('.jpg', annotated_frame, [cv2.IMWRITE_JPEG_QUALITY, 80]) b64_image = base64.b64encode(buffer).decode() await manager.broadcast(json.dumps({ "type": "frame", "image": b64_image, "stats": stats })) await asyncio.sleep(0.033) # 约 30 fps4.5 动态 Excel 写入（app.py）
async def log_stats_periodically(): try: while is_running: await asyncio.sleep(1) stats = latest_stats.copy() if not stats: continue wb = load_workbook(excel_file_path) ws = wb.active headers = [cell.value for cell in ws[1]] # 找出新类别 new_cats = [c for c in stats.keys() if c not in headers[1:]] for cat in new_cats: ws.cell(row=1, column=ws.max_column+1, value=cat) wb.save(excel_file_path) # 重新加载以获取最新表头 wb = load_workbook(excel_file_path) ws = wb.active headers = [cell.value for cell in ws[1]] row = [datetime.now().strftime("%Y-%m-%d %H:%M:%S")] for cat in headers[1:]: row.append(stats.get(cat, 0)) ws.append(row) wb.save(excel_file_path) except asyncio.CancelledError: print("表格写入任务已取消，不再写入新数据") raise4.6 前端模块化（HTML + CSS + JS）
index.html（仅结构，引用外部资源）：
<!DOCTYPE html> <html> <head> <link rel="stylesheet" href="/static/style.css"> <title>YOLOv8 智能检测系统</title> </head> <body> <div class="container"> ...（视频区、统计面板等） </div> <script src="/static/script.js"></script> </body> </html>script.js（WebSocket 连接与消息处理）：
ws.onmessage = function(event) { const data = JSON.parse(event.data); if (data.type === 'frame') { document.getElementById('videoFeed').src = `data:image/jpeg;base64,${data.image}`; let html = ''; for (const [obj, cnt] of Object.entries(data.stats)) { html += `<div class="stat-item"><span>${obj}</span><span>${cnt}</span></div>`; } document.getElementById('statsList').innerHTML = html; } };五、运行效果
5.1 终端启动日志
INFO: Started server process [51988] INFO: Waiting for application startup. ✓ YOLO模型加载成功: ./model/yolov8n.pt ✓ 摄像头初始化完成 ✓ YOLO模型加载完成 INFO: Application startup complete. INFO: Uvicorn running on http:myip (Press CTRL+C to quit) 🌐 使用默认浏览器打开: http:myip INFO: ('127.0.0.1', 34007) - "WebSocket /ws" [accepted] INFO: connection open ✓ Excel日志文件已创建: work\20260501_121356.xlsx ✓ 检测已启动，Excel记录已开启 INFO: 127.0.0.1:9109 - "POST /start HTTP/1.1" 200 OK5.2 浏览器界面截图
5.3 Excel 文件内容示例
六、总结
本文完整地介绍了一个基于 FastAPI、WebSocket 和 YOLOv8 的实时目标检测系统。从架构设计、代码实现到部署运行，每个环节都力求简洁明了。系统不仅具备实时检测、动态统计、Excel 自动记录等实用功能，还实现了自动打开浏览器、关闭网页自动退出等用户体验优化。所有代码均可直接运行，适合作为 AI 视觉项目的基础原型或教学案例。
如果您觉得有帮助，欢迎给个 Star ⭐！