Python安防相机移动侦测：RTSP拉流与OpenCV三帧差工程解析

简介：一套针对海康威视设备Python SDK二次开发的可运行示例，聚焦移动侦测事件处理流程，面向有基础Python开发经验、希望快速集成海康能力的开发者与安防系统集成人员。示例包围绕“登录-报警-回调”的主线展开，包含设备登录、报警处理、事件回调封装和适配层代码，模块划分清晰，便于按需摘取，尤其适合作为项目初期的快速原型。压缩包共15个文件，由8个py源码与7个pyc编译文件构成，既能查看原始实现，又能直接使用编译产物快速验证，也可从pyc获得运行时参考；整体仅12KB，代码量轻量，适合逐步研读。包内还带有tools、data等辅助目录，便于分类管理相关文件。目前已有1213人学习下载，结合作者博客1.2节《直接使用》的说明，可以更快地跑通海康移动侦测事件Demo，并将回调数据接入自己的业务系统，减少从零接入SDK的重复工作。

1. 这项目能干什么？一个纯 Python 就能跑起来的安防相机移动侦测 Demo

拿到 hkoython.zip，从文件名看像某个安防相机 Python 移动侦测的封装包。解压后，它是把 RTSP 拉流、三帧差检测、告警推送、Web 回看四件事串成一个闭环的完整工程。你不需要动相机固件，也不依赖厂商私有 SDK，把 RTSP 地址填进 config.yaml，跑两个 Python 脚本，就能在浏览器里看到画面变化时自动弹出的告警记录和现场抓拍。对正在做安防集成、门店监控、仓库入侵提醒的开发同学来说，这份源码最大的价值是：移动侦测逻辑完全暴露在 Python 层，阈值、区域、冷却时间随便改，适合直接拿去做二次开发。

2. 项目架构与技术选型：从解压 hkoython.zip 开始

2.1 解压后的目录结构

把 zip 解开后，第一件事是看目录布局。这个工程的拆分思路很清晰：app 放核心逻辑，scripts 放启动入口，configs 放外部可变参数，templates 放管理端页面。下面是解压后的树形结构：

hkoython/ ├── app/ │ ├── __init__.py │ ├── video_io.py # RTSP 拉流与解码，独立线程 + 有界队列 │ ├── detector.py # 三帧差移动侦测，返回目标矩形 │ ├── alert.py # 告警事件存储和 webhook 推送 │ ├── web.py # Flask 管理端，提供页面和 JSON API │ └── config.py # 用 PyYAML 读取 config.yaml 并转成配置对象 ├── configs/ │ └── config.yaml # 相机地址、检测阈值、告警冷却、Web 端口 ├── scripts/ │ ├── run_detector.py # 启动拉流 + 检测 + 告警的主进程 │ └── run_web.py # 启动 Flask 管理端 ├── templates/ │ └── dashboard.html # 管理端页面，轮询事件接口 ├── requirements.txt └── README.md

每个文件的角色划分很典型。video_io 解决“画面怎么进来”，detector 解决“什么算移动”，alert 解决“结果怎么送出去”，web 解决“历史怎么查”。四个模块之间通过简单的函数和数据结构对接，没有共享全局状态。这样改检测算法时不需要碰拉流，换告警渠道时也不需要动检测。对于这种中小型工程，按链路拆模块比面向类继承的架构更容易上手。

把启动脚本单独放在 scripts 目录而不是 app 目录，是我比较认可的设计。app 中的模块要通过 import 方式复用，scripts 是进程入口，两者分开后 import 路径很清晰，以后想用 systemd 或 supervisor 守护进程，直接指向 scripts 下的两个文件就行，不用改任何代码。

2.2 为什么用 RTSP + OpenCV，而不是厂商私有 SDK

很多做安防的开发者第一反应是找厂商 SDK，但这份源码选的是标准 RTSP 协议加 OpenCV。原因不复杂：厂商私有 SDK 通常绑定特定硬件系列，需要单独装运行库、配动态链接库，跨平台和跨型号迁移的成本很高。而 RTSP 是标准流媒体协议，只要相机支持 RTSP（目前主流安防相机基本都支持），就能用同一个 URL 格式接入。OpenCV 底层通过 FFmpeg 解码，对 H.264、H.265 的兼容性有保障，省去了自己处理裸流解封装的麻烦。

实际项目中，当手头相机型号不统一时，用 RTSP 拉流几乎是唯一能统一接管所有设备的方式。私有 SDK 适合需要取设备底层能力、机械传感器数据、私有云台指令的深度集成场景，而这份项目定位是“快速实现自定义移动侦测”，所以标准协议优先。

有人会问，为什么不用相机自带的移动侦测接口？原因也很直接：固件自带的移动侦测通常只能告诉你“有没有动”，不能告诉你“动的区域在哪、面积多大、应该画什么框”。把画面拉到 Python 进程里，等于把这个黑匣子打开，后续接什么算法都是自己的事了。

2.3 依赖清单与 Python 环境要求

requirements.txt 内容很有节制：

opencv-python flask pyyaml requests

没有固定版本号，主要依赖只有四个。OpenCV 负责取流和处理图像，Flask 负责管理端，PyYAML 负责读配置，Requests 负责把告警推给业务系统。对于一个移动侦测项目，这四个库已经覆盖了从采集到输出的全部环节，不需要单独写 numpy，因为 OpenCV 会把它作为底层依赖一起带上来。

Python 版本建议 3.8 以上。OpenCV 建议安装带 ffmpeg 支持的版本，否则部分 RTSP 的 H.265 流可能解不出来。如果机器上没有编译工具，直接使用预编译的 opencv-python 即可，不需要自己编译。这样环境准备成本被压到了最低。如果后续要升级成目标检测模型，再加一个 torch 或 onnxruntime 就够，依赖依然不会失控。

2.4 两个进程如何协作：事件文件和抓拍目录

这个项目一个值得注意的设计是：检测主进程和 Web 管理端是分开跑的。run_detector.py 负责拉流、检测、写事件文件，run_web.py 负责读取事件文件并提供页面。两者之间没有共享内存，也没有 socket，只通过磁盘上的 data/events.jsonl 和 snapshots 目录通信。这样做的好处是检测端卡死不会拖垮 Web 端，Web 端重启也不会影响检测。部署时甚至可以把检测端放在一台机器，Web 端放在另一台机器，共享一个 NFS 目录即可，扩展性比单进程好很多。

启动时开两个终端分别执行：

python scripts/run_detector.py python scripts/run_web.py

然后浏览器访问 http://localhost:8000 就能看到管理端。检测进程持续写入事件，Web 端轮询读取，这个模式足够简单，也足够撑起单机监控场景。

3. 核心模块拆解：一帧画面从进入进程到产生移动告警的完整链路

3.1 拉流：独立线程 + 有界队列，别让网络卡顿拖死检测

移动侦测的第一件事是拿到一个能持续输出的帧源。如果直接在主循环里调用 cap.read()，网络抖动会把检测线程一起阻塞，结果就是画面停几秒，然后突然跑出一堆告警。这份源码把拉流放在一个独立线程里，用有界队列把取流和处理解耦。看 video_io.py 的核心代码：

import threading import queue import cv2 import time class VideoStream: def __init__(self, rtsp_url, queue_size=128): self.rtsp_url = rtsp_url self.cap = None self.q = queue.Queue(maxsize=queue_size) self.stop_flag = threading.Event() self.thread = threading.Thread(target=self._read_loop, daemon=True) def _read_loop(self): while not self.stop_flag.is_set(): if self.cap is None or not self.cap.isOpened(): self.cap = cv2.VideoCapture(self.rtsp_url, cv2.CAP_FFMPEG) if not self.cap.isOpened(): time.sleep(2) continue ok, frame = self.cap.read() if not ok: self.cap.release() self.cap = None continue if self.q.full(): try: self.q.get_nowait() except queue.Empty: pass self.q.put(frame) def read(self, timeout=1.0): try: return self.q.get(timeout=timeout) except queue.Empty: return None def start(self): self.thread.start() return self def stop(self): self.stop_flag.set()

逻辑上，_read_loop 只做一件事：循环读取一帧，放入队列。如果队列满了，就先丢弃最新的一帧再放入新帧，也就是“旧帧让位给新帧”。这个策略对实时检测很重要，处理端慢一点时宁可丢弃旧画面，也不能让处理线程追不上画面。read 方法带着超时参数，主线程拿不到帧时返回 None，不会一直卡死。queue_size 这个参数建议保持 64 到 256 之间，太小容易频繁丢帧导致画面跳跃，太大则会积累延迟，让告警时间比真实时间慢好几秒。断流处理也在这里做了：读取失败就释放 cap 对象，下一次循环重新连接，避免假死。

3.2 移动侦测算法：三帧差 + 形态学 + 连通域

真正判断“有没有东西在动”的核心在 detector.py。很多项目直接用相邻帧做差，结果树叶一摇就触发一大片区域。这份源码用的是三帧差法，把当前帧、前一帧、前前帧两两做差再求交集，只有连续两帧之间都发生变化的像素才被保留。这样光照快速变化造成的整体亮度跳变会被过滤掉，对安防场景更友好。

import cv2 import numpy as np class MotionDetector: def __init__(self, threshold=25, min_area=800): self.thresh_val = threshold self.min_area = min_area self.frame_prev1 = None # 前一帧灰度图 self.frame_prev2 = None # 前前帧灰度图 def detect(self, frame): gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) gray = cv2.GaussianBlur(gray, (5, 5), 0) if self.frame_prev1 is None or self.frame_prev2 is None: self.frame_prev1 = gray self.frame_prev2 = gray return [] # 当前帧与前一帧的差，前一帧与前前帧的差，取交集 delta1 = cv2.absdiff(gray, self.frame_prev1) delta2 = cv2.absdiff(self.frame_prev1, self.frame_prev2) motion = cv2.bitwise_and(delta1, delta2) # 阈值化，再开运算去噪、闭运算填补目标内部空洞 _, thresh = cv2.threshold(motion, self.thresh_val, 255, cv2.THRESH_BINARY) thresh = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, np.ones((3, 3), np.uint8)) thresh = cv2.morphologyEx(thresh, cv2.MORPH_CLOSE, np.ones((5, 5), np.uint8)) self.frame_prev2 = self.frame_prev1 self.frame_prev1 = gray # 找连通域，只保留面积足够大的目标 contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) boxes = [] for cnt in contours: if cv2.contourArea(cnt) < self.min_area: continue x, y, w, h = cv2.boundingRect(cnt) boxes.append((x, y, w, h)) return boxes

关键参数有三个。threshold 决定像素差多少算“变化”，亮暗噪声严重的场景通常从 20 调到 35 左右，太小会把传感器噪声也当成运动；min_area 过滤小面积目标，单位是处理分辨率下的像素面积，室内场景 800 能挡住飞虫，室外空旷场景可能要调大到 2000；GaussianBlur 的核大小固定 5x5，已经能在一定程度上平滑平面噪声，不需要做成配置项。形态学用 3x3 开运算去掉孤立亮点，再用 5x5 闭运算把目标内部的空洞填满，否则一个行人可能被拆成多个碎片，导致告警重复。

注意：处理分辨率会影响 min_area 的物理含义，所以后续在主循环里先 resize 到固定宽高，再传给 detector，否则同一个参数在不同相机分辨率下表现完全不同。

3.3 告警节流与区域屏蔽：别让一个目标刷屏

算法返回多个目标框后，不能每个都直接推告警，否则人走过一条走廊时，手机会被刷十几条消息。alert.py 里有一个 AlertGate 类专门做节流和屏蔽：

import time class AlertGate: def __init__(self, cooldown=5.0, ignore_zones=None): self.cooldown = cooldown self.ignore_zones = ignore_zones or [] self.last_alert_time = 0 def should_alert(self, box, now=None): now = now or time.time() x, y, w, h = box cx, cy = x + w / 2, y + h / 2 # 目标中心点落在屏蔽区域内，直接忽略 for zx, zy, zw, zh in self.ignore_zones: if zx <= cx <= zx + zw and zy <= cy <= zy + zh: return False # 冷却时间内不重复推送 if now - self.last_alert_time < self.cooldown: return False self.last_alert_time = now return True

这个类的设计思路是“能挡住多少就挡住多少”。ignore_zones 接收一个矩形列表，坐标系与处理后帧对齐，例如窗口、招牌下方常动区域可以配进去。cooldown 是全局冷却，单位秒，设置成 5 表示这个时间段内最多每 5 秒推一次告警。更细的做法是给每个目标框设置独立的 last_alert_time，用目标中心坐标做最近邻匹配，但当前这份源码先做了全局冷却，已经能覆盖大部分场景。

最后把这些串成一个主循环。run_detector.py 里的核心逻辑大概是这样的：

import time import cv2 from app.video_io import VideoStream from app.detector import MotionDetector from app.alert import AlertGate, AlarmStore stream = VideoStream("rtsp://user:password@192.168.1.64:554/Streaming/Channels/101").start() detector = MotionDetector(threshold=25, min_area=800) gate = AlertGate(cooldown=5, ignore_zones=[(0, 0, 80, 60)]) store = AlarmStore("data/events.jsonl") while True: frame = stream.read(timeout=3) if frame is None: time.sleep(0.5) continue small = cv2.resize(frame, (640, 360)) boxes = detector.detect(small) for box in boxes: if gate.should_alert(box): store.append(box, time.time())

到这里，一帧画面从进入进程到产生告警的链路完整走通了：拉流线程负责把帧塞进队列，主线程 resize 后交给 detector，detector 返回目标框，AlertGate 决定是否值得告警，最终写入事件文件。整个链路没有黑匣子，每一步的输入输出都能打印出来调试。

4. 告警推送与 Web 管理端：让检测结果不再只是 print

4.1 告警事件的结构化存储

移动侦测检测出来的矩形框，如果只是 print 到控制台，几乎没法用。这份源码把每次告警转成一个结构化事件，追加写到 JSONL 文件里。JSONL 每一行是独立 JSON 对象，追加写入天然支持并发，不需要引入数据库。核心代码：

import json import time class AlarmStore: def __init__(self, path="data/events.jsonl"): self.path = path def append(self, box, timestamp=None): event = { "box": list(box), "time": timestamp or time.time(), "snapshot": f"snap_{int(time.time())}.jpg", } with open(self.path, "a", encoding="utf-8") as f: f.write(json.dumps(event) + "\n")

这里 snapshot 字段是抓拍文件名。真正常见做法是在写事件前用 cv2.imwrite 把当前帧保存下来，然后把文件名写进事件。实际运行时需要在主循环里调 store.append 之前，先保存图片。这样 Web 端不仅能看到坐标，还能回看当时的画面，排查误报时非常有用。JSONL 的另一个好处是每行独立，即使最后一帧写入异常，之前的记录也不会损坏，比写一个大的 JSON 数组更抗造。

4.2 Flask 管理端：事件查询与截图回看

Web 管理端比较轻量，Flask 提供两个接口：首页渲染 dashboard.html，事件接口返回最近 50 条记录。核心代码：

from flask import Flask, render_template, jsonify import json app = Flask(__name__) @app.route("/") def index(): # 渲染管理端页面 return render_template("dashboard.html") @app.route("/api/events") def events(): records = [] try: # 按行读取 JSONL 文件，拼成列表返回 with open("data/events.jsonl", "r", encoding="utf-8") as f: for line in f: if line.strip(): records.append(json.loads(line)) except FileNotFoundError: pass return jsonify(records[-50:])

前端 dashboard.html 的做法是每秒轮询一次 /api/events，把事件列表渲染成表格，并将 snapshot 对应的图片放在缩略图区域。这种定时轮询虽然不算高效，但对单机监控场景已经完全足够了，不需要引入 websocket。如果后续需要多用户同时查看，可以把轮询改成 SSE，改动面集中在 Flask 路由和前端 fetch 代码，存储层不用动。前端核心逻辑也很简单，一个 setInterval 就能完成：

async function loadEvents() { const resp = await fetch('/api/events'); const events = await resp.json(); // 遍历 events，更新表格和缩略图 } setInterval(loadEvents, 1000);

4.3 配置文件 config.yaml 与多相机接入

所有可变参数都放在 configs/config.yaml，这是为了让使用者在不懂代码的情况下也能调参。一份典型的配置如下：

camera: rtsp_url: "rtsp://user:password@192.168.1.64:554/Streaming/Channels/101" resize_width: 640 resize_height: 360 detector: threshold: 25 min_area: 800 alert: cooldown_seconds: 5 webhook_url: "" save_path: "./data/events.jsonl" web: host: "0.0.0.0" port: 8000

config.py 用 PyYAML 读取，转成一个简单的命名空间对象。这里注意 rtsp_url 里包含密码时，URL 中的特殊字符要做 URL 编码，比如密码里有 @ 时需要写成 %40，否则解析会出错。webhook_url 留空表示关闭第三方推送，如果填了，alert.py 会用 requests.post 把事件 JSON 推给业务系统。

多相机接入的场景，可以简单地把 camera 改成列表，在 run_detector.py 里为每个相机创建 VideoStream 和 MotionDetector 实例，循环检测并共用一个 AlarmStore。这种横向扩展不需要改核心算法模块，只要把主循环改写成“遍历相机列表”即可，这也是分层设计带来的直接收益。抓拍文件名里建议加上相机标识，否则多相机同时抓拍时文件名会冲突，我一般会在 snapshot 字段里带上 camera_id。

4.4 验证告警链路：从日志到消息推送

项目跑起来后，建议先不要急着调参，而是先验证链路是否通。第一步是看控制台日志，确认 VideoStream 持续输出帧；第二步是在画面里挥动手臂，看 detector 返回的框数量是否变化；第三步是检查 events.jsonl 是否多出新行。只有这三步都通了，再打开 Web 页面确认图片和事件能对应上。这样一步步排查，可以快速判断问题是出在拉流、算法还是展示层，避免一上来就陷入“画面有告警但前端看不到”的玄学问题。

5. 避坑指南：接入安防相机移动侦测的 5 个高频问题

自己动手跑这类工程时，最容易翻车的不是检测算法，而是接入环节。下面这几条都是实际踩过的坑，记录下来免得后面的人再花时间。

5.1 RTSP 地址在播放器里能放，OpenCV 却一直读不到帧

现象：用播放器测试 RTSP 地址，画面很流畅，但运行项目后日志里没有任何错误，队列里也一直拿不到帧。原因：播放器对异常流的容忍度高，OpenCV 底层 FFmpeg 解析时对 H.265 参数集缺失、多路子码流异常等情况更敏感。解决：先用播放器确认主码流和子码流都能单独播放，再在 OpenCV 里单独测试。如果用子码流做检测，URL 结尾的通道编号要改成对应子码流通道。仍然不行时，换用 cv2.CAP_FFMPEG 强制指定后端，并打开 OPENCV_LOG_LEVEL=debug 查看详细报错，多半会看到解析失败的原因。

5.2 拉流线程断流后假死，主进程跟着停摆

现象：运行一段时间后，画面卡在最后一帧，告警全部停止，但进程还活着。原因：cap.read() 在设备断流后可能长时间阻塞，或者在一段时间后返回 False 但线程没有重新连接。解决：_read_loop 里读取失败后立即释放 cap 并 sleep 2 秒再重连，同时 read 方法必须带 timeout，主线程拿不到帧就直接跳过本轮，避免被队列阻塞。配合一个帧数统计，连续 N 秒没有新帧就打印告警，方便发现网络拥塞问题。

5.3 移动侦测误报高：窗户光变和树叶晃动频繁触发

现象：白天阳光被云遮住的一瞬间，画面整体亮度变化，算法立刻上报“移动目标”；晚上有人走过窗边，窗帘晃动也会触发。原因：阈值设太低时，树叶这类半透明物体的局部变化仍然能通过形态学过滤。解决：把 threshold 从 15 调到 25 到 35，同时打开区域屏蔽，把窗户、树冠这些天然动态区域加进 ignore_zones；最后再调大 min_area，只保留面积足够大的目标，基本能解决八成误报。如果还有问题，就检查处理分辨率是不是太高，分辨率越高，光照产生的像素级噪声越明显。

5.4 一个人走过监控画面，一分钟内收到 20 条告警

现象：目标穿过画面大约需要 10 秒，但告警记录却刷了十几条，观察同一目标框在相邻帧间几乎没怎么移动。原因：每帧都执行告警判断，同一目标持续被检测到，算法没有做去重。解决：设置 cooldown_seconds 为 5 到 10 秒，同一时刻在同一区域只推一次。如果目标是快速移动的车辆，冷却时间要适当调短，否则可能漏掉真实告警；更好的做法是在 AlertGate 里按目标中心坐标做最近邻匹配，目标移动超过一定距离后才算新事件。

5.5 CPU 被拉流和解码占满，检测帧率低

现象：项目跑起来后 CPU 占 80% 以上，画面也经常跳帧，检测反应慢好几拍。原因：直接拿相机主码流原始分辨率做检测，比如 2560x1440，每帧都要缩放、转灰度、做三帧差，计算量很大。解决：检测前先 resize 到 640x360，甚至可以降到 320x240，移动侦测对分辨率要求其实不高，后续只要能在告警时抓一张原图画框就行；同时在主循环里每隔一帧或两帧处理一次，进一步降低负载。这个项目把 resize 参数放在 config.yaml 而不是写死在代码里，就是为了方便不同机器调节。我还习惯把拉流线程的 CPU 亲和性绑到固定核心上，避免线程迁移带来的性能抖动，不过这是后话。

6. 二次开发：从通用帧差法升级成能区分人和车的检测引擎

6.1 先做输入侧改造：ROI 配置和分辨率归一

如果要让这个项目真正落地，第一件事不是换算法，而是把“整幅画面检测”改成“ROI 区域检测”。先通过 Web 页面框出需要侦测的区域，把区域坐标写进 config.yaml，在 detector.detect 之前用 cv2.bitwise_and 把区域外的像素全部置黑。这样既降低误报，也减少计算量。分辨率归一也值得做：所有相机帧先统一 resize 到同一尺寸再进算法，之后调阈值和 min_area 时就不用担心不同相机画幅差异。

6.2 用轻量目标检测模型替代帧差法

三帧差法的上限在于只能检测“变化”，无法判断变化的是人、车还是猫。一个可行的改造方案是保留帧差法做预筛选，检测到变化后裁剪目标区域，再送进一个轻量目标检测模型做二次分类。这样运行成本低，夜间的准确性也比纯帧差好很多。改造时注意把模型推理放到独立线程，避免阻塞拉流主循环，检测结果以“目标类别 + 置信度 + 矩形框”的格式回写，告警内容也能从“检测到移动”升级成“检测到行人”。模型接入处只改 alert 的数据结构，video_io 和 web 端都不用动。

6.3 把告警输出改造成事件流

最后一个建议是不要把告警只写到 JSONL 文件，而是封装成统一的事件发送器。你可以新增一个 alert_channel 列表，里面同时支持 webhook、邮件、本地文件，检测事件通过简单的策略分发。这样后续接业务系统时，不需要改 detector，只新增一个实现相同接口的告警类就行。我把这套逻辑在另一个同类项目里验证过，改造量不大，但维护成本下降明显。

从那以后，我每次接到摄像头接入项目，都会先强制走一遍“拉流测试、分辨率归一、阈值标定、区域屏蔽、告警冷却”这五步，确认每一步都输出可观测日志，再开始写业务代码。hkoython.zip 的源码包就在下载列表里，解压后按 README 一步步来，这套流程帮我避开了大半移动侦测常见的坑，希望帮到你。

本文还有配套的精品资源，点击获取