基于DAMOYOLO-S的零售客流量分析MySQL数据存储与可视化最近和一位开连锁便利店的朋友聊天他正为门店管理发愁。每天客流量到底有多少哪个时段最忙顾客进来后都爱在哪个货架前停留这些数据他基本靠感觉或者让店员粗略统计既不准确也费时费力。他问我现在AI这么火有没有什么办法能帮他“看清”店里的情况这让我想起了之前接触过的一个轻量级目标检测模型——DAMOYOLO-S。它速度快、精度也不错特别适合部署在算力有限的边缘设备上比如店里的普通摄像头。更重要的是检测出来的结果不能只是一堆数字得存起来、能分析、能直观地看到。这不就是典型的“数据采集-存储-可视化”链路吗所以我花时间帮他梳理并实现了一套方案用DAMOYOLO-S做实时客流检测把识别到的客流数据人数、位置、时间结构化后存入MySQL数据库最后通过一个简单的Web页面把数据图表展示出来。今天我就把这个从摄像头到数据大屏的完整落地过程分享给大家如果你也有类似的线下场景数据分析需求这套方案或许能给你一些启发。1. 方案全景从看见到看懂在深入技术细节之前我们先看看这套方案到底能干什么以及它是如何运作的。简单来说它的核心目标是将物理空间里的顾客行为转化为可量化、可分析的数字洞察。整个过程可以拆解为三个核心环节感知层摄像头DAMOYOLO-S这是系统的“眼睛”。部署在门店入口、主要通道和重点货架区域的普通网络摄像头持续拍摄视频流。DAMOYOLO-S模型就像是一个不知疲倦的“观察员”实时分析每一帧画面精准地识别出画面中的每一个人并标记出他们的位置。数据层MySQL数据库这是系统的“记忆中枢”。感知层识别出的原始数据比如“第5秒画面中有3个人”是杂乱且瞬时的。我们需要把它们清洗、整理变成有结构、带时间戳、可关联的记录然后存入MySQL。例如一条记录可能是“2023-10-27 14:05:03入口区域进入1人当前区域总人数5人”。这样数据就从“看见”变成了“记住”。应用层Web可视化这是系统的“决策仪表盘”。存储在MySQL里的数据是冰冷的数字。通过一个Web页面我们可以用折线图展示全天的客流波动用热力图呈现店内的人流密度分布用柱状图对比不同货架前的停留人数。管理者一眼就能看懂店铺运营状况从而做出更科学的决策比如优化排班、调整商品陈列。整个方案的流程如下图所示你可以清晰地看到数据是如何流动的[摄像头视频流] - [DAMOYOLO-S实时检测] - [数据清洗与结构化] - [写入MySQL] - [Web后端API查询] - [前端图表渲染] - [运营人员查看]它的价值在于将前沿的轻量级AI模型与成熟稳定的数据库、Web技术相结合用较低的成本和复杂度解决了一个非常实际的商业问题如何客观、持续、自动化地了解你的顾客。2. 核心引擎DAMOYOLO-S客流检测实战要让系统“看得见”选择一款合适的检测模型是关键。我们选择了DAMOYOLO-S主要是看中它在精度和速度之间的优秀平衡非常适合在门店的普通工控机或边缘计算盒子上运行。2.1 为什么是DAMOYOLO-S你可能听说过YOLO系列模型它们以快著称。DAMOYOLO可以看作是YOLO的一个优化变种而DAMOYOLO-S是其更小巧的版本。对于客流统计这个场景它有几个突出的优点轻量高效模型参数较少计算量小在Intel NUC或Jetson Nano这类设备上也能达到实时检测比如每秒20帧以上保证流畅分析不卡顿。精度足够对于店内环境下站立、行走的顾客其检测精度mAP足以满足商业统计需求误检和漏检率在可接受范围内。便于部署社区支持较好容易转换为ONNX或TensorRT等格式方便在不同硬件平台上部署。当然如果门店环境特别复杂比如光线剧烈变化、严重遮挡你可能需要收集一些店内数据对模型进行微调效果会更好。不过对于大多数标准场景直接用开源的预训练模型已经能跑出不错的效果。2.2 从视频流到结构化数据模型检测只是第一步。摄像头拍到的每一帧经过模型处理后会输出一个包含多个检测框的列表。每个框代表一个识别到的人并带有坐标和置信度。我们的任务是把这些原始数据变成有业务意义的记录。假设我们关注三个区域entrance入口、checkout收银台、shelf_drinks饮品货架。我们需要写一个处理程序主要做三件事区域判断根据检测框的坐标判断这个人位于哪个预定义的区域内。行为判断通过对比连续帧中同一个人的位置变化判断他是“进入”、“离开”还是在“停留”。数据组装将时间、区域、行为类型、人数等信息打包成一个结构化的字典或对象。下面是一个简化的Python代码示例展示了这个处理过程的核心逻辑import cv2 import numpy as np from collections import defaultdict, deque import time # 假设我们已经有了一个加载好的DAMOYOLO-S模型detector # 以及定义好的区域多边形坐标area_polygons {entrance: [...], checkout: [...]} def process_frame(frame, frame_count, tracker, area_polygons): 处理单帧画面进行检测、跟踪、区域判断和数据生成。 # 1. 使用DAMOYOLO-S检测人 detections detector.detect(frame) # 返回格式: [[x1, y1, x2, y2, conf, cls], ...] persons [det for det in detections if det[5] 0] # 假设类别0是‘person’ # 2. 使用跟踪器如ByteTrack获取当前帧每个人的唯一ID和位置 tracked_objects tracker.update(persons, frame) current_people_in_area defaultdict(set) # 记录当前帧每个区域内有谁用ID表示 event_records [] # 用于存储本帧产生的事件记录 for obj in tracked_objects: obj_id, x1, y1, x2, y2 obj[:5] center_x, center_y (x1 x2) / 2, (y1 y2) / 2 # 3. 判断此人中心点落在哪个区域 current_area None for area_name, polygon in area_polygons.items(): if cv2.pointPolygonTest(np.array(polygon), (center_x, center_y), False) 0: current_area area_name current_people_in_area[area_name].add(obj_id) break # 4. 与历史轨迹对比判断行为此处简化逻辑实际需更复杂的轨迹管理 # 假设tracker.history保存了每个ID上一帧的区域 prev_area tracker.get_previous_area(obj_id) if prev_area is not None and current_area ! prev_area: # 区域发生变化生成一条“离开”和“进入”事件 if prev_area: event_records.append({ timestamp: time.time(), area: prev_area, event_type: leave, people_count: 1 }) if current_area: event_records.append({ timestamp: time.time(), area: current_area, event_type: enter, people_count: 1 }) # 更新该ID的当前区域 tracker.update_history(obj_id, current_area) # 5. 生成每个区域的当前总人数快照 for area_name, id_set in current_people_in_area.items(): event_records.append({ timestamp: time.time(), area: area_name, event_type: snapshot, people_count: len(id_set) }) return event_records # 主循环 cap cv2.VideoCapture(0) # 打开摄像头 while True: ret, frame cap.read() if not ret: break events process_frame(frame, frame_count, tracker, area_polygons) # 接下来就是将events列表中的记录存入MySQL这段代码勾勒出了核心思路检测 - 跟踪 - 区域匹配 - 行为判断 - 生成事件。其中event_records列表里就是我们想要的、结构化的客流数据。3. 数据基石MySQL表设计与高效写入检测程序产生了源源不断的数据流我们需要一个可靠的地方存放它们。MySQL作为最流行的关系型数据库之一结构清晰、查询方便非常适合这个场景。3.1 设计一张“客流事件表”我们的数据核心是“事件”。每次有人进入区域、离开区域或者我们定期记录一下某个区域有多少人都是一个事件。为此我们设计一张表CREATE TABLE customer_flow_events ( id INT AUTO_INCREMENT PRIMARY KEY, timestamp DATETIME(3) NOT NULL, -- 精确到毫秒 area VARCHAR(50) NOT NULL, -- 区域名称如 entrance, shelf_drinks event_type VARCHAR(20) NOT NULL, -- 事件类型: enter, leave, snapshot people_count INT NOT NULL, -- 该事件涉及的人数进入/离开为1快照为当前总人数 camera_id VARCHAR(50), -- 摄像头编号可选 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); -- 为常用的查询字段创建索引大幅提升查询速度 CREATE INDEX idx_timestamp ON customer_flow_events(timestamp); CREATE INDEX idx_area_event ON customer_flow_events(area, event_type);这张表非常简洁但包含了分析所需的所有维度时间、地点、事件、人数。有了它我们就能回答诸如“今天上午10点到11点入口进了多少人”、“饮品货架前同时停留人数最多是多少”这类问题。3.2 批量写入提升性能检测程序每秒可能产生多条记录。如果每条都单独执行一次INSERT语句会给数据库造成很大压力。更高效的做法是批量写入。我们可以在检测程序中先将事件记录暂存到一个列表里每隔一定时间比如1秒或当列表达到一定长度比如100条时一次性写入数据库。import pymysql from datetime import datetime import json class MySQLBatchInserter: def __init__(self, host, user, password, database, batch_size100): self.connection pymysql.connect(hosthost, useruser, passwordpassword, databasedatabase) self.cursor self.connection.cursor() self.batch_size batch_size self.event_buffer [] def add_event(self, event_dict): 将单条事件添加到缓冲区 self.event_buffer.append(event_dict) if len(self.event_buffer) self.batch_size: self.flush() def flush(self): 将缓冲区所有事件批量写入数据库 if not self.event_buffer: return sql INSERT INTO customer_flow_events (timestamp, area, event_type, people_count) VALUES (%s, %s, %s, %s) # 准备批量数据 values [ (datetime.fromtimestamp(ev[timestamp]), ev[area], ev[event_type], ev[people_count]) for ev in self.event_buffer ] try: self.cursor.executemany(sql, values) self.connection.commit() print(f已批量插入 {len(values)} 条记录) self.event_buffer.clear() except Exception as e: print(f批量插入失败: {e}) self.connection.rollback() def close(self): 关闭前确保所有数据已写入 self.flush() self.cursor.close() self.connection.close() # 在检测主循环中使用 db_inserter MySQLBatchInserter(localhost, root, password, store_analytics) while True: events process_frame(...) # 获取事件列表 for ev in events: db_inserter.add_event(ev) # ... 其他处理 # 程序结束时 db_inserter.close()使用executemany进行批量插入可以显著减少网络往返和数据库事务开销让数据存储环节变得高效而稳定。4. 洞察呈现Web数据可视化实战数据存好了最后一步就是让它们“说话”。我们搭建一个简单的Web应用用图表把数据背后的故事讲出来。这里我们用最经典的Flask做后端提供数据API用ECharts在前端绘制图表。4.1 后端API从MySQL到JSON后端需要提供几个关键的API接口按前端的需求查询和聚合数据。from flask import Flask, jsonify, request import pymysql from datetime import datetime, timedelta app Flask(__name__) def get_db_connection(): return pymysql.connect(hostlocalhost, userroot, passwordpassword, databasestore_analytics) app.route(/api/flow/hourly, methods[GET]) def get_hourly_flow(): 获取指定日期、区域的小时级进出人数统计 area request.args.get(area, entrance) query_date request.args.get(date, datetime.now().strftime(%Y-%m-%d)) conn get_db_connection() cursor conn.cursor(pymysql.cursors.DictCursor) # 查询该区域当天每小时的进入和离开人数 sql SELECT HOUR(timestamp) as hour, SUM(CASE WHEN event_type enter THEN people_count ELSE 0 END) as enter_count, SUM(CASE WHEN event_type leave THEN people_count ELSE 0 END) as leave_count FROM customer_flow_events WHERE area %s AND DATE(timestamp) %s GROUP BY HOUR(timestamp) ORDER BY hour cursor.execute(sql, (area, query_date)) results cursor.fetchall() cursor.close() conn.close() # 格式化数据补全24小时 hourly_data {i: {enter: 0, leave: 0} for i in range(24)} for row in results: hour row[hour] hourly_data[hour][enter] int(row[enter_count]) hourly_data[hour][leave] int(row[leave_count]) return jsonify({ area: area, date: query_date, data: [{hour: k, **v} for k, v in hourly_data.items()] }) app.route(/api/heatmap/data, methods[GET]) def get_heatmap_data(): 获取热力图数据过去一小时内各区域的平均人数快照 # 获取过去一小时每个区域的平均人数通过snapshot事件计算 conn get_db_connection() cursor conn.cursor(pymysql.cursors.DictCursor) sql SELECT area, AVG(people_count) as avg_count FROM customer_flow_events WHERE event_type snapshot AND timestamp %s GROUP BY area one_hour_ago datetime.now() - timedelta(hours1) cursor.execute(sql, (one_hour_ago,)) results cursor.fetchall() cursor.close() conn.close() return jsonify({data: results}) if __name__ __main__: app.run(debugTrue)4.2 前端图表让数据一目了然前端页面使用ECharts库它功能强大且易于使用。我们主要创建三种图表客流趋势折线图展示指定区域如入口全天每小时进出人数的变化一眼看出高峰时段。区域热度柱状图对比不同区域如各个货架的平均停留人数发现热门区域。实时人数仪表盘显示当前店内总人数及各区域实时人数。以下是一个简化的HTML/JS示例展示如何绘制客流趋势图!DOCTYPE html html head meta charsetutf-8 title门店客流分析看板/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script /head body div idhourlyChart stylewidth: 800px; height: 400px;/div script // 初始化图表 const chartDom document.getElementById(hourlyChart); const myChart echarts.init(chartDom); // 从后端API获取数据 fetch(/api/flow/hourly?areaentrance) .then(response response.json()) .then(data { const hours data.data.map(item item.hour :00); const enterData data.data.map(item item.enter); const leaveData data.data.map(item item.leave); const option { title: { text: 入口区域客流趋势 (${data.date}), left: center }, tooltip: { trigger: axis }, legend: { data: [进入人数, 离开人数], top: 10% }, xAxis: { type: category, data: hours, name: 时间 }, yAxis: { type: value, name: 人数 }, series: [ { name: 进入人数, type: line, data: enterData, smooth: true, lineStyle: { color: #5470c6 }, itemStyle: { color: #5470c6 } }, { name: 离开人数, type: line, data: leaveData, smooth: true, lineStyle: { color: #91cc75 }, itemStyle: { color: #91cc75 } } ] }; myChart.setOption(option); }) .catch(error console.error(获取数据失败:, error)); /script /body /html通过这样一个看板店长每天打开网页就能对门店的客流情况了如指掌再也不用凭经验猜测了。5. 总结与展望把这套方案跑通后我朋友试用了一段时间反馈很不错。他说最直观的感受是以前很多模糊的感觉现在都有了清晰的数据支撑。比如他发现下午4点到5点看似人不多但饮品货架前的停留率很高于是调整了该时段的促销策略效果立竿见影。回顾整个实现过程技术选型上我们有意避开了那些庞大复杂的系统而是用DAMOYOLO-SMySQLFlask/ECharts这样轻量、成熟的组合核心是快速验证和落地。对于大多数中小型零售门店这个方案的成本和复杂度都是可控的。当然这只是一个起点。如果数据积累多了还可以做很多有意思的扩展。比如结合天气、节假日等外部数据预测未来的客流量或者对顾客动线进行更深度的聚类分析优化店铺布局甚至可以将识别出的客流数据与交易系统打通分析“客流-转化”之间的关系。技术最终要服务于业务。这套方案的价值不在于用了多炫酷的模型而在于它实实在在地打通了从物理世界感知到数字世界分析的全链路把AI从“玩具”变成了能创造商业价值的“工具”。如果你正面临类似的线下场景数字化需求不妨从这个思路开始尝试先跑通一个最小闭环再逐步迭代完善。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。