DAMOYOLO-S与数据库联动检测结果实时入库与查询你有没有想过当AI模型在摄像头前“看到”一个人、一辆车时这些信息除了在屏幕上显示一下还能做什么如果这些“看见”的瞬间——谁、在哪儿、什么时候、有多确定——都能被自动记录下来变成一条条可以随时查找、分析的数据那会怎样这就是我们今天要聊的。很多朋友用DAMOYOLO-S做实时目标检测效果不错速度快精度也够用。但检测结果往往像流水一样看一眼就过去了很难沉淀下来。比如你想知道昨天下午3点到5点仓库入口有多少辆叉车经过或者想统计一周内某个区域出现“安全帽”的频次。如果靠人工回看录像或者盯着实时画面统计那工作量简直不敢想。其实只要把DAMOYOLO-S的检测结果和数据库“牵个线”这些问题就能轻松解决。我们可以让AI在“看见”的同时自动把关键信息——对象类别、位置坐标、置信度、时间戳——整理好存进像MySQL这样的数据库里。这样一来每一次检测都变成了一条可追溯、可查询的数据记录。再配上一个简单的网页查询界面业务人员不用懂代码也能随时按时间、按类别去检索历史记录做各种分析。下面我就带你一步步搭建这样一个完整的应用流水线。从实时检测、数据入库到网页查询咱们用最直白的方式把它跑通。1. 整个方案是干什么的简单说我们要做三件事让DAMOYOLO-S持续工作它负责分析视频流比如摄像头画面认出画面里的东西。把“认出”的结果存起来每次识别出目标我们不是只画个框就完事而是把“这是什么类别”、“在画面哪里位置”、“有多确定置信度”、“什么时候看到的时间戳”这些信息规规矩矩地存到MySQL数据库的一张表里。做个查询页面看看存了什么做一个简单的网页让用户可以通过选择时间范围、目标类别等条件从数据库里把历史检测记录查出来以表格形式展示。最终你会得到一个能7x24小时运行的“AI观察员”它不仅能实时告警还默默记下了所有“看到”的事情随时等你来“翻旧账”。这对于安防监控、生产流程统计、智慧交通流量分析等场景特别有用。2. 动手之前先把环境准备好工欲善其事必先利其器。咱们先把需要的东西装好。2.1 安装核心的“干活”工具首先你需要一个能运行Python的环境。我们通过pip安装几个关键的Python库# 安装PyTorchDAMOYOLO-S基于它。请根据你的CUDA版本去PyTorch官网选择对应命令这里以CPU版本为例 pip install torch torchvision # 安装DAMOYOLO-S。可能需要从源码或特定仓库安装这里假设可以通过pip安装其核心依赖 pip install opencv-python # 用于处理视频流 pip install Pillow # 图像处理 pip install numpy # 数值计算 # 安装数据库连接和Web框架 pip install pymysql # 连接MySQL数据库 pip install flask # 用来快速搭建我们的查询网页 pip install flask-cors # 处理网页跨域请求如果前端单独部署注意DAMOYOLO-S的具体安装方式可能因版本而异。如果上述命令无法直接安装DAMOYOLO-S你可能需要查阅其官方GitHub仓库按照说明克隆源码并安装。2.2 准备一个数据库“仓库”我们需要一个数据库来存数据。这里用MySQL因为它常见、易用。安装MySQL如果你电脑上没有可以去MySQL官网下载安装或者用Docker快速启动一个。创建数据库和表登录MySQL执行下面的SQL语句创建一个数据库和一张专门存检测结果的表。-- 创建一个数据库名字叫ai_detection_log CREATE DATABASE IF NOT EXISTS ai_detection_log; USE ai_detection_log; -- 创建一张表用来记录每一次检测到的目标 CREATE TABLE IF NOT EXISTS detection_records ( id INT AUTO_INCREMENT PRIMARY KEY, -- 每条记录的唯一ID自动增长 timestamp DATETIME NOT NULL, -- 检测到目标的时间点 object_class VARCHAR(50) NOT NULL, -- 目标的类别比如‘person’ ‘car’ confidence FLOAT NOT NULL, -- 检测的置信度0到1之间的小数 x_min INT, -- bounding box左上角x坐标 y_min INT, -- bounding box左上角y坐标 x_max INT, -- bounding box右下角x坐标 y_max INT, -- bounding box右下角y坐标 frame_width INT, -- 视频帧的宽度 frame_height INT, -- 视频帧的高度 source_info VARCHAR(255) -- 可选的来源信息如摄像头ID ); -- 为了查询更快我们在经常用来查询的字段上创建索引 CREATE INDEX idx_timestamp ON detection_records(timestamp); CREATE INDEX idx_object_class ON detection_records(object_class);这张表就像我们的“流水账本”每一行代表AI在某一帧里发现的一个目标。所有关键信息都在这儿了。3. 核心环节让检测结果自动进数据库现在我们来写Python脚本把DAMOYOLO-S和MySQL连接起来。3.1 先写好数据库的“搬运工”我们先写一个工具类专门负责和数据库打交道包括连接数据库和插入数据。# db_helper.py import pymysql from datetime import datetime class DetectionDBHelper: def __init__(self, hostlocalhost, user你的用户名, password你的密码, databaseai_detection_log): 初始化数据库连接。 记得把你的用户名和你的密码换成你自己MySQL的。 self.connection pymysql.connect( hosthost, useruser, passwordpassword, databasedatabase, charsetutf8mb4 ) print(数据库连接成功) def insert_detection_record(self, record_data): 把一条检测记录插入到数据库。 record_data 是一个字典包含表里需要的所有字段。 sql INSERT INTO detection_records (timestamp, object_class, confidence, x_min, y_min, x_max, y_max, frame_width, frame_height, source_info) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) # 确保时间戳是datetime格式 if isinstance(record_data[timestamp], datetime): ts record_data[timestamp] else: ts datetime.now() # 如果没提供就用当前时间 data_tuple ( ts, record_data[object_class], record_data[confidence], record_data[x_min], record_data[y_min], record_data[x_max], record_data[y_max], record_data.get(frame_width, 640), # 提供默认值 record_data.get(frame_height, 480), record_data.get(source_info, default_camera) ) try: with self.connection.cursor() as cursor: cursor.execute(sql, data_tuple) self.connection.commit() # 提交事务让数据真正存进去 # print(f记录插入成功: {record_data[object_class]}) return True except Exception as e: print(f插入记录时出错: {e}) self.connection.rollback() # 出错就回滚 return False def close(self): 用完记得关闭连接 self.connection.close() print(数据库连接已关闭。)这个类就是个“搬运工”你给它一条检测记录字典格式它就能帮你存到数据库的detection_records表里。3.2 主程序检测入库一条龙接下来是主脚本它要做三件事加载DAMOYOLO-S模型、读取视频流、检测目标、保存结果到数据库。# realtime_detection_to_db.py import cv2 import torch from datetime import datetime import time from db_helper import DetectionDBHelper # 假设DAMOYOLO-S的模型加载和推理函数如下 # 注意这里需要你根据DAMOYOLO-S的实际使用方式来调整 def load_damoyolo_model(model_pathdamoyolo_s.pth): 加载DAMOYOLO-S模型。你需要根据官方文档调整这部分代码。 # 示例伪代码实际请参考DAMOYOLO-S官方示例 # model torch.load(model_path) # model.eval() # return model print(f加载模型: {model_path}) # 此处返回一个假模型用于演示流程 return damoyolo_model def detect_objects(model, frame): 使用模型对一帧图像进行检测。你需要根据官方文档调整这部分代码。 # 示例伪代码实际推理代码 # with torch.no_grad(): # results model(frame) # return results # 应返回包含bbox, conf, cls的列表 # 为了演示这里返回一些模拟的检测结果 height, width frame.shape[:2] simulated_detections [ {bbox: [100, 100, 200, 200], conf: 0.95, cls: person}, {bbox: [300, 150, 400, 300], conf: 0.87, cls: car}, ] return simulated_detections def main(): # 1. 初始化数据库助手 db_helper DetectionDBHelper() # 2. 加载检测模型 model load_damoyolo_model() # 3. 打开视频源可以是摄像头0也可以是视频文件 video_source 0 # 0代表默认摄像头也可以是视频文件路径如‘test.mp4’ cap cv2.VideoCapture(video_source) if not cap.isOpened(): print(无法打开视频源) return print(开始实时检测与入库... (按 q 键退出)) try: while True: ret, frame cap.read() if not ret: break current_time datetime.now() frame_height, frame_width frame.shape[:2] # 4. 对当前帧进行目标检测 detections detect_objects(model, frame) # 5. 遍历每个检测到的目标存入数据库 for det in detections: x_min, y_min, x_max, y_max det[bbox] record { timestamp: current_time, object_class: det[cls], confidence: det[conf], x_min: int(x_min), y_min: int(y_min), x_max: int(x_max), y_max: int(y_max), frame_width: frame_width, frame_height: frame_height, source_info: fcamera_{video_source} } # 调用“搬运工”存数据 db_helper.insert_detection_record(record) # 6. 可选在画面上显示检测框方便实时观看 for det in detections: x1, y1, x2, y2 map(int, det[bbox]) label f{det[cls]} {det[conf]:.2f} cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 255, 0), 2) cv2.putText(frame, label, (x1, y1-10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0,255,0), 2) cv2.imshow(Real-time Detection, frame) # 按‘q’退出循环 if cv2.waitKey(1) 0xFF ord(q): break # 控制一下处理频率避免太快模拟实时流 time.sleep(0.03) finally: # 7. 收尾工作释放资源 cap.release() cv2.destroyAllWindows() db_helper.close() print(程序结束数据库连接已关闭。) if __name__ __main__: main()这段代码干了啥它打开摄像头每一帧都送给DAMOYOLO-S模型去检测。每检测到一个目标比如一个人就立刻把它的信息打包成一个字典然后调用我们之前写好的db_helper把这个字典“搬运”到MySQL数据库里存起来。同时为了让你能看到效果它还在视频画面上把检测框画出来。现在运行这个脚本你的数据库就会开始“涨”数据了。每条数据都是一个实实在在的检测事件。4. 做个网页看看我们存了些什么数据存进去了总不能老用SQL命令查吧我们做个简单的Web页面让不懂技术的人也能方便地查询。4.1 后端用Flask提供查询API后端很简单就是用Flask写一个接口接收前端的查询条件去数据库里找数据然后返回。# app.py (Flask后端) from flask import Flask, request, jsonify, render_template from flask_cors import CORS import pymysql from datetime import datetime, timedelta app Flask(__name__) CORS(app) # 允许跨域请求如果前端和后端在不同端口 def get_db_connection(): 创建数据库连接 return pymysql.connect( hostlocalhost, user你的用户名, password你的密码, databaseai_detection_log, charsetutf8mb4 ) app.route(/) def index(): 提供查询页面 return render_template(index.html) # 这个HTML文件我们稍后创建 app.route(/api/query, methods[GET]) def query_detections(): 查询检测记录的API接口 # 从请求参数中获取查询条件 start_time request.args.get(start_time) end_time request.args.get(end_time) object_class request.args.get(object_class) min_confidence request.args.get(min_confidence, typefloat, default0.0) limit request.args.get(limit, typeint, default100) # 默认返回100条 # 构建SQL查询语句 sql SELECT * FROM detection_records WHERE confidence %s params [min_confidence] if start_time: sql AND timestamp %s params.append(start_time) if end_time: sql AND timestamp %s params.append(end_time) if object_class and object_class ! all: sql AND object_class %s params.append(object_class) sql ORDER BY timestamp DESC LIMIT %s params.append(limit) # 执行查询 conn get_db_connection() try: with conn.cursor(pymysql.cursors.DictCursor) as cursor: # 返回字典格式 cursor.execute(sql, params) results cursor.fetchall() # 将datetime对象转换为字符串方便JSON序列化 for record in results: if isinstance(record[timestamp], datetime): record[timestamp] record[timestamp].strftime(%Y-%m-%d %H:%M:%S) return jsonify({success: True, data: results, count: len(results)}) except Exception as e: return jsonify({success: False, error: str(e)}) finally: conn.close() app.route(/api/class_list, methods[GET]) def get_class_list(): 获取数据库中所有不重复的目标类别用于前端下拉框 conn get_db_connection() try: with conn.cursor() as cursor: cursor.execute(SELECT DISTINCT object_class FROM detection_records ORDER BY object_class) classes [row[0] for row in cursor.fetchall()] return jsonify({success: True, classes: classes}) except Exception as e: return jsonify({success: False, error: str(e)}) finally: conn.close() if __name__ __main__: app.run(debugTrue, port5000) # 在5000端口启动服务这个后端提供了两个主要接口一个是/api/query用来按条件查询记录另一个是/api/class_list用来获取数据库里都有哪些物体类别方便前端做筛选。4.2 前端一个简单的查询页面前端就是一个HTML页面加上一点JavaScript用来发请求和展示数据。!-- templates/index.html -- !DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 titleDAMOYOLO-S 检测记录查询系统/title style body { font-family: sans-serif; margin: 20px; background-color: #f5f5f5; } .container { max-width: 1200px; margin: auto; background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); } h1 { color: #333; } .query-form { background: #e9ecef; padding: 15px; border-radius: 5px; margin-bottom: 20px; } .form-group { margin-bottom: 10px; } label { display: inline-block; width: 120px; } input, select, button { padding: 8px; margin: 5px; border: 1px solid #ccc; border-radius: 4px; } button { background-color: #007bff; color: white; border: none; cursor: pointer; } button:hover { background-color: #0056b3; } #results { margin-top: 20px; } table { width: 100%; border-collapse: collapse; margin-top: 10px; } th, td { border: 1px solid #ddd; padding: 8px; text-align: left; } th { background-color: #007bff; color: white; } tr:nth-child(even) { background-color: #f2f2f2; } .loading { display: none; color: #007bff; } /style /head body div classcontainer h1 DAMOYOLO-S 检测历史记录查询/h1 p在这里你可以查询AI模型识别到的所有目标记录。/p div classquery-form h3筛选条件/h3 div classform-group label forstartTime开始时间/label input typedatetime-local idstartTime label forendTime结束时间/label input typedatetime-local idendTime /div div classform-group label forobjectClass目标类别/label select idobjectClass option valueall全部类别/option !-- 类别选项将通过JS动态加载 -- /select label forminConfidence最低置信度/label input typenumber idminConfidence min0 max1 step0.1 value0.5 /div div classform-group label forlimit最大返回条数/label input typenumber idlimit min1 max1000 value100 button onclickqueryRecords()查询/button button onclickresetForm()重置/button span idloading classloading查询中.../span /div /div div idresults h3查询结果 (span idresultCount0/span 条记录)/h3 div idresultTableContainer/div /div /div script // 页面加载时先获取所有类别 window.onload function() { fetch(/api/class_list) .then(response response.json()) .then(data { if (data.success) { const select document.getElementById(objectClass); data.classes.forEach(cls { const option document.createElement(option); option.value cls; option.textContent cls; select.appendChild(option); }); } }); }; function queryRecords() { const startTime document.getElementById(startTime).value; const endTime document.getElementById(endTime).value; const objectClass document.getElementById(objectClass).value; const minConfidence document.getElementById(minConfidence).value; const limit document.getElementById(limit).value; // 构建查询URL let url /api/query?min_confidence${minConfidence}limit${limit}; if (startTime) url start_time${startTime}; if (endTime) url end_time${endTime}; if (objectClass objectClass ! all) url object_class${objectClass}; // 显示加载中 document.getElementById(loading).style.display inline; document.getElementById(resultTableContainer).innerHTML ; // 发送请求 fetch(url) .then(response response.json()) .then(data { document.getElementById(loading).style.display none; if (data.success) { displayResults(data.data); document.getElementById(resultCount).textContent data.count; } else { alert(查询失败: data.error); } }) .catch(error { document.getElementById(loading).style.display none; alert(请求出错: error); }); } function displayResults(records) { if (records.length 0) { document.getElementById(resultTableContainer).innerHTML p未找到匹配的记录。/p; return; } let tableHtml table thead tr thID/th th时间戳/th th目标类别/th th置信度/th th位置 (x1,y1,x2,y2)/th th帧分辨率/th th来源/th /tr /thead tbody; records.forEach(record { tableHtml tr td${record.id}/td td${record.timestamp}/td td${record.object_class}/td td${record.confidence.toFixed(3)}/td td(${record.x_min}, ${record.y_min}, ${record.x_max}, ${record.y_max})/td td${record.frame_width}×${record.frame_height}/td td${record.source_info || N/A}/td /tr; }); tableHtml /tbody/table; document.getElementById(resultTableContainer).innerHTML tableHtml; } function resetForm() { document.getElementById(startTime).value ; document.getElementById(endTime).value ; document.getElementById(objectClass).value all; document.getElementById(minConfidence).value 0.5; document.getElementById(limit).value 100; document.getElementById(resultTableContainer).innerHTML ; document.getElementById(resultCount).textContent 0; } /script /body /html这个页面看起来挺清爽的。上面是筛选条件你可以选时间范围、物体类别、最低置信度。点“查询”按钮下面就会以表格形式列出所有符合条件的检测记录。每条记录的时间、什么东西、在哪、置信度多少都清清楚楚。5. 把它跑起来看看效果启动数据库确保你的MySQL服务正在运行。启动检测入库程序在一个终端运行python realtime_detection_to_db.py。打开摄像头你会看到实时检测画面同时数据库开始记录。启动Web查询服务在另一个终端运行python app.py。打开网页查询打开浏览器访问http://localhost:5000。你就能看到刚才做的查询页面了。试着查一下数据。比如选择过去一小时类别选“person”点击查询。你应该能看到这段时间内所有检测到“人”的记录每条记录都带着精确的时间戳和位置信息。这意味着AI的“眼睛”看到的东西已经变成了可以随时查阅、分析的结构化数据。6. 总结走完这一套流程你会发现把AI检测结果和数据库联动起来并没有想象中那么复杂。核心就是三个环节的衔接检测、结构化、存储与查询。DAMOYOLO-S负责“看”和“认”我们的Python脚本负责把识别结果整理成字典格式然后通过pymysql这个“搬运工”存到MySQL。最后用Flask搭一个轻量级的Web桥梁让业务人员能通过网页直接访问这些数据。这套方案的价值在于它让AI的感知能力从“实时显示”延伸到了“历史追溯”和“数据分析”。你可以基于这些数据做很多事生成不同时段的客流统计报表、分析特定区域的安全事件频次、甚至训练更精准的二次分析模型。数据一旦被规整地存下来想象力空间就大了很多。当然这只是一个起点。在实际项目中你可能还需要考虑更多比如数据库的性能优化数据量大了怎么办、检测结果的去重、更复杂的聚合查询或者把数据推送到更大的数据中台。但无论如何这个“检测-入库-查询”的流水线已经为你打通了从AI感知到数据应用的关键一环。下次当你想知道“AI到底看到了什么”的时候就不用再盯着录像发愁了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。