各位做弱电、安防、运维的朋友,是不是都有过这种经历:监控大屏铺满一整面墙,几十上百个画面来回切,眼睛盯久了酸涩发胀;半夜施工现场怕进贼,小区车库怕剐蹭,仓库怕起火冒烟,值班室不敢离人,只能靠咖啡硬撑。好不容易闭眼眯一会儿,第二天回放录像一查,关键画面恰恰就在你打盹那几分钟。
传统监控系统最大的问题不是“录得不够清楚”,而是“看得不够及时”。摄像头本身没有判断能力,它把画面录下来,但有没有人闯入、有没有烟雾火焰、有没有车辆违停,全靠人眼去轮巡、去发现。可人的注意力是有限的,夜班更是生理上扛不住。
最近我在工地上反复折腾,搭了一套免费的本地AI监控值守方案,把原来“人盯屏”的模式改成了“AI盯屏”。摄像头还是原来的摄像头,录像机还是原来的录像机,只是多加了一层本地AI推理服务,就能做到24小时自动画面分析,检测到异常事件立刻截图、录段视频、推送告警到手机。关键是全程本地运行,不上传视频流,不依赖云服务,也不需要每年交订阅费。
这篇文章把整套方案从原理到落地完整梳理了一遍,包括硬件选型、软件环境、代码实现、告警推送以及排查思路。零基础的朋友可以按步骤从上往下搭,已经在做监控项目的朋友可以直接跳到你最关心的章节。
1. 本地AI监控到底在解决什么问题
1.1 传统监控的痛点
传统监控系统可以简单理解成三个环节:采集、传输、存储。摄像头采集画面,通过网线或光纤传到录像机,录像机把视频流存储到硬盘里。这个流程是“被动记录型”的,它不关心画面里发生了什么。
这就导致几个很现实的问题:
- 实时性差:画面有没有异常,只能靠人盯着屏幕。
- 夜间效率低:夜班盯屏非常消耗精力,人很容易疲劳、漏看。
- 回放成本高:出事后翻录像,几个小时甚至十几个小时的海量录像,人工查找效率极低。
- 误报漏报并存:如果靠运动检测(移动侦测)来提醒,风吹树叶、灯光变化、飞虫爬过都会触发误报,真实入侵反而被大量无效告警淹没。
移动侦测为什么误报率高?因为它只是对比“这一帧”和“上一帧”的像素差异,只要画面里任意区域发生变化,它就会报警。它不区分变化的是一个人、一辆车、一只猫,还是树叶被风吹动。换句话说,传统移动侦测根本没有“理解画面”的能力。
1.2 本地AI能做什么
本地AI方案引入了一台带GPU(或高性能CPU)的计算设备,在本地运行目标检测、行为识别、烟火识别等算法模型。它把视频流实时解码成图像帧,每一帧都交给AI模型做推理,模型会输出画面里有哪些目标、目标在什么位置、置信度是多少。
这样系统就从“会录像”进化到了“会判断”:
| 能力 | 传统方案 | 本地AI方案 |
|---|---|---|
| 画面记录 | ✅ 支持 | ✅ 支持 |
| 人形检测 | ❌ 无,靠移动侦测 | ✅ 识别“人”,而不是“画面变化” |
| 车辆识别 | ❌ 无 | ✅ 识别车辆、车牌区域 |
| 烟火检测 | ❌ 无 | ✅ 识别火焰、烟雾特征 |
| 区域入侵 | 仅全画面 | ✅ 可划定重点区域 |
| 告警推送 | 部分支持,误报率高 | ✅ 按置信度过滤,推送截图和短视频 |
| 视频数据处理 | 本地或云端 | ✅ 全程本地处理 |
举一个最典型的场景:夜间工地大门门口,一只野猫路过触发传统移动侦测,系统推送了一条告警;20分钟后有人翻越围栏,系统反而没有识别出来。换成AI方案后,野猫经过时模型识别出目标类别置信度低,不会触发告警;人翻越围栏时,模型识别出“人”这个类别,同时判断该人位于“禁区”区域内,于是立刻告警。
1.3 为什么选择本地部署而不是云服务
市面上也有不少云AI监控产品,但实际落地时会遇到几个绕不开的问题:
- 视频流上云隐私风险高:摄像头画面涉及工地、仓库、小区、园区,视频数据传到云端,一旦平台出现安全问题,画面内容就可能泄露。
- 订阅费用持续性强:很多云AI服务按路数、按天、按月收费,路数一多,每年费用不少。
- 依赖公网稳定性:厂区、工地网络波动时,云服务直接断开,本地录像正常但AI分析停摆。
- 延迟变量多:视频上传、云端推理、结果回传,链路长了,告警延迟可能从秒级变成数十秒级别。
本地AI部署则完全不同。视频流从摄像头到本地推理设备,全程走内网,毫秒级延迟,隐私数据不出硬件设备。断网了摄像头本地录像继续,AI分析继续,告警可以先存数据库,等网络恢复后再推送到手机。
1.4 这个方案适合谁用
- 弱电工程商、安防集成商,想给自己的监控项目增加智能化卖点。
- 物业、园区、仓库、工地值班人员,想减少夜间盯屏压力。
- 个人开发者、极客玩家,家里装了摄像头,想自己搞一套智能告警。
2. 整体方案与技术选型
2.1 系统架构
整套系统可以分为四个层次:
- 视频源层:现有的网络摄像头(IPC)或录像机(NVR),通过RTSP协议输出视频流。
- 接入层:本地AI推理服务器负责拉取RTSP视频流,解码成连续帧。这里可以用流媒体网关统一处理,也可以用OpenCV直接解码。
- 分析层:AI模型对每一帧做推理,检测人、车、烟火等目标,并结合预先设置的布防区域、布防时段做判断。
- 告警层:触发目标事件后,系统自动截图、录制短视频、保存事件记录,并通过Server酱、钉钉机器人、企业微信机器人等渠道推送到手机微信或App。
如果是单路摄像头的个人项目,可以简化:一台带GPU的电脑,Python脚本拉流 → 推理 → 告警,三个环节串起来就够了。
如果是多路摄像头的工程级项目,建议引入流媒体服务器做统一接入。摄像头 RTSP 流先推到本地流媒体服务,AI分析程序统一从流媒体服务取流,避免多路摄像头因码流过大多路并发导致网络拥堵。
2.2 核心组件对比
| 组件 | 推荐方案 | 说明 |
|---|---|---|
| 流媒体网关 | MediaMTX / ZLMediaKit | MediaMTX较轻量,适合个人和小型项目;ZLMediaKit功能更强,适合工程集成 |
| 视频解码 | OpenCV / FFmpeg | OpenCV适合快速实现,FFmpeg适合自定义抽帧策略 |
| AI目标检测 | YOLOv8 / YOLOv9 / RT-DETR | 社区生态好,模型文件开源,推理速度快 |
| 推理加速 | ONNX Runtime GPU版 / TensorRT / OpenVINO | 根据你的显卡类型和框架习惯选择 |
| 告警推送 | Server酱、钉钉/企业微信机器人、邮件、MQTT | 需要根据现场网络环境选 |
| 存储 | 本地磁盘保存截图和短视频 | 最小成本方案;量大可接MinIO对象存储 |
YOLO系列是当前工程里用得最广的目标检测模型。YOLOv8(以及后来的YOLOv9、YOLO11版本)训练生态完善,官方提供了预训练权重,可以直接检测人、车、猫、狗等80种常见目标。对于弱电监控场景,我们最常用的是其中的“person”(人)类别,以及“car”“truck”“bus”(车)类别。
2.3 技术选型的几个原则
结合我实际搭建过程中的经验,选型时有几个原则值得参考:
第一,能用本地现成能力就不要造轮子。目标检测模型直接下载YOLO的预训练权重即可,不用自己标注数据集训练,除非你有特殊的检测目标。
第二,推理设备选择要平衡成本和性能。只带4路以内1080P摄像头,一张GTX 1650/RTX 3060级别的显卡就可以带动;十几路以上的场景,再考虑性能更强的显卡或边缘计算盒子。
第三,告警链路要尽量简单可靠。告警推送通道优先选择微信/钉钉机器人这类接口清晰的Webhook方案,不要自己搭App,太重了。
第四,整个系统要支持降级运行。AI推断服务挂了,摄像头原本的录像功能不能受影响,系统设计时各模块要解耦。
3. 环境准备与硬件建议
3.1 硬件清单
本地AI监控方案的核心硬件是AI推理设备。不同规模的监控点数,对硬件要求差别很大。下面给出几个参考档位:
| 应用规模 | 摄像头路数 | 推荐硬件 | 备注 |
|---|---|---|---|
| 入门体验 | 1-2路 | 普通PC + GTX 1650/RTX 3050显卡 | 内存16G,Windows/Linux均可 |
| 中小场景 | 4-8路 | 二手工作站 + RTX 3060/4060显卡 | 建议用Linux系统,服务更稳定 |
| 工程级 | 8-16路 | 服务器 + RTX 4070及以上显卡 | 可加多卡,注意散热和电源功率 |
| 极低功耗 | 1-2路 | 树莓派5 / Jetson Orin Nano | 适合室外弱电机柜,功耗低 |
注意,实际选择时不能只看摄像头路数,还要看摄像头的分辨率、帧率和码流大小。4路4K摄像头和4路720P摄像头的解码压力差距很大。
3.2 软件环境
软件环境以本文示例为准,版本号不写死,因为不同时期安装得到的最新版本会有差异。你需要按实际情况调整。
推荐使用Ubuntu 22.04 LTS作为AI推理服务器的系统,原因如下:
- 生态成熟,NVIDIA驱动、CUDA、Python环境安装资料齐全。
- 没有Windows自动更新、杀毒软件等干扰因素。
- 适合7x24小时长时间运行。
Python版本建议使用3.10或3.11。推理框架使用PyTorch(GPU版),检测模型使用YOLOv8系列。
在开始安装前,先确认你的显卡驱动是否正常。Ubuntu下可以用nvidia-smi命令查看:
nvidia-smi如果输出类似下面的信息,说明驱动正常:
+-----------------------------------------------------------------------------+ | NVIDIA-SMI 535.146.02 Driver Version: 535.146.02 CUDA Version: 12.2 | |-------------------------------+----------------------+----------------------+ | GPU Name Persistence-M| Bus-Id Disp.A | Volatile Uncorr. ECC | | Fan Temp Perf Pwr:Usage/Cap| Memory-Usage | GPU-Util Compute M. | |===============================+======================+======================| | 0 NVIDIA GeForce RTX 3060 Off | 00000000:01:00.0 On | N/A | +-------------------------------+----------------------+----------------------+如果驱动没有安装,先安装NVIDIA驱动和CUDA环境,再继续后面的步骤。
3.3 软件安装步骤
创建一个工作目录,建议结构如下:
project/ ├── config.yaml # 系统配置 ├── main.py # 主程序 ├── detector.py # 目标检测封装 ├── alert.py # 告警推送模块 ├── recorder.py # 截图录像模块 ├── requirements.txt # 依赖清单 └── models/ # AI模型文件存放目录安装Python依赖。创建一个requirements.txt:
opencv-python ultralytics numpy pyyaml requests然后执行安装:
pip install -r requirements.txt如果你的电脑有NVIDIA显卡,建议先安装对应版本的CUDA和cuDNN,再安装PyTorch GPU版。PyTorch安装命令可以在官网生成,这里给一个参考示例(具体以官网为准):
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121安装完成后,快速验证GPU是否可以被PyTorch调用:
import torch print(torch.__version__) print(torch.cuda.is_available()) print(torch.cuda.get_device_name(0))如果输出True和你的显卡型号,说明GPU环境已就绪。
3.4 摄像头RTSP地址准备
在写代码之前,先确认摄像头的RTSP取流地址。不同品牌的摄像头RTSP地址格式不同,但绝大多数走标准RTSP协议。常见格式是:
rtsp://用户名:密码@IP地址:554/Streaming/Channels/101海康威视摄像头常见的取流地址格式:
rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101其中101代表主码流第一通道,102代表子码流第一通道。AI分析一般建议用子码流,分辨率稍低但帧率稳定,解码压力小;保存告警视频时再临时切换主码流,保证画面清晰。
大华摄像头常见的取流地址格式:
rtsp://admin:password@192.168.1.64:554/cam/realmonitor?channel=1&subtype=0如果不知道摄像头的RTSP地址,有两个办法:
- 查看摄像头品牌对应的SDK文档或官方说明。
- 使用VLC播放器测试地址是否可用。在VLC中打开网络流,输入RTSP地址,如果画面正常播放,说明地址有效。
建议先用VLC确认地址能通,再开始写代码。这样可以避免代码写完了、发现半天排查的是RTSP地址错误。
4. 核心原理拆解
4.1 视频流如何被“喂”给AI
摄像头输出的RTSP流是连续的H.264或H.265编码视频。AI模型不能直接处理编码后的视频流,必须先解码还原成单张图像帧,然后逐帧分析。
这个链路是:
RTSP视频流 → FFmpeg/OpenCV解码 → 图像帧 → AI模型推理 → 检测结果OpenCV的VideoCapture类可以直接拉取RTSP流,底层调用FFmpeg做解码。每次调用cap.read()会返回一帧BGR格式的图像。
这种方式的优点是实现简单,缺点是一帧一帧串行处理,如果AI推理速度跟不上视频帧率,会导致处理积压。实际工程里有两种解决思路:
- 抽帧分析:不分析每一帧,而是每隔N帧分析一次,比如每秒分析2-5帧。监控场景不需要每帧都分析,目标在画面里停留的时间通常以秒计。
- 多线程/多进程:拉流解码线程和AI推理线程分离,解码线程不断读帧放入队列,推理线程从队列取帧分析。
4.2 目标检测模型的工作方式
目标检测模型的任务是回答两个问题:画面里有什么?它们在哪个位置?
YOLO系列模型采用单阶段检测思路,训练完成后会输出每个目标的类别、置信度、边界框坐标。边界框就是目标在画面中的矩形区域,通常用四个值表示:中心点x坐标、中心点y坐标、宽度、高度。
我们使用Ultralytics官方库加载模型时,代码可以很简单:
from ultralytics import YOLO model = YOLO("yolov8n.pt") # 加载YOLOv8n模型,n代表nano,尺寸最小、速度最快 results = model(frame) # 对一帧图像做推理推理结果里,每个检测目标包含以下关键属性:
cls:类别ID。YOLOv8 COCO预训练模型里,0代表人,2代表车,7代表卡车。conf:置信度,取值范围0到1,越大代表模型越确信。xyxy或xywh:目标的边界框坐标。
如果画面里出现一个人,模型会输出类似这样的结果:
Boxes(Xyxy): [ x1, y1, x2, y2, conf, cls] [ 142.3, 89.6, 380.1, 512.0, 0.8873, 0.0]这段结果的含义是:在图像坐标(142, 89)到(380, 512)这个区域内,有一个“人”,模型置信度88.73%。
4.3 布防区域与布防时段
检测到目标还不够,我们还需要做两道过滤:
第一道:目标类别过滤。画面里可能有行人、车辆、动物、树叶。监控场景下,你关心的可能只有“人”和“车辆”,其他目标全部忽略。
代码实现上,直接检查检测结果的类别ID是否在允许列表里:
ALLOW_CLASSES = [0, 2, 7] # 人、小汽车、卡车第二道:布防区域过滤。画面不是整幅都需要监控。比如工地大门口,你只关心门口东侧这块区域,没有其他地方。这时可以在画面上人工划定一个矩形区域,只有目标中心点落在矩形区域内才触发告警。
假设摄像头是1920x1080分辨率,想监控画面中央偏左的矩形区域,可以用归一化坐标来配置:
zone: x1: 0.2 # 左上角x坐标占画面宽度的20% y1: 0.3 # 左上角y坐标占画面高度的30% x2: 0.8 # 右下角x坐标占画面宽度的80% y2: 0.9 # 右下角y坐标占画面高度的90%代码里判断目标中心点是否在区域内:
def is_in_zone(cx, cy, zone): x1 = zone["x1"] * frame_width y1 = zone["y1"] * frame_height x2 = zone["x2"] * frame_width y2 = zone["y2"] * frame_height return x1 <= cx <= x2 and y1 <= cy <= y2第三道:布防时段过滤。白天工作人员正常进出,不需要告警;夜间无人时段,才进入布防状态。布防时段可以直接在配置文件里写:
schedule: enabled: true start: "22:00" # 晚上10点开始布防 end: "06:00" # 早上6点解除布防实际操作里,很多值班场景是“全天布防”加上“白名单时间段静默”。比如白天不告警,只记录不推送;夜间告警并推送。这样既保留了完整的证据链,又不会让告警轰炸值班手机。
4.4 告警去重
如果AI每秒分析2帧,画面里有个人站了10秒,就会触发20次检测。如果每次都推送告警,手机微信会被刷屏。
解决办法是事件锁存机制:一旦检测到目标并推送告警,记录当前告警时间和目标位置;在接下来的冷却时间内(比如60秒),如果检测到的还是同一个区域的目标,则不重复推送,只更新事件状态。
这个逻辑用一段伪代码描述:
last_alert_time = {} def should_alert(camera_id, zone_id, target_type, now): key = f"{camera_id}_{zone_id}_{target_type}" if key in last_alert_time: diff = (now - last_alert_time[key]).seconds if diff < COOL_DOWN_SECONDS: return False last_alert_time[key] = now return True多个目标进入画面,会产生多个告警事件。但在值班人员手机上,同一时间段、同一区域的同类目标,合并成一条提示“检测到1个/2个人进入布防区域”会更友好。
除了时间冷却,还可以加一个区域冷却:人一直在区域内走动,虽然位置变化但始终在同一个布防区,只要没离开该区域,就不重复报警。离开区域或者目标消失后,再恢复该区域的报警能力。
4.5 告警图片和短视频怎么保存
事件触发后,自动保存两张图片和一段视频:
- 检测到目标时,保存一张原始画面截图。
- 在截图基础上绘制边界框和类别文字,保存一张标注图。
- 从触发时间点往前回溯几秒、往后持续几秒,保存一段短视频作为证据。
短视频录制有一个细节:AI检测到目标时已经是当前帧,往前回溯需要用到环形缓冲区技术。程序一直把最近10秒的视频帧缓存在内存队列里,一旦触发告警,就把队列里的历史帧取出来,加上新来的几秒帧,一起编码成MP4文件。
这样保存下来的视频能完整还原“目标出现前->目标出现->目标移动”的全过程。Python里可以用OpenCV的VideoWriter把帧序列写成本地文件。
5. 完整实战:搭建本地AI自动值守系统
下面进入完整实战部分。我们以一台Ubuntu 22.04主机、单路RTSP摄像头为例,实现以下功能:
- 拉取RTSP视频流。
- 使用YOLOv8模型检测画面中的人。
- 人出现在布防区域内,自动截图、录像。
- 触发告警后,通过Server酱推送到手机微信。
整个项目代码会分成几个文件,方便以后扩展多路摄像头。
5.1 项目配置
创建config.yaml配置文件:
camera: rtsp_url: "rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101" name: "大门入口" model: weights: "yolov8n.pt" conf_threshold: 0.5 target_classes: [0] # 0代表person zone: x1: 0.1 y1: 0.1 x2: 0.9 y2: 0.9 alert: cooldown_seconds: 30 serverchan_sendkey: "YOUR_SENDKEY" # Server酱的SendKey,在sct.ftqq.com获取 screenshot_dir: "./data/screenshots" video_dir: "./data/videos" schedule: enabled: true start: "00:00" end: "23:59"配置项说明:
conf_threshold:置信度阈值,低于这个值的目标被忽略。设为0.5时有较好的平衡,如果你发现误报比较多,可以提到0.6或0.7。target_classes:要检测的类别列表。COCO数据集中0代表人,2代表小汽车,7代表卡车。zone:布防区域,这里默认是全画面,你可以改成自己关心的局部区域。cooldown_seconds:同一目标重复告警的冷却时间。serverchan_sendkey:Server酱的SendKey。Server酱是一个微信推送服务,扫码登录后生成SendKey,通过HTTP请求就能把消息推送到微信。这个对于个人项目非常方便。
5.2 封装目标检测模块
创建detector.py文件:
# -*- coding: utf-8 -*- """ 目标检测模块:封装YOLOv8模型加载和推理 """ from ultralytics import YOLO class ObjectDetector: def __init__(self, weights_path, conf_threshold=0.5): self.model = YOLO(weights_path) self.conf_threshold = conf_threshold def detect(self, frame): """ 对输入帧做目标检测 :param frame: OpenCV读取的BGR图像 :return: 检测结果列表,每个元素为 (cls_id, conf, x1, y1, x2, y2) """ results = self.model(frame, verbose=False) detections = [] for result in results: boxes = result.boxes if boxes is None: continue for box in boxes: cls_id = int(box.cls[0]) conf = float(box.conf[0]) if conf < self.conf_threshold: continue x1, y1, x2, y2 = box.xyxy[0].tolist() detections.append((cls_id, conf, x1, y1, x2, y2)) return detections这个类只负责一件事:给一帧图像,返回检测到的所有目标。主程序不需要关心模型细节,方便以后替换成其他模型。
5.3 封装告警模块
创建alert.py文件:
# -*- coding: utf-8 -*- """ 告警模块:支持Server酱微信推送 """ import requests import time import os class AlertManager: def __init__(self, sendkey, cooldown_seconds=30): self.sendkey = sendkey self.cooldown_seconds = cooldown_seconds self.last_alert_time = {} def should_alert(self, camera_id, zone_id, target_type): """ 判断当前目标是否应该触发告警(时间冷却去重) """ key = f"{camera_id}_{zone_id}_{target_type}" now = time.time() if key in self.last_alert_time: diff = now - self.last_alert_time[key] if diff < self.cooldown_seconds: return False self.last_alert_time[key] = now return True def send_serverchan(self, title, content): """ 通过Server酱推送消息到微信 """ url = f"https://sctapi.ftqq.com/{self.sendkey}.send" data = { "title": title, "desp": content } try: resp = requests.post(url, data=data, timeout=10) result = resp.json() if result.get("code") == 0: print(f"[告警] 推送成功: {title}") else: print(f"[告警] 推送失败: {result.get('message')}") except Exception as e: print(f"[告警] 推送异常: {e}")5.4 保存图片和录像
创建recorder.py文件:
# -*- coding: utf-8 -*- """ 录像模块:保存告警截图和短视频 """ import cv2 import os import time from collections import deque class FrameBuffer: """ 环形缓冲器:保存最近N秒的视频帧,用于告警前回溯 """ def __init__(self, max_frames=100): self.buffer = deque(maxlen=max_frames) def add(self, frame): self.buffer.append(frame.copy()) def get_frames(self): return list(self.buffer) class EventRecorder: def __init__(self, screenshot_dir="./data/screenshots", video_dir="./data/videos"): self.screenshot_dir = screenshot_dir self.video_dir = video_dir os.makedirs(screenshot_dir, exist_ok=True) os.makedirs(video_dir, exist_ok=True) def save_screenshot(self, frame, event_name="alert"): """ 保存一张告警截图 """ filename = f"{event_name}_{int(time.time())}.jpg" filepath = os.path.join(self.screenshot_dir, filename) cv2.imwrite(filepath, frame) return filepath def save_video(self, frames, event_name="alert", fps=20): """ 将帧列表保存为MP4文件 """ if not frames: return None filename = f"{event_name}_{int(time.time())}.mp4" filepath = os.path.join(self.video_dir, filename) height, width = frames[0].shape[:2] writer = cv2.VideoWriter( filepath, cv2.VideoWriter_fourcc(*"mp4v"), fps, (width, height) ) for frame in frames: writer.write(frame) writer.release() return filepath5.5 主程序
创建main.py文件:
# -*- coding: utf-8 -*- """ 本地AI监控值守服务主程序 """ import cv2 import yaml import time from detector import ObjectDetector from alert import AlertManager from recorder import EventRecorder, FrameBuffer from datetime import datetime def load_config(path="config.yaml"): with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) def is_in_schedule(cfg, now=None): """ 判断当前时间是否在布防时段内 """ schedule = cfg.get("schedule", {}).get("enabled", False) if not schedule: return True start = cfg["schedule"]["start"] end = cfg["schedule"]["end"] now = now or datetime.now().strftime("%H:%M") return start <= now <= end def is_in_zone(cx, cy, frame_w, frame_h, cfg): """ 判断目标中心点是否在布防区域内 """ zone = cfg.get("zone", {}) if not zone: return True x1 = zone.get("x1", 0) * frame_w y1 = zone.get("y1", 0) * frame_h x2 = zone.get("x2", 1) * frame_w y2 = zone.get("y2", 1) * frame_h return x1 <= cx <= x2 and y1 <= cy <= y2 def draw_detections(frame, detections, class_names): """ 在画面上绘制检测框 """ for cls_id, conf, x1, y1, x2, y2 in detections: label = class_names.get(cls_id, str(cls_id)) cv2.rectangle(frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 0, 255), 2) text = f"{label} {conf:.2f}" cv2.putText(frame, text, (int(x1), int(y1) - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2) return frame def main(): cfg = load_config() # 初始化各模块 detector = ObjectDetector( weights_path=cfg["model"]["weights"], conf_threshold=cfg["model"]["conf_threshold"] ) alert = AlertManager( sendkey=cfg["alert"]["serverchan_sendkey"], cooldown_seconds=cfg["alert"]["cooldown_seconds"] ) recorder = EventRecorder( screenshot_dir=cfg["alert"]["screenshot_dir"], video_dir=cfg["alert"]["video_dir"] ) class_names = { 0: "person", 1: "bicycle", 2: "car", 3: "motorcycle", 5: "bus", 7: "truck" } target_classes = set(cfg["model"]["target_classes"]) # 打开RTSP视频流 cap = cv2.VideoCapture(cfg["camera"]["rtsp_url"]) if not cap.isOpened(): print(f"无法打开摄像头: {cfg['camera']['rtsp_url']}") return # 设置解码缓存,减少延迟 cap.set(cv2.CAP_PROP_BUFFERSIZE, 3) # 事件触发帧缓冲 frame_buffer = FrameBuffer(max_frames=50) event_frames = [] recording = False record_frames_count = 0 event_id = None print(f"开始监控: {cfg['camera']['name']}") print(f"RTSP地址: {cfg['camera']['rtsp_url']}") while True: ret, frame = cap.read() if not ret: print("读取视频帧失败,尝试重连...") cap.release() time.sleep(3) cap = cv2.VideoCapture(cfg["camera"]["rtsp_url"]) continue # 始终保存到环形缓冲区 frame_buffer.add(frame) # 判断是否在布防时段内 if not is_in_schedule(cfg): time.sleep(1) continue # 目标检测 detections = detector.detect(frame) frame_h, frame_w = frame.shape[:2] triggered_detections = [] for cls_id, conf, x1, y1, x2, y2 in detections: if cls_id not in target_classes: continue cx = (x1 + x2) / 2 cy = (y1 + y2) / 2 if not is_in_zone(cx, cy, frame_w, frame_h, cfg): continue triggered_detections.append((cls_id, conf, x1, y1, x2, y2)) if triggered_detections: camera_id = cfg["camera"]["name"] target_type = ",".join([class_names.get(c[0], str(c[0])) for c in triggered_detections]) if alert.should_alert(camera_id, "default", target_type): print(f"[事件] 检测到目标: {target_type}, 数量: {len(triggered_detections)}") # 保存截图 shot_path = recorder.save_screenshot(frame, event_name="alert") print(f"[事件] 截图已保存: {shot_path}") # 准备录像:环形缓冲区里的历史帧 + 当前帧 event_frames = frame_buffer.get_frames() # 绘制检测框并保存标注图 annotated = draw_detections(frame.copy(), triggered_detections, class_names) annot_path = recorder.save_screenshot(annotated, event_name="annotated") print(f"[事件] 标注图已保存: {annot_path}") # 推送告警 alert.send_serverchan( title=f"{cfg['camera']['name']} 检测到{target_type}", content=f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" f"检测目标: {target_type}\n" f"数量: {len(triggered_detections)}\n" f"置信度: {max([d[1] for d in triggered_detections]):.2f}\n" f"截图: 见服务器目录 {shot_path}" ) # 进入录像状态,持续采集35帧(约2秒) recording = True record_frames_count = 0 if recording: record_frames_count += 1 event_frames.append(frame) if record_frames_count >= 35: video_path = recorder.save_video(event_frames, event_name="alert") print(f"[事件] 录像已保存: {video_path}") recording = False event_frames = [] # 显示实时画面(可选,需要桌面环境) # cv2.imshow("Monitor", frame) # if cv2.waitKey(1) & 0xFF == ord("q"): # break time.sleep(0.1) cap.release() cv2.destroyAllWindows() if __name__ == "__main__": main()5.6 运行与验证
在项目目录下执行:
python main.py正常情况下会看到类似输出:
开始监控: 大门入口 RTSP地址: rtsp://admin:password@192.168.1.64:554/Streaming/Channels/101 [事件] 检测到目标: person, 数量: 1 [事件] 截图已保存: ./data/screenshots/alert_1738123456.jpg [事件] 标注图已保存: ./data/screenshots/annotated_1738123456.jpg [事件] 录像已保存: ./data/videos/alert_1738123456.mp4 [告警] 推送成功: 大门入口 检测到person验证步骤:
- 让一个人走进摄像头布防区域,观察程序是否输出事件日志。
- 检查
data/screenshots目录下是否生成了截图文件。 - 检查
data/videos目录下是否生成了MP4文件。 - 查看手机微信是否收到了Server酱推送的消息。
如果在室内没有真实摄像头,可以先用电脑摄像头或手机摄像头做测试。也可以下载一个测试视频文件,把rtsp_url替换成本地视频文件路径,OpenCV同样支持读取视频文件。这样即使没有摄像头,也能先跑通整套检测逻辑。
5.7 把服务部署成后台常驻进程
上面的代码直接用python main.py运行,会占用一个终端窗口。一旦终端关闭,程序就停了。实际使用中需要让它作为后台服务7x24小时运行。
推荐使用systemd服务来管理。创建/etc/systemd/system/ai-monitor.service文件:
[Unit] Description=Local AI Monitor Service After=network.target [Service] Type=simple WorkingDirectory=/home/your_user/project ExecStart=/usr/bin/python3 /home/your_user/project/main.py Restart=always RestartSec=5 Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target然后执行:
sudo systemctl daemon-reload sudo systemctl enable ai-monitor sudo systemctl start ai-monitor查看服务状态和日志:
sudo systemctl status ai-monitor journalctl -u ai-monitor -f用systemd管理的好处是:程序崩溃后会自动重启,开机自动启动,不用人工干预。
6. 进阶优化:多路摄像头与烟火检测
6.1 多路摄像头并发处理
前面单路摄像头的代码可以直接扩展为多路。最简单的改造方法是为每路摄像头创建一个独立线程,每个线程里跑一套检测逻辑:
import threading def monitor_camera(cfg): # 每路摄像头的检测主循环 pass threads = [] for cam_cfg in all_cameras: t = threading.Thread(target=monitor_camera, args=(cam_cfg,)) t.start() threads.append(t) for t in threads: t.join()但需要注意,多路摄像头同时推理时,GPU显存和算力会被分摊。在实际项目中,更适合的方案是使用消息队列或任务队列框架,统一管理多路视频流。
6.2 烟火识别
除了目标检测,火灾预警是监控场景的高频需求。实现烟火识别有两条路:
- 使用现成的烟火检测模型,很多开源模型仓库里都有基于YOLO训练的烟火检测权重。
- 使用传统图像处理做辅助,比如颜色分析、帧差法检测动态区域、频率分析等。
传统方法成本低但准确率有限。简单示例是用HSV颜色空间过滤火焰颜色:
import cv2 import numpy as np def detect_fire_color(frame): hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV) # 火焰颜色范围 lower = np.array([0, 100, 100]) upper = np.array([40, 255, 255]) mask = cv2.inRange(hsv, lower, upper) fire_ratio = cv2.countNonZero(mask) / (frame.shape[0] * frame.shape[1]) return fire_ratio > 0.005 # 阈值自行调整但在复杂环境下,这种方法的误报率较高。工程化方案建议直接使用训练好的烟火检测模型,把它和人员检测模型并列执行,检测到烟火时走独立的告警通道,告警级别高于普通人员入侵。
6.3 ONNX Runtime加速推理
Ultralytics库本身会自适应使用GPU,但在生产环境中,将YOLO模型导出为ONNX格式,再使用ONNX Runtime推理,往往可以获得更低的推理延迟和更可控的资源占用。
导出ONNX模型:
from ultralytics import YOLO model = YOLO("yolov8n.pt") model.export(format="onnx", dynamic=True)然后在代码里使用YOLO("yolov8n.onnx")加载。ONNX Runtime如果配置了CUDA Execution Provider,推理速度会进一步提高。
7. 常见问题与排查思路
根据实际搭建过程中的经验,整理几个高频问题:
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 程序启动后无法打开摄像头 | RTSP地址错误、密码包含特殊字符未转义、网络不通 | 先用VLC测试RTSP地址;检查摄像机IP是否可达;密码中有@、:等特殊字符时做URL编码 |
| 画面卡顿或延迟高 | 摄像头码流太大、解码性能不够、缓冲区排队 | 改用子码流;降低摄像头帧率;调大CAP_PROP_BUFFERSIZE;缩小画面尺寸 |
| GPU利用率很低,CPU占用高 | OpenCV解码占CPU、模型未使用GPU推理 | 检查PyTorch的CUDA是否可用;确认模型权重加载到了GPU上;用nvidia-smi观察进程 |
| 频繁误报 | 置信度阈值太低、检测目标类别不精确、布防区域太宽 | 提高conf_threshold;限定target_classes;缩小布防区域;开启布防时段 |
| 漏报严重 | 置信度阈值太高、摄像头角度不理想、目标太小 | 降低阈值到0.35-0.4;调整摄像头角度;增加模型输入分辨率 |
| 告警重复推送 | 未正确配置冷却时间 | 检查cooldown_seconds配置;检查should_alert逻辑里key的设计 |
| 推理速度慢 | 模型太大、显卡性能不足、输入分辨率太高 | 换yolov8n小模型;降低推理画幅;用TensorRT加速 |
| 长时间运行后程序卡死 | 内存泄漏、RTSP流断开重连异常 | 给主循环加try-except;定期重启服务;查看systemd日志 |
| 服务器重启后服务没有自动拉起 | systemd服务未enabled | 执行sudo systemctl enable ai-monitor |
7.1 RTSP断流重连的健壮性处理
RTSP流在弱网、摄像头重启、路由器抖动时可能断开。代码里的重连逻辑是基础版本,实际生产建议加上指数退避重连:
reconnect_interval = 5 while True: cap = cv2.VideoCapture(rtsp_url) if cap.isOpened(): reconnect_interval = 5 # 成功连接后重置间隔 break print(f"连接失败,{reconnect_interval}秒后重试...") time.sleep(reconnect_interval) reconnect_interval = min(reconnect_interval * 2, 60)这样可以避免在摄像头离线期间程序疯狂尝试连接,消耗CPU和带宽。
7.2 夜间低照度画面识别率低
夜间环境光不足时,摄像头画面噪点多、对比度低,YOLO模型识别率会明显下降。几个常用的改善手段:
- 确认摄像头开启了红外夜视模式,画面尽量清晰。
- 在AI推理前对图像做预处理:直方图均衡化、降噪。
- 选择带补光功能的摄像头,或者在重点区域增加照明。
- 使用专门针对低照度场景微调的检测模型。
8. 最佳实践与工程建议
8.1 安全与合规边界
本地AI监控虽然技术门槛不高,但使用时要特别注意合规问题。
- 监控区域如果是公共区域,要确保有合法的监控布设授权,并按规定做好提示标识。
- 涉及员工、访客等个人信息采集时,要遵循相关法律法规要求。
- 告警截图和视频片段属于敏感数据,本地存储要注意访问权限控制。
- 不要把包含人脸、车牌等信息的视频流直接推到公网或第三方服务,除非确认链路合规且加密。
8.2 配置管理
配置集中放在config.yaml里是工程化第一步。更进一步,可以按摄像头建独立配置文件,或者使用配置中心统一管理多台推理服务器的配置。
摄像头密码不要明文存在仓库里。建议从环境变量或密钥管理服务读取:
import os rtsp_url = os.getenv("CAMERA_RTSP_URL", "rtsp://...")这样代码入库时不会泄露密码。
8.3 告警升级机制
单层告警在白天可行,但夜间值班时可能没人盯手机。建议设计告警升级机制:
- 检测到事件后,第一优先级推送到值班人员微信。
- 2分钟内未确认(未点击“已处理”),升级推送到项目经理或保安队长。
- 重大事件(如检测到烟火),直接电话语音告警。
这个机制可以依托于企业微信机器人、钉钉机器人或第三方告警平台实现。初期可以先用最简单的方式:Server酱推送到一个群,多人同时接收。
8.4 数据存储与清理策略
截图和录像文件会持续增长,需要设计定期清理策略。简单做法是用定时任务清掉30天前的文件:
find /data/screenshots -type f -name "*.jpg" -mtime +30 -delete find /data/videos -type f -name "*.mp4" -mtime +30 -delete生产环境建议把告警截图和录像接入对象存储,并按摄像头和时间建立目录结构:
data/ ├── camera_001/ │ ├── 2025-01-15/ │ │ ├── alert_10_30_00.jpg │ │ └── alert_10_30_00.mp4 ├── camera_002/8.5 标准化的告警事件记录
除了截图和录像,建议把每次告警的元数据写入SQLite或MySQL,方便事后查询和统计。
事件表可以设计成:
CREATE TABLE alert_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, camera_id TEXT NOT NULL, event_type TEXT NOT NULL, confidence REAL, zone TEXT, screenshot_path TEXT, video_path TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );有了事件记录,后续可以按天、周、月统计数据,比如“本周一共发生了多少次人员入侵告警”“主要集中几点”等。这对优化布防策略很有价值。
8.6 定期评估识别效果
AI模型不是部署完就一劳永逸的。随着季节变化、光照变化、摄像头角度微调,识别效果会漂移。建议:
- 每月抽看一批告警截图,统计误报率。
- 定期统计漏报事件,分析原因。
- 如果需要更高准确率,收集现场数据做模型微调训练。
9. 总结与下一步学习方向
整套方案从硬件选型、环境搭建到代码实现,核心思路是:摄像头负责采集,本地AI负责理解画面,告警服务负责通知人。摄像头、录像机不用换,只是在现有监控系统旁边加了一台运行推理服务的计算设备,就能把“被动录像”升级成“主动值守”。
目前这套方案已经可以实现:
- 24小时不间断画面分析;
- 人形、车辆检测;
- 按布防区域、布防时段精准告警;
- 告警截图、短视频自动留存;
- 手机微信实时接收告警推送。
如果你在搭建过程中踩了坑,先检查三个最容易出问题的地方:RTSP地址是否可用、GPU环境是否配置正确、布防区域和置信度阈值是否合理。这三点解决了,整套系统就稳定了一大半。
下一步想深入的话,有几个方向可以参考:多路摄像头的智能调度与负载均衡、基于现场数据微调专属检测模型、把AI检测结果接入到现有弱电管理平台中,或者增加人脸识别、车牌识别等更细分的能力。