走廊行人检测：YOLOv8 + ZeroMQ 实时推流实战

最近更新：2026-08-01 | 字数总计：3.6k | 阅读估时：15分钟 | 阅读量：次

前置条件
方案设计
数据标注与模型训练
客户端推流端
服务端检测端
1. 事件分段录像逻辑
总结
TODO / 后续优化
参考

Note

本文由 AI 依据开发手记草稿使用 write-blog skill 自动生成

家住顶楼，楼顶都是设备管道但仍有人上来晒衣服，穿过消防走廊时总有人不轻关防火门（防火门离书房只有三米），好几次写着代码就被砰的一声吓得心跳漏拍。于是想用走廊监控的实时视频检测行人，一旦检测到异常动静就发出告警。上周末手动标注了一批数据并用 YOLOv8 训练了检测模型，这套”截图 → 推流 → 检测 → 录像”的链路也跑通了，记录一下实现过程

前置条件

硬件：神眸摄像头（固件不支持 RTSP）、电脑 A（运行雷电模拟器 + 神眸 APP）、电脑 B（YOLOv8 GPU版推理服务器）
软件：Python（关键第三方库：ultralytics、opencv-python、mss、pyautogui、pyzmq）
数据：自建走廊行人数据集，用 make sense 在线工具标注

方案设计

神眸摄像头不支持 RTSP 实时视频流协议，无法直接拉流，所以绕开协议层、从画面源头入手：

电脑 A 用雷电模拟器运行神眸 APP，打开摄像头直播
pyautogui 定位模拟器窗口，mss 区域截图
截图经 JPG 压缩后，通过 pyzmq（ZMQ）推送到电脑 B
电脑 B 运行 ultralytics（YOLOv8）检测，识别行人并通过 opencv-python进行可视化以及按事件分段录像保存

Important

为什么不直接 RTSP/FFmpeg 推流？神眸固件不支持 RTSP，无法从外部拉流。屏幕截图方案对协议零依赖，任何能显示画面的 APP 都能接入，代价是画面分辨率受模拟器窗口大小限制。当前方案由豆包给出

整体数据流架构：

系统架构简图

ChatGPT 辅助生成的架构图更细化，可供对照：

系统整体架构分解图

数据标注与模型训练

采集走廊场景视频抽帧，用 make sense 标注行人边界框，导出 YOLO 格式数据集。训练只要一行命令：

1 2	yolo task=detect mode=train model=yolov8n.pt \ data=yolo-corridor-pedestrian.yaml epochs=30 workers=0 batch=16

PS：yolo-corridor-pedestrian.yaml 中指出了训练数据、验证数据的路径等。workers=0 规避 Windows 下 DataLoader 多进程的坑，batch=16 按显存调整。训练 batch 0 的预览如下：

训练 batch 0 的标注框与预测框预览

客户端推流端

客户端职责：截图 → 控帧 → 压缩 → 推送，关键点：

窗口定位：pyautogui.getAllWindows() 按标题关键词匹配雷电模拟器窗口，免去手填坐标
精准控帧：用时间戳判断帧间隔而非累加 sleep，避免帧率漂移
JPG 压缩：imencode 把 BGR 大图压成字节流，大幅降低局域网带宽占用
非阻塞发送：zmq.NOBLOCK 发送，缓存满时（zmq.Again）直接丢弃当前帧，保证画面始终是最新帧

客户端推流完整代码（mss 截图 + ZMQ 推送）

import mss
import cv2
import numpy as np
import zmq
import time
import pyautogui

# ==================== 配置区 ====================
SERVER_IP = "192.168.0.107"
SERVER_PORT = 5555
# 雷电模拟器窗口关键词
WINDOW_KEYWORD = "雷电模拟器"
# 画面裁剪偏移，剔除模拟器顶部标题栏、侧边按钮
# 左上角偏移：OFFSET_TOP为正值，表示画面左上角向下偏移裁剪、OFFSET_LEFT为正值，表示画面左上角向右偏移裁剪
OFFSET_TOP = 50
OFFSET_LEFT = 0
# 右下角偏移：OFFSET_BOTTOM为正值，表示画面右下角向下偏移裁剪、OFFSET_RIGHT为正值，表示画面右下角向右偏移裁剪
OFFSET_BOTTOM = -50
OFFSET_RIGHT = -57
# 推流帧率控制
TARGET_FPS = 13
ZMQ_HWM = 5
JPG_QUALITY = 70
FRAME_INTERVAL = 1.0 / TARGET_FPS
# =================================================

# ZMQ 推送套接字初始化
ctx = zmq.Context()
push_sock = ctx.socket(zmq.PUSH)
# 设置发送缓存上限，超过直接丢帧，防止延迟堆积
push_sock.setsockopt(zmq.SNDHWM, ZMQ_HWM)
push_sock.connect(f"tcp://{SERVER_IP}:{SERVER_PORT}")

# 查找雷电模拟器窗口
target_win = None
for win in pyautogui.getAllWindows():
    if WINDOW_KEYWORD in win.title:
        target_win = win
        break
if not target_win:
    raise Exception(f"未找到标题包含「{WINDOW_KEYWORD}」的雷电模拟器窗口，请先打开模拟器并运行神眸APP")

# 计算截图区域：窗口坐标 + 偏移裁剪
win_x, win_y = target_win.left, target_win.top # 窗口左上角坐标
capture_area = {
    "top": win_y + OFFSET_TOP,
    "left": win_x + OFFSET_LEFT,
    "width": target_win.width + OFFSET_RIGHT,
    "height": target_win.height + OFFSET_BOTTOM
}
# 高性能截图器
sct = mss.mss()
encode_param = [cv2.IMWRITE_JPEG_QUALITY, JPG_QUALITY]

print(f"模拟器截图推流启动，限制帧率{TARGET_FPS}FPS，ZMQ缓存上限{ZMQ_HWM}帧")
last_frame_time = time.time()

while True:
    now = time.time()
    # 精准时间戳控帧，避免sleep累积延时
    if now - last_frame_time < FRAME_INTERVAL:
        continue
    last_frame_time = now

    # 截取屏幕指定区域
    raw_img = np.array(sct.grab(capture_area))
    # mss输出BGRA格式，转为OpenCV标准BGR
    img_bgr = cv2.cvtColor(raw_img, cv2.COLOR_BGRA2BGR)

    # JPG压缩编码，降低网络传输体积
    ret, jpg_buffer = cv2.imencode(".jpg", img_bgr, encode_param)
    if not ret:
        continue

    # 非阻塞发送，服务端处理缓慢时缓冲区满直接丢弃当前帧
    try:
        push_sock.send(jpg_buffer.tobytes(), zmq.NOBLOCK)
    except zmq.Again:
        # 缓存已满，丢弃本帧，保证画面永远是最新帧
        pass

服务端检测端

服务端拆成三个进程，视频读取、图像推理、界面渲染、录像保存完全分离、互不阻塞：

进程	职责
主线程	ZMQ 拉流（送入 `input_q` 队列）、图像解码、任务分发、窗口渲染、事件状态机
推理子进程	读取 `input_q` 中的原始输入帧，YOLOv8 GPU 推理（`YOLO(MODEL_PATH)(frame, conf=CONF_THRESH, iou=IOU_THRESH)[0]`），结果写回 `output_q` 供主线程绘制以及继续分发给录像子进程
录像子进程	接收 `start`/`frame`/`end` 指令，行人事件独立保存为本地 MP4

事件分段录像逻辑

以”连续无行人超过 EVENT_GAP_SEC（1.5s）”作为事件结束判据：

检测到行人且未在录制 → 发送 start，新建按时间戳命名的 MP4
持续有行人 → 持续写入 frame
无行人但仍在缓冲期（< 1.5s）→ 继续写帧，防止行人短暂遮挡被误切成两个事件
无行人超过缓冲期 → 发送 end，关闭文件，等待下次行人出现

服务端完整代码（ZMQ 拉流 + YOLO 推理 + 事件录像）

import zmq
import cv2
import numpy as np
import signal
import os
import time
from datetime import datetime
from ultralytics import YOLO
from multiprocessing import Process, Queue, Event
from queue import Empty, Full

# ==================== 全局配置 ====================
LISTEN_PORT = 5555
MODEL_PATH = r"runs/detect/train-8/weights/best.pt"
CONF_THRESH = 0.6
IOU_THRESH = 0.5
INFER_QUEUE_MAX = 6
ZMQ_RECV_HWM = 6
ZMQ_RECV_TIMEOUT = 5
WIN_NAME = "YOLO Pedestrian Detection"
YOLO_VERBOSE_PRINT = False

# 录像配置：事件独立分文件核心参数
SAVE_VIDEO = True
VIDEO_QUEUE_MAX = 12       # 录像缓冲队列上限
VIDEO_FPS = 15
VIDEO_DIR = "./runs/detect/ped_event_video"
VIDEO_EXT = ".mp4"
FOURCC = cv2.VideoWriter_fourcc(*"mp4v")
EVENT_GAP_SEC = 1.5        # 连续无行人超过1.5s，判定事件结束，切分新文件
# YOLO处理后的画面显示帧率限制，避免imshow高频刷屏
SHOW_FPS = 30
# 计算帧间隔毫秒
SHOW_INTERVAL_MS = int(1000 / SHOW_FPS)

WAIT_PROC_TIMEOUT = 3.5    # 子进程优雅等待超时时间
SHOW_ORIGIN_FRAME = False   # 输出原始视频帧（未经过YOLO处理的），仅测试用
# =================================================

# 推理子进程：纯YOLO计算，无IO、无窗口
def infer_process(in_q: Queue, out_q: Queue, exit_evt: Event):
    import signal
    # Windows必须忽略SIGINT，屏蔽控制台广播中断
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    try:
        model = YOLO(MODEL_PATH)
        print("[推理进程] YOLO模型加载完成", flush=True)
    except Exception as e:
        import traceback
        print("[推理进程] 模型加载失败！", flush=True)
        print(traceback.format_exc(), flush=True)
        # 加载失败直接退出子进程
        return
    try:
        while not exit_evt.is_set():
            try:
                frame = in_q.get(timeout=0.05)
            except Empty:
                continue
            # 模型推理段单独捕获，打印AI相关报错
            try:
                res = model(frame, conf=CONF_THRESH, iou=IOU_THRESH, verbose=YOLO_VERBOSE_PRINT)[0]
                boxes = res.boxes.data.cpu().numpy() if res.boxes else np.empty((0,6))
                try:
                    out_q.put((frame, boxes), timeout=0.05)
                except Full:
                    # 输出队列满，丢弃当前帧，不阻塞，继续下一轮循环判断退出事件
                    pass
            except Exception as e:
                # CUDA OOM、张量错误、模型推理异常打印完整堆栈
                import traceback
                print("[推理进程] YOLO推理发生致命错误：", flush=True)
                print(traceback.format_exc(), flush=True)
                # 抛出异常，终止推理进程，不会静默卡死
                raise e
    except KeyboardInterrupt:
        pass
    # 排空输出队列，释放Queue Feeder线程
    while True:
        try:
            out_q.get_nowait()
        except Empty:
            break
    # 主动销毁模型+释放CUDA显存
    import torch
    del model
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
    # 加长GPU释放缓冲
    time.sleep(0.8)
    print("[推理进程] 退出", flush=True)

# ===================== 独立录像进程：负责文件创建、编码、磁盘写入 =====================
def video_writer_process(vid_q: Queue, exit_evt: Event):
    import signal
    # Windows必须忽略SIGINT，屏蔽控制台广播中断
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    os.makedirs(VIDEO_DIR, exist_ok=True)
    writer = None
    current_video_path = ""
    frame_size = None
    try:
        while not exit_evt.is_set():
            try:
                msg = vid_q.get(timeout=0.1)
            except Empty:
                continue

            msg_type = msg[0]
            if msg_type == "start":
                # 新行人事件：关闭上一段视频，新建独立文件
                if writer is not None:
                    writer.release()
                    print(f"[录像进程] 上次事件文件已保存：{current_video_path}", flush=True)
                draw_frame, start_ts = msg[1], msg[2]
                h, w = draw_frame.shape[:2]
                frame_size = (w, h)
                # 按事件起始时间命名，每个事件单独文件
                file_name = f"ped_event_{start_ts}{VIDEO_EXT}"
                current_video_path = os.path.join(VIDEO_DIR, file_name)
                writer = cv2.VideoWriter(current_video_path, FOURCC, VIDEO_FPS, frame_size)
                print(f"[录像进程] 新行人事件开始，创建视频：{current_video_path}", flush=True)
                writer.write(draw_frame)

            elif msg_type == "frame":
                # 事件持续中，写入当前帧
                draw_frame = msg[1]
                if writer is not None:
                    writer.write(draw_frame)

            elif msg_type == "end":
                # 行人事件结束，关闭视频
                if writer is not None:
                    writer.release()
                    print(f"[录像进程] 行人事件结束，文件保存完成：{current_video_path}", flush=True)
                    writer = None
                    current_video_path = ""
    except KeyboardInterrupt:
        pass
    # 全局退出，收尾当前视频
    if writer is not None:
        writer.release()
        print(f"[录像进程] 程序退出，收尾视频：{current_video_path}", flush=True)
    else:
        print("[录像进程] 退出", flush=True)

def main():
    exit_event = Event()
    # 推理队列
    input_q = Queue(maxsize=INFER_QUEUE_MAX)
    output_q = Queue(maxsize=INFER_QUEUE_MAX)
    infer_proc = Process(target=infer_process, args=(input_q, output_q, exit_event), daemon=True)
    infer_proc.start()
    print(f"[主线] YOLO推理进程启动")

    # 录像队列与进程
    vid_queue = None
    vid_proc = None
    if SAVE_VIDEO:
        vid_queue = Queue(maxsize=VIDEO_QUEUE_MAX)
        vid_proc = Process(target=video_writer_process, args=(vid_queue, exit_event), daemon=True)
        vid_proc.start()
        print(f"[主线] 录像进程启动，每个行人事件将独立保存至 {VIDEO_DIR}")

    def safe_put_end(q: Queue):
        try:
            q.put_nowait(("end",))
        except Full:
            # 丢弃最旧一帧，强制塞入end保证视频闭合
            try:
                q.get_nowait()
                q.put_nowait(("end",))
            except Empty:
                pass
    def stop_recording(vid_proc):
        # 信号处理函数 sigint_handler、finally 块统一加这段
        if vid_proc and vid_proc.is_alive():
            safe_put_end(vid_queue)
            # 短暂留时间给录像进程消费缓存
            time.sleep(0.2)

    # 主线事件状态机变量
    is_recording = False       # 当前是否正在录制行人事件
    no_ped_start_time = None   # 首次无行人的时间戳，用于判断事件结束

    # Ctrl+C 退出
    def sigint_handler(sig, frame):
        print("\n收到Ctrl+C，准备退出，等待资源收尾...", flush=True)
        while True:
            try:
                input_q.get_nowait()
            except Empty:
                break
        exit_event.set()
    signal.signal(signal.SIGINT, sigint_handler)

    # ZMQ 拉流套接字
    ctx = zmq.Context()
    pull_sock = ctx.socket(zmq.PULL)
    pull_sock.setsockopt(zmq.RCVHWM, ZMQ_RECV_HWM) # 接收缓冲上限4帧，抑制延迟堆积
    pull_sock.setsockopt(zmq.RCVTIMEO, ZMQ_RECV_TIMEOUT) # 缩短空等时间，从10ms→5ms
    pull_sock.setsockopt(zmq.LINGER, 0)
    pull_sock.bind(f"tcp://0.0.0.0:{LISTEN_PORT}")

    cv2.namedWindow(WIN_NAME, cv2.WINDOW_AUTOSIZE)
    print(f"[主线] 检测服务启动，监听端口 {LISTEN_PORT} | Ctrl+C / Q键退出")

    zero_frame = np.zeros((480,640,3), dtype=np.uint8)
    last_draw = zero_frame
    # 画面限帧：上一次渲染时间戳
    last_show_time = time.time()
    try:
        while not exit_event.is_set():
            frame = None
            # 接收推流JPG
            try:
                jpg_bytes = pull_sock.recv()
                arr = np.frombuffer(jpg_bytes, np.uint8)
                frame = cv2.imdecode(arr, cv2.IMREAD_COLOR)
            except zmq.Again:
                pass

            if frame is not None:
                # 送入推理（非阻塞，满则丢）
                try:
                    input_q.put_nowait(frame)
                except Full:
                    pass

            draw = None
            has_ped = False

            # 获取推理结果
            try:
                img, boxes = output_q.get_nowait()
                # 绘制检测框
                draw = img.copy()
                has_ped = len(boxes) > 0
                for b in boxes:
                    x1, y1, x2, y2 = map(int, b[:4])
                    conf, cls = b[4:]
                    cv2.rectangle(draw, (x1,y1), (x2,y2), (0,255,0), 2)
                    cv2.putText(draw, f"{conf:.2f}", (x1,y1-6), cv2.FONT_HERSHEY_SIMPLEX,0.45,(0,255,0),1)
                last_draw = draw
            except Empty:
                pass

            draw = draw if draw is not None else last_draw
            # 限帧控制：只有间隔达标才刷新窗口
            show_current_frame = False
            now = time.time()
            if SHOW_ORIGIN_FRAME:
                if frame is not None:
                    cv2.imshow(WIN_NAME, frame)
            if (now - last_show_time) * 1000 >= SHOW_INTERVAL_MS:
                if not SHOW_ORIGIN_FRAME:
                    cv2.imshow(WIN_NAME, draw)
                last_show_time = now
                show_current_frame = True

            # waitKey必须每轮都调用，否则窗口卡死、按键失效
            key = cv2.waitKey(5)
            if key & 0xFF == ord("q"):
                # 新增：清空推理输入队列，减少子进程待处理任务
                while True:
                    try:
                        input_q.get_nowait()
                    except Empty:
                        break
                break

            # ===================== 事件分段核心状态逻辑 =====================
            if not SAVE_VIDEO:
                continue
            if SHOW_ORIGIN_FRAME:
                continue
            if not show_current_frame: #控制输出帧率
                continue
            now_ts = datetime.now().strftime("%Y%m%d_%H%M%S")
            now_sec = time.time()

            if has_ped:
                no_ped_start_time = None
                if not is_recording:
                    # 新事件触发：发送start消息，新建视频
                    try:
                        vid_queue.put_nowait(("start", draw, now_ts))
                    except Full:
                        pass
                    is_recording = True
                else:
                    # 持续行人，正常写入帧
                    try:
                        vid_queue.put_nowait(("frame", draw))
                    except Full:
                        pass
            else:
                # 当前帧无行人
                if is_recording:
                    if no_ped_start_time is None:
                        no_ped_start_time = now_sec
                    # 计算无行人持续时长
                    gap_duration = now_sec - no_ped_start_time
                    if (gap_duration < EVENT_GAP_SEC):
                        # 还在缓冲期，继续写帧（防止遮挡误切分事件）
                        try:
                            vid_queue.put_nowait(("frame", draw))
                        except Full:
                            pass
                    else:
                        # 超过缓冲时间，判定事件结束，关闭当前视频
                        safe_put_end(vid_queue)
                        is_recording = False
                        no_ped_start_time = None

    finally:
        # 资源回收
        # 1. 先下发录像结束标记，给子进程处理队列时间
        stop_recording(vid_proc)
        exit_event.set()
        time.sleep(1.5)

        # 2. 优先等待子进程退出，再销毁网络、窗口资源
        print("[主线] 等待推理进程退出...", flush=True)
        if infer_proc.is_alive():
            infer_proc.join(timeout=WAIT_PROC_TIMEOUT)
            if infer_proc.is_alive():
                print("[警告] 推理进程超时未退出，强制终止", flush=True)
                infer_proc.terminate()
                infer_proc.join()

        if vid_proc is not None and vid_proc.is_alive():
            print("[主线] 等待录像进程退出...", flush=True)
            vid_proc.join(timeout=WAIT_PROC_TIMEOUT)
            if vid_proc.is_alive():
                print("[警告] 录像进程超时未退出，强制终止", flush=True)
                vid_proc.terminate()
                vid_proc.join()

        # 3. 子进程全部回收完毕，再销毁ZMQ、OpenCV窗口
        pull_sock.close()
        ctx.term()
        cv2.destroyAllWindows()
        print("[主线] 程序资源全部释放完毕")

if __name__ == "__main__":
    main()

项目地址：ultralytics

总结

这套方案用”屏幕截图 + ZeroMQ 推流”绕开了摄像头协议限制，把 YOLO 检测能力接到了任何能显示画面的监控设备上。多进程 + 队列的架构把推理、渲染、录像彻底解耦，事件分段录像让回放只关注有行人的片段，实测效果如下：

走廊行人检测实时效果

TODO / 后续优化

防火门关闭剧烈噪音检测（本人社恐😓，考虑到邻里和谐，暂不发出告警声，仅实验性质）
事件消息通过微信推送，第一时间收到有人上楼顶的通知
行人重识别（ReID），识别是否为同一人反复进出

参考

本文作者：muggledy
版权声明：本博客所有文章除特别声明外，均采用知识共享署名—非商业性使用—禁止演绎 4.0 国际许可协议 © CC BY-NC-ND 4.0 进行许可。非商用转载请注明出处！严禁商业转载！
本文链接：https://blog.muggledy.top/posts/c684fa62/

2026-08-01 该篇文章被 muggledy 打上标签: YOLO OpenCV 归为分类: 编程