前言通过前13篇专栏连载我们已经搭建完成一套界面美观、运行稳定、多路并发、实时预警的工业AI视觉检测系统涵盖多线程稳流、多路RTSP并发、单窗口监控大屏、独立语音告警、分机位抓拍存证、断流自愈、EXE工控部署等全套核心能力。但目前项目的数据存储方式仍为纯文本日志存在明显商用短板txt日志杂乱无章无法快速检索、筛选、统计无法按机位、时间、违规类型做数据汇总不能生成标准报表项目验收、厂区台账归档不规范无历史记录回溯能力不满足工业安防溯源要求本篇作为商用项目最终闭环篇新增工业项目标配的数据能力轻量SQLite本地数据库存储、违规数据结构化录入、历史记录查询筛选、自动生成Excel报表导出。无需部署MySQL、无需服务器单机离线即可运行完美适配工控机离线部署场景补齐项目最后一块商用短板。一、技术方案选型工业最优解1.1 数据库选型SQLite摒弃笨重的MySQL、PostgreSQL选用SQLite轻量嵌入式数据库完全适配工控机项目无服务、零配置、单文件存储开箱即用完全离线运行无需联网、无需安装数据库环境支持事务写入杜绝数据丢失、重复存储兼容EXE打包工控机部署零依赖1.2 报表工具选型openpyxl专门用于读写Excel表格支持自定义表头、样式、数据填充可自动生成标准化违规统计报表适配项目验收、厂区安全台账归档需求。二、环境依赖安装pip install openpyxlSQLite为Python内置库无需额外安装直接开箱即用。三、数据库表结构设计工业标准化设计通用违规记录表适配所有机位、所有违规类型字段完整、规范便于后续查询统计id主键自增ID唯一数据标识cam_name监控机位名称warn_type违规类型安全帽/工装/裸露warn_time违规精确时间img_path抓拍图片本地路径create_date数据录入日期用于按日统计四、数据库工具类全局封装封装数据库初始化、数据插入、历史查询、数据统计核心方法线程安全写入杜绝多机位并发写入报错。import os import sqlite3 import threading from datetime import datetime # 数据库读写锁解决多线程并发写入报错 DB_LOCK threading.Lock() # 数据库文件路径 def get_db_path(): return os.path.join(ROOT_PATH, violation_data.db) # 初始化数据库数据表 def init_db(): db_path get_db_path() with DB_LOCK: conn sqlite3.connect(db_path) cur conn.cursor() # 创建违规记录表 cur.execute( CREATE TABLE IF NOT EXISTS violation_record ( id INTEGER PRIMARY KEY AUTOINCREMENT, cam_name TEXT, warn_type TEXT, warn_time TEXT, img_path TEXT, create_date TEXT ) ) conn.commit() conn.close() # 插入单条违规数据 def insert_violation_data(cam_name, warn_type, warn_time, img_path): create_date datetime.now().strftime(%Y-%m-%d) db_path get_db_path() with DB_LOCK: try: conn sqlite3.connect(db_path) cur conn.cursor() cur.execute( INSERT INTO violation_record (cam_name,warn_type,warn_time,img_path,create_date) VALUES (?,?,?,?,?), (cam_name, warn_type, warn_time, img_path, create_date) ) conn.commit() print(f 数据入库成功{cam_name} {warn_type}) except Exception as e: print(f❌ 数据入库失败{str(e)}) finally: conn.close() # 按日期查询历史记录 def query_record_by_date(date_str): db_path get_db_path() res_list [] with DB_LOCK: conn sqlite3.connect(db_path) cur conn.cursor() cur.execute(SELECT * FROM violation_record WHERE create_date?, (date_str,)) res_list cur.fetchall() conn.close() return res_list # 统计当日违规数量 def get_today_statistics(): today datetime.now().strftime(%Y-%m-%d) records query_record_by_date(today) stat_dict { 总违规数: len(records), no_hat: 0, no_suit: 0, bare_skin: 0 } for item in records: warn_type item[2] if warn_type in stat_dict: stat_dict[warn_type] 1 return stat_dict五、Excel报表导出工具封装支持按指定日期自动生成Excel报表自定义表头、规整排版自动保存至报表目录可直接用于项目归档、安全检查。from openpyxl import Workbook from openpyxl.styles import Font, Alignment # 违规类型中文映射 WARN_TYPE_MAP { no_hat: 未佩戴安全帽, no_suit: 未穿戴工作服, bare_skin: 皮肤裸露违规 } def export_excel_report(date_strNone): 导出指定日期的违规报表 :param date_str: 日期默认今日 :return: 报表保存路径 if not date_str: date_str datetime.now().strftime(%Y-%m-%d) records query_record_by_date(date_str) if not records: print(ℹ️ 当前日期无违规数据无需导出报表) return None # 创建报表保存目录 report_dir os.path.join(ROOT_PATH, report) os.makedirs(report_dir, exist_okTrue) report_path os.path.join(report_dir, f违规统计报表_{date_str}.xlsx) # 初始化工作簿 wb Workbook() ws wb.active ws.title 违规记录统计 # 设置表头 headers [序号, 监控机位, 违规类型, 违规时间, 抓拍文件路径] ws.append(headers) # 遍历写入数据 for idx, item in enumerate(records, 1): row_data [ idx, item[1], WARN_TYPE_MAP.get(item[2], item[2]), item[3], item[4] ] ws.append(row_data) # 样式美化 for col in range(1, 6): cell ws.cell(row1, columncol) cell.font Font(boldTrue) cell.alignment Alignment(horizontalcenter, verticalcenter) # 自适应列宽 ws.column_dimensions[A].width 8 ws.column_dimensions[B].width 20 ws.column_dimensions[C].width 18 ws.column_dimensions[D].width 22 ws.column_dimensions[E].width 45 wb.save(report_path) print(f✅ 报表导出成功{report_path}) return report_path六、适配原有项目数据自动入库改造修改摄像头异步抓拍方法在抓拍存图的同时自动结构化入库无需手动操作全程无感记录所有违规数据完美兼容原有多线程、多路检测架构。修改IndustrialCamera 类 async_save 方法def async_save(self, frame, warn_type): now time.time() if now - self.last_warn[warn_type] 3: return self.last_warn[warn_type] now save_dir get_cam_save_dir(self.cam_name) fname datetime.now().strftime(%H%M%S) f_{warn_type}.jpg path os.path.join(save_dir, fname) cv2.imwrite(path, frame) # 写入txt日志保留原有功能 log_path os.path.join(save_dir, log.txt) warn_time get_time_str() log f[{warn_time}] {self.cam_name} 违规{warn_type} 文件{fname}\n with open(log_path,a,encodingutf-8) as f: f.write(log) # 新增数据库自动入库 insert_violation_data( cam_nameself.cam_name, warn_typewarn_type, warn_timewarn_time, img_pathpath ) print(f✅ [{self.cam_name}] 抓拍数据入库完成 {fname})七、初始化数据库主函数新增在程序启动时初始化数据库保证首次运行自动创建数据表无需手动干预。if __name__ __main__: # 初始化数据库 init_db() print(✅ 数据库初始化完成) model YOLO(MODEL_PATH) cam_objs [] # 初始化并启动所有摄像头线程 for cfg in CAM_LIST: cam IndustrialCamera(cfg[name], cfg[rtsp]) cam_objs.append(cam) threading.Thread(targetcam.read_thread, daemonTrue).start() threading.Thread(targetcam.infer_thread, args(model,), daemonTrue).start() print(✅ 工业监控大屏系统启动成功) cv2.namedWindow(Industrial Super Monitor Screen, cv2.WINDOW_NORMAL) # 主循环渲染大屏 while not EXIT_FLAG.is_set(): big_screen concat_grid_frames(cam_objs, 1920, 1080) cv2.imshow(Industrial Super Monitor Screen, big_screen) # 快捷键按E导出今日报表 key cv2.waitKey(1) 0xFF if key ord(e): export_excel_report() if key 27: EXIT_FLAG.set() break cv2.destroyAllWindows() executor.shutdown() print(✅ 系统安全退出资源释放完成)八、新增实用功能快捷键报表导出新增全局快捷键交互运行系统时按下E键即可一键导出当日所有违规数据Excel报表实时归档、随时统计。九、完整项目目录结构商用规范project ├─ main.py # 主程序 ├─ best.onnx # 推理模型 ├─ violation_data.db # 全局违规数据库文件 ├─ warn_save # 分机位抓拍截图日志 │ └─ 日期 │ ├─ cam1_车间作业区 │ ├─ cam2_出入口 │ └─ cam3_高空机位 └─ report # 自动生成Excel报表 └─ 违规统计报表_日期.xlsx十、功能总结项目终极闭环本篇完成了工业视觉项目最核心的商用数据能力收尾至此整套系统彻底摆脱“玩具级demo”属性完全达到商用验收标准结构化数据存储SQLite数据库稳定存储所有违规记录支持永久溯源多线程安全写入加锁处理多路摄像头同时写入不报错、不丢数据一键Excel导出标准化报表自动生成适配厂区安全台账、项目验收数据统计查询支持按日期筛选、违规类型统计方便数据分析全离线运行无需服务器、无需联网工控机完美适配兼容全部旧功能保留大屏展示、多线程并发、语音告警、断流重连、抓拍日志所有能力十一、最终专栏完结预告至此OpenCV工业视觉全套实战专栏1-14篇正式完结整套教程从零起步依次完成图像预处理 → YOLO推理训练 → ROI区域检测 → RTSP视频流 → 抓拍日志 → 语音声光告警 → 多线程性能优化 → 多路并发监控 → 单窗口大屏UI → 数据库数据存储 → Excel报表导出实现算法性能界面数据部署全链路商用闭环。后续可拓展高阶功能Web后台可视化、远程设备管理、云端数据上报、用户权限登录、违规弹窗推送等需要可以随时继续更新进阶番外篇