当前位置：首页 > news >正文

OpenCV工业视觉进阶14｜SQLite违规数据库存储+历史记录查询+Excel报表导出，完成项目商用数据闭环（验收必备）

news 2026/7/22 2:49:27

前言

通过前13篇专栏连载，我们已经搭建完成一套界面美观、运行稳定、多路并发、实时预警的工业AI视觉检测系统，涵盖多线程稳流、多路RTSP并发、单窗口监控大屏、独立语音告警、分机位抓拍存证、断流自愈、EXE工控部署等全套核心能力。

但目前项目的数据存储方式仍为纯文本日志，存在明显商用短板：

txt日志杂乱无章，无法快速检索、筛选、统计
无法按机位、时间、违规类型做数据汇总
不能生成标准报表，项目验收、厂区台账归档不规范
无历史记录回溯能力，不满足工业安防溯源要求

本篇作为商用项目最终闭环篇，新增工业项目标配的数据能力：轻量SQLite本地数据库存储、违规数据结构化录入、历史记录查询筛选、自动生成Excel报表导出。无需部署MySQL、无需服务器，单机离线即可运行，完美适配工控机离线部署场景，补齐项目最后一块商用短板。

一、技术方案选型（工业最优解）

1.1 数据库选型：SQLite

摒弃笨重的MySQL、PostgreSQL，选用SQLite轻量嵌入式数据库，完全适配工控机项目：

无服务、零配置、单文件存储，开箱即用
完全离线运行，无需联网、无需安装数据库环境
支持事务写入，杜绝数据丢失、重复存储
兼容EXE打包，工控机部署零依赖

1.2 报表工具选型：openpyxl

专门用于读写Excel表格，支持自定义表头、样式、数据填充，可自动生成标准化违规统计报表，适配项目验收、厂区安全台账归档需求。

二、环境依赖安装

pip install openpyxl

SQLite为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_str=None): """ 导出指定日期的违规报表 :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_ok=True) 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(row=1, column=col) cell.font = Font(bold=True) cell.alignment = Alignment(horizontal="center", vertical="center") # 自适应列宽 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",encoding="utf-8") as f: f.write(log) # 新增：数据库自动入库 insert_violation_data( cam_name=self.cam_name, warn_type=warn_type, warn_time=warn_time, img_path=path ) 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(target=cam.read_thread, daemon=True).start() threading.Thread(target=cam.infer_thread, args=(model,), daemon=True).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