结合 GB28181/JT1078 视频流的 YOLO 实时安防告警系统开发

前言

在之前介绍 GB28181 国标协议 和 JT1078 车载视频协议 的文章中，我们详细拆解了如何将千百路摄像头、执法仪和车载设备接入到流媒体服务器中。

那么，如何利用这些庞大的实时监控流实现智能安防（AI Surveillance）？
比如：当监控画面中出现“人员未佩戴安全帽”、“陌生人非法闯入限制区域”、“危险区域烟火”等情况时，系统能否在 1 秒内 自动抓拍并向管理后台发送警报？

在实际落地时，如果直接用简单的 cv2.VideoCapture 循环读取帧并送入 YOLO，你会发现视频画面开始严重滞后（十几秒甚至几分钟延迟），并且内存迅速飙升直至程序崩溃。这是因为 OpenCV 自带的帧缓冲区积压以及网络抖动造成的。

本文将带你设计一套**“多线程解耦 + 实时丢帧队列”**的工业级 AI 推理架构，打通从国标视频流到 YOLO 实时告警推送的完整链路。

---

一、工业级安防 AI 架构设计

在工业现场，视频流的输入和 AI 推理必须异步运行。如果推理耗时大于视频解码耗时，就会产生阻塞。

我们的系统采用以下多线程拓扑架构：

graph TD
    A[国标摄像头 GB28181/JT1078] -->|通过 ZLMediaKit 等流媒体转流| B(RTSP/RTMP/HTTP-FLV 视频流)
    B -->|Thread 1: 拉流线程| C[实时帧丢弃队列 Queue maxsize=1]
    C -->|Thread 2: 推理线程| D[YOLO 检测引擎]
    D -->|匹配到违规类别且触发冷却机制| E[自动抓拍 & 绘制警告框]
    E -->|异步线程| F[发送 HTTP Webhook / 邮件告警]
    E -->|保存| G[本地告警图片库]

为什么必须使用“丢帧队列”？

OpenCV 的 VideoCapture 默认会缓存未处理的帧。当网络延迟或 YOLO 推理稍慢时，缓冲区便会不断累积。
通过将拉流（Decoder）和推理（Inference）拆分到两个独立线程，并使用一个 maxsize=1 的队列。当推理线程忙碌时，拉流线程会将队列中积压的旧帧直接抛弃（Drop），确保推理线程每次拿到的都是当前最新的那一帧。这能将系统延迟控制在 200 毫秒以内。

---

二、开发环境准备

请在 Python 环境中安装必要的依赖库：

pip install ultralytics opencv-python requests

---

三、实时 AI 告警系统源码

新建 security_alert_system.py，代码中包含完整的双线程低延迟架构、违规告警逻辑、检测冷却机制（防止重复刷爆警报）以及异步告警发送功能：

import time
import threading
import queue
import cv2
import requests
from ultralytics import YOLO

# ==================== 1. 系统参数配置 ====================
# 支持本地视频文件、RTSP、RTMP 或 HTTP-FLV 监控流
STREAM_URL = "rtsp://admin:123456@192.168.1.64:554/h264/ch1/main/av_stream" 
# 为了方便本地测试，可以换成视频文件路径："test_monitor.mp4" 或摄像头 0

MODEL_PATH = "yolo11n.pt"  # 实际安防中建议使用专门针对安全帽、火灾训练的定制 best.pt

# 定义违规监控类别（COCO 类别中：0-person 表示人员入侵闯入）
# 如果你训练了安全帽模型，可将其设为 1-no_helmet, 2-no_vest 等
VIOLATION_CLASSES = [0] 

# 告警配置
WEBH0OK_URL = "https://your-admin-backend.com/api/v1/alerts"  # 接收告警的后台接口
ALERT_COOLDOWN_SECONDS = 10  # 相同类别的告警冷却时间，防止1秒内发送成百上千次警报

# ==================== 2. 全局状态容器 ====================
frame_queue = queue.Queue(maxsize=1)  # 严格限制长度为 1 的实时帧队列
stop_event = threading.Event()         # 线程停止控制信号
last_alert_time = {}                   # 存储每个类别的上一次告警时间 {class_id: timestamp}

# ==================== 3. 线程 1：超低延迟拉流线程 ====================
def stream_reader_thread(stream_url, q):
    """
    独立拉流线程：持续读取视频帧，利用丢弃机制确保队列中永远只有最新的一帧
    """
    cap = cv2.VideoCapture(stream_url)
    if not cap.isOpened():
        print(f"[Error] 无法连接到视频流: {stream_url}")
        stop_event.set()
        return

    print(f"[Info] 成功连接到视频流，拉流中...")
    while not stop_event.is_set():
        ret, frame = cap.read()
        if not ret:
            print("[Warning] 视频流读取中断，尝试重新连接...")
            cap.release()
            time.sleep(2)
            cap = cv2.VideoCapture(stream_url)
            continue

        # 如果队列满了，直接把旧帧丢弃，换入最新的一帧
        if q.full():
            try:
                q.get_nowait()  # 移除旧帧，不等待
            except queue.Empty:
                pass
        
        q.put(frame)

    cap.release()
    print("[Info] 拉流线程安全结束")

# ==================== 4. 异步告警发送函数 ====================
def send_alert_async(class_name, confidence, image_path):
    """
    在独立线程中异步发送告警信息和截图，防止网络请求阻塞 AI 主推理线程
    """
    def task():
        payload = {
            "event": "Safety Violation Alert",
            "class_name": class_name,
            "confidence": f"{confidence:.2f}",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        }
        
        # 准备抓拍的图片文件进行上传
        try:
            with open(image_path, 'rb') as f:
                files = {'file': (image_path, f, 'image/jpeg')}
                response = requests.post(WEBH0OK_URL, data=payload, files=files, timeout=5)
                
            if response.status_code == 200:
                print(f"[Alert Sent Success] 成功推送违规告警 {class_name} 状态码: 200")
            else:
                print(f"[Alert Sent Failed] 推送告警返回错误代码: {response.status_code}")
        except Exception as e:
            print(f"[Alert Sent Error] 网络请求失败: {e}")

    # 启动后台线程执行网络上传
    t = threading.Thread(target=task)
    t.start()

# ==================== 5. 线程 2：YOLO 推理与判定引擎 ====================
def ai_inference_thread(q, model_path):
    """
    推理线程：从队列获取最新的帧，进行 YOLO 检测，分析是否有安全违规
    """
    model = YOLO(model_path)
    print(f"[Info] YOLO 模型 {model_path} 加载成功，AI 推理就绪...")

    while not stop_event.is_set():
        try:
            # 阻塞获取帧，超时时间 1 秒
            frame = q.get(timeout=1)
        except queue.Empty:
            continue

        # 运行 YOLO 推理
        # verbose=False 关闭控制台冗余输出，device=0 可使用 GPU
        results = model(frame, verbose=False)
        annotated_frame = frame.copy()
        
        violation_detected = False
        detected_violations = []

        # 遍历检测框
        for box in results[0].boxes:
            conf = box.conf[0].item()
            cls_id = int(box.cls[0].item())
            class_name = model.names[cls_id]

            # 判定是否属于违规监控类型，并且置信度大于阈值
            if cls_id in VIOLATION_CLASSES and conf > 0.5:
                violation_detected = True
                
                # 绘制显眼的红色警报框与背景高亮
                x1, y1, x2, y2 = box.xyxy[0].tolist()
                cv2.rectangle(annotated_frame, (int(x1), int(y1)), (int(x2), int(y2)), (0, 0, 255), 3)
                
                label = f"WARNING: {class_name.upper()} {conf:.2f}"
                cv2.putText(annotated_frame, label, (int(x1), int(y1) - 10),
                            cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
                
                detected_violations.append((cls_id, class_name, conf))

        # 如果发现违规行为，且不在告警冷却时间内，则触发报警
        if violation_detected:
            current_time = time.time()
            for cls_id, class_name, conf in detected_violations:
                last_time = last_alert_time.get(cls_id, 0)
                
                # 冷却时间验证
                if current_time - last_time > ALERT_COOLDOWN_SECONDS:
                    last_alert_time[cls_id] = current_time  # 更新冷却时间戳
                    
                    # 生成本地报警截图文件名
                    img_filename = f"alerts/violation_{cls_id}_{int(current_time)}.jpg"
                    cv2.imwrite(img_filename, annotated_frame)
                    print(f"\a[⚠️ ALERT SYSTEM] 检测到违规行为: {class_name.upper()} | 截图已保存至: {img_filename}")
                    
                    # 异步推送告警
                    send_alert_async(class_name, conf, img_filename)

        # 渲染监控画面
        # 如果存在警告，在画面上方显示巨大红色警报条
        if violation_detected:
            cv2.rectangle(annotated_frame, (0, 0), (annotated_frame.shape[1], 50), (0, 0, 255), -1)
            cv2.putText(annotated_frame, "SECURITY WARNING - INTRUSION DETECTED", (30, 35),
                        cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255, 255, 255), 2)

        # 显示画面
        cv2.imshow("Industrial AI Surveillance System", annotated_frame)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            stop_event.set()
            break

    cv2.destroyAllWindows()
    print("[Info] 推理线程安全结束")

# ==================== 6. 系统入口 ====================
if __name__ == "__main__":
    import os
    # 创建本地告警文件夹
    os.makedirs("alerts", exist_ok=True)

    # 启动拉流线程
    t_reader = threading.Thread(target=stream_reader_thread, args=(STREAM_URL, frame_queue))
    t_reader.daemon = True  # 设置为守护线程，主程序退出时自动结束
    t_reader.start()

    # 启动推理线程
    ai_inference_thread(frame_queue, MODEL_PATH)
    
    # 善后处理
    stop_event.set()
    t_reader.join()
    print("[Info] 系统已安全关闭。")

---

四、如何对接国标流媒体服务器？

在真实的安防项目中，前端摄像头并不是直接向 Python 脚本提供 RTSP 流的，而是通过 GB28181 协议 注册到国标平台（如 ZLMediaKit 或 LiveGBS），车载设备则通过 JT1078 协议 接入。

以下是实现“国标摄像头接入 -> AI 分析”的真实标准步骤：

第一步：国标接入与转流

将摄像头通过国标 SIP 服务注册至流媒体服务器。以 ZLMediaKit 为例，当国标摄像头注册并上线后，流媒体服务器会自动将接收到的 RTP 视频流转换为以下格式的通用流地址：

# RTSP 播放地址
rtsp://192.168.1.100:554/live/34020000001320000001?token=xxxx
# HTTP-FLV 播放地址（高并发、适合网页）
http://192.168.1.100:80/live/34020000001320000001.flv

第二步：将转换后的流注入 AI 引擎

你只需将本篇系统中的 STREAM_URL 参数修改为上述流媒体服务器转换出来的 RTSP 或 FLV 流地址，Python 代码就能平滑无缝地接入几十台国标监控设备的画面。

---

五、工业生产级调优指南

1. 硬件加速（NVIDIA TensorRT）
   在拥有成百上千路监控的工厂区，纯 Python 推理是无法负荷的。
   请将你的训练模型通过 model.export(format="engine", half=True, device=0) 导出为 TensorRT (Float16) 格式。
   在我们的测试中，使用 TensorRT 可以将 YOLO 在 NVIDIA T4/RTX4060 上的单帧推理耗时从 15ms 压缩至 2~4ms，效率翻了数倍。

2. 动态冷却过滤与去重（Cooldown Mechanism）
   如果在画面中有人逗留了 5 分钟，系统绝不能发送数万条报警信息。
   本系统中实现的 ALERT_COOLDOWN_SECONDS 保证了同一类型警报在 10 秒内只报警一次。实际生产中，可以采用更高级的策略：记录被检测物体的 Tracking ID。只有当新 ID 越界进入时才触发一次性报警，或者结合 Redis 缓存实现全局跨设备的分布式报警去重。

3. 负样本抑制
   在户外安防中，树叶晃动、光线剧烈变化或飞鸟极易引起 YOLO 的误检（特别是火灾、安全帽检测）。
   建议在数据集中加入一定比例的“负样本”（即不包含任何目标的纯背景图）一起训练，并适当调高 conf 阈值至 0.55~0.60。

---

结语

本篇文章中，我们通过巧妙的“丢帧队列”多线程架构，打通了国标安防视频流与 YOLO 模型之间的通道，成功解决了网络传输与深度推理之间的同步阻塞痛点。

至此，我们的 AI 系统运行在高性能服务器端。但如果我们希望把目标检测应用部署到用户的手机端、平板电脑，或者纯前端网页上，不想搭建任何昂贵的后端 GPU 服务器，该怎么做？

下一篇文章我们将进入 Web AI 领域，使用 ONNX Runtime Web + Canvas 打造纯前端低成本的 YOLO 实时检测程序，别忘了继续关注！