前言上一篇我们已经成功跑通RTSP实时视频流 OpenCV全套预处理 YOLO实时推理实现了厂区画面实时检测。但是能检测 ≠ 能上线。工业项目、安防项目、政企验收项目最看重的是证据留存、数据溯源、违规记录可查。只弹窗不保存、不记录、不留痕属于半成品无法交付。本节课补齐工业项目最后一块刚需功能画面自动绘制北京时间时间戳违规瞬间自动截图保存按日期自动创建文件夹归档违规类型、时间、点位日志Txt记录防重复截图防抖机制学完本篇你的系统是完整可验收、可落地、可交付的工业级安防系统。一、工业安防落地标准功能清单正规厂区AI检测系统必须具备时间水印画面实时显示标准北京时间取证有效违规截图触发违规瞬间自动保存高清帧自动分文件夹按日期归档不乱码、不堆积日志记录每条违规记录时间类型摄像头点位防抖防刷屏避免一秒保存几十张重复截图二、工具函数封装自动建文件夹 时间获取工业项目统一规范截图、日志必须按日期分层存储方便后期运维查阅。import os import time import cv2 import numpy as np from datetime import datetime from ultralytics import YOLO # 自动创建当日保存文件夹 def get_save_dir(): # 获取当前日期 today datetime.now().strftime(%Y-%m-%d) save_path os.path.join(warn_save, today) if not os.path.exists(save_path): os.makedirs(save_path) return save_path # 获取当前时间字符串画面水印日志专用 def get_time_str(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S)三、画面实时时间戳水印绘制所有监控取证画面必须带标准时间否则无效。def draw_time_text(frame): time_str get_time_str() # 在左上角绘制时间水印 cv2.putText(frame, time_str, (20, 35), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2) return frame四、违规日志写入 TXT永久溯源每条违规自动写入日志包含时间、违规类型、抓拍文件名。def write_log(warn_type, img_name): log_time get_time_str() log_dir get_save_dir() log_path os.path.join(log_dir, log.txt) log_text f[{log_time}] 违规类型{warn_type} 抓拍文件{img_name}\n with open(log_path, a, encodingutf-8) as f: f.write(log_text)五、智能截图保存 防抖机制工业核心不加防抖模型每帧检测都会截图一秒十几张磁盘瞬间爆满。工业标准防抖同一违规类型 3 秒内只保存一张。# 全局防抖记录字典 last_warn_time { no_hat: 0, no_suit: 0, bare_skin: 0 } def save_warn_image(frame, warn_type): now time.time() # 防抖3秒间隔 if now - last_warn_time[warn_type] 3: return False # 更新最后抓拍时间 last_warn_time[warn_type] now save_dir get_save_dir() file_name datetime.now().strftime(%H%M%S) f_{warn_type}.jpg save_path os.path.join(save_dir, file_name) cv2.imwrite(save_path, frame) write_log(warn_type, file_name) print(f✅ 违规抓拍成功{file_name}) return True六、复用全套OpenCV预处理流水线完全沿用前七篇工业最优预处理代码保持项目统一性。def industrial_preprocess(frame): h, w frame.shape[:2] # ROI区域锁定 mask np.zeros_like(frame) x1, y1 int(w*0.1), int(h*0.2) x2, y2 int(w*0.9), int(h*0.9) cv2.rectangle(mask, (x1, y1), (x2, y2), (255, 255, 255), -1) frame_roi cv2.bitwise_and(frame, mask) # 工业标准预处理 frame_blur cv2.GaussianBlur(frame_roi, (5, 5), 0) gray cv2.cvtColor(frame_blur, cv2.COLOR_BGR2GRAY) binary cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 15, 3) kernel np.ones((3, 3), np.uint8) binary cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel) binary cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel) result cv2.cvtColor(binary, cv2.COLOR_GRAY2BGR) return result, (x1, y1, x2, y2)七、终极完整可上线代码实时检测水印截图日志防抖下面是可以直接部署工控机、直接交付客户的完整版工程代码。import os import time import cv2 import numpy as np from datetime import datetime from ultralytics import YOLO # 加载工业模型 model YOLO(best.onnx) # 防抖全局变量 last_warn_time { no_hat: 0, no_suit: 0, bare_skin: 0 } def get_save_dir(): today datetime.now().strftime(%Y-%m-%d) save_path os.path.join(warn_save, today) if not os.path.exists(save_path): os.makedirs(save_path) return save_path def get_time_str(): return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def draw_time_text(frame): time_str get_time_str() cv2.putText(frame, time_str, (20, 35), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2) return frame def write_log(warn_type, img_name): log_time get_time_str() log_dir get_save_dir() log_path os.path.join(log_dir, log.txt) log_text f[{log_time}] 违规类型{warn_type} 抓拍文件{img_name}\n with open(log_path, a, encodingutf-8) as f: f.write(log_text) def save_warn_image(frame, warn_type): now time.time() if now - last_warn_time[warn_type] 3: return False last_warn_time[warn_type] now save_dir get_save_dir() file_name datetime.now().strftime(%H%M%S) f_{warn_type}.jpg save_path os.path.join(save_dir, file_name) cv2.imwrite(save_path, frame) write_log(warn_type, file_name) print(f✅ 抓拍保存{file_name}) return True def industrial_preprocess(frame): h, w frame.shape[:2] mask np.zeros_like(frame) x1, y1 int(w*0.1), int(h*0.2) x2, y2 int(w*0.9), int(h*0.9) cv2.rectangle(mask, (x1, y1), (x2, y2), (255, 255, 255), -1) frame_roi cv2.bitwise_and(frame, mask) frame_blur cv2.GaussianBlur(frame_roi, (5, 5), 0) gray cv2.cvtColor(frame_blur, cv2.COLOR_BGR2GRAY) binary cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 15, 3) kernel np.ones((3, 3), np.uint8) binary cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel) binary cv2.morphologyEx(binary, cv2.MORPH_CLOSE, kernel) result cv2.cvtColor(binary, cv2.COLOR_GRAY2BGR) return result, (x1, y1, x2, y2) if __name__ __main__: cap cv2.VideoCapture(rtsp://admin:123456192.168.1.64:554/stream) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) while True: ret, frame cap.read() if not ret: print(视频流断开等待重连...) continue pre_frame, roi_pos industrial_preprocess(frame) x1, y1, x2, y2 roi_pos results model.predict(pre_frame, imgsz1280, conf0.25, iou0.4, verboseFalse) warn_flag False warn_class # 遍历检测结果判断违规类型 for res in results: boxes res.boxes for box in boxes: cls_id int(box.cls[0]) cls_name model.names[cls_id] if cls_name in [no_hat, no_suit, bare_skin]: warn_flag True warn_class cls_name # 违规自动截图日志 if warn_flag: save_warn_image(draw_time_text(frame), warn_class) # 绘制画面 draw_frame res.plot() cv2.rectangle(draw_frame, (x1, y1), (x2, y2), (0, 0, 255), 2) draw_frame draw_time_text(draw_frame) cv2.imshow(Industrial Detect System, draw_frame) if cv2.waitKey(1) 0xFF 27: break cap.release() cv2.destroyAllWindows()八、自动生成的文件结构非常规范运行后自动生成目录无需手动建文件夹warn_save ├─ 2026-07-18 │ ├─ 142230_no_hat.jpg │ ├─ 142512_bare_skin.jpg │ └─ log.txt日志内容示例[2026-07-18 14:22:30] 违规类型no_hat 抓拍文件142230_no_hat.jpg九、本篇工业落地总结本节课完成了AI视觉系统从“能跑”到“能用、可验收”的质变实现画面标准时间水印取证合规违规自动抓拍、按日期归档日志永久留存支持溯源排查防抖机制杜绝刷屏截图、磁盘爆满目前整套专栏已经完成图像预处理 → ROI区域检测 → 实时视频流 → YOLO推理 → 违规取证留存完整工业闭环。十、下篇预告下一篇程序打包 EXE 工控机无环境运行 后台常驻运行真正脱离Python环境双击直接运行适配现场工控机部署是项目交付最后一步
