Python ONVIF取流工具包：自动发现摄像头、RTSP拉流与截图

简介适合网络视频监控与安防设备接入开发者的实用资源包围绕通过 Python 与 ONVIF 协议获取 RTSP 视频流这一核心需求给出从设备认证、取流到抓图截图的可落地代码。资源共三个文件包括可直接运行的 Python 脚本、OpenCV 调用方法说明文本以及第三方调用软件安装包压缩包仅20.35MB便于快速部署和学习。已有2152人学习。方案基于 pyonvif 库完成 ONVIF 设备连接与配置获取得到真实直播地址再借助 OpenCV 读取视频帧并保存图片文档部分补充了取流参数设置与注意事项安装包则适合在无编程环境下预览摄像头画面。对需要打通跨品牌网络摄像机、搭建视频分析原型或维护监控系统的技术人员可从中直接获得关键代码思路、环境依赖说明和排错要点减少自行摸索时间。1. 先定位这个 Python ONVIF 取流包到底解决什么问题做安防设备接入的开发者大概都经历过这种场景项目里要对接一批网络摄像头厂家SDK只支持WindowsLinux环境下拿不到视频码流翻遍文档也没找到RTSP地址怎么拼。这个资源包里的Python脚本就是为了终结这种局面——它走ONVIF标准协议自动发现设备、完成鉴权、拿到RTSP取流地址再用OpenCV把视频帧抓下来保存成截图。整个流程绕开了厂商私有SDK一套代码兼容不同品牌的ONVIF设备。适合两类人一是做监控平台、安防系统集成的后端开发需要把多品牌摄像头统一接入二是做视觉算法的同学经常要批量抓取监控画面作为数据集。项目的核心价值在于把ONVIF的鉴权细节和RTSP地址获取封装成了可直接调用的方法你不用去啃协议文档改改IP和账号密码就能出图。2. 架构与选型从设备发现到截图落地要打通哪几层2.1 ONVIF 协议的四段链路每段对应什么工具ONVIF协议并不是一个单独接口它是一组基于SOAP的Web服务协议集合。要把一台摄像头的视频流拿到手上整条调用链分为四个环节缺一不可。第一个环节是设备发现。摄像头接入局域网后并不广播自己的IP客户端要通过WS-Discovery协议发送组播探测报文设备收到后返回ProbeMatch响应从中解析出设备型号和Web服务地址。这个环节可以自己构造SOAP报文也可以借助现成的库去发。第二个环节是设备鉴权。拿到设备服务地址后需要调用GetSystemDateAndTime、GetDeviceInformation这些方法验证账号密码。这里有个常见坑ONVIF要求客户端时间与设备时间偏差在容忍范围内否则鉴权失败后面会专门讲。第三个环节是媒体服务协商。认证通过后进入Media服务先调用GetProfiles拿到视频编码配置的ProfileToken再调用GetStreamUri传入ProfileToken和码流类型设备返回一个带鉴权参数的RTSP地址。第四个环节才是真正的视频码流拉取。拿到RTSP地址后用OpenCV的VideoCapture或者FFmpeg去建流、解码、抓帧。这个包在设计上把四段链路分别封装成模块便于单独调试——排查问题的时候能直接定位到是协议层挂了还是拉流层挂了。2.2 技术栈与目录结构一个文件负责一个职责这个工具包基于Python 3.8开发核心依赖有三个onvif-zeep负责ONVIF协议的SOAP通信OpenCV负责RTSP拉流与图像处理requests兜底处理一些HTTP请求。onvif-zeep是目前Python生态里对ONVIF协议支持最完整的库底层用zeep解析WSDL文档并动态生成客户端对开发者比较友好。onvif_rtsp_tool/ ├── main.py # 命令行人机交互入口 ├── config.yaml # 设备IP、账号、码流参数配置 ├── requirements.txt # 依赖清单 ├── core/ │ ├── __init__.py │ ├── discovery.py # WS-Discovery 设备发现模块 │ ├── authenticate.py # ONVIF 鉴权与服务初始化 │ ├── stream_uri.py # RTSP 地址协商与拼接 │ └── snapshot.py # 视频码流抓帧与截图保存 ├── utils/ │ ├── logger.py # 日志输出 │ └── time_sync.py # 设备时间同步检查 └── output/ # 截图输出目录目录分层逻辑比较直白main.py只做参数解析和流程编排core目录下每个文件对应协议链路的一环utils里放的是跨模块复用的工具函数。二次开发时如果要增加录像功能在core目录加一个recorder.pymain.py里注册一下命令就行。3. 核心模块拆解设备发现、鉴权、RTSP 取流、抓帧截图全流程3.1 设备发现组播探测报文与响应解析设备发现模块的核心逻辑是发送WS-Discovery组播消息并监听设备响应。这个方法的实现并不复杂难点在SOAP报文的格式必须符合标准以及响应报文的解析要兼容不同厂商的字段差异。# core/discovery.py import socket import xml.etree.ElementTree as ET WS_DISCOVERY_ADDR 239.255.255.250 WS_DISCOVERY_PORT 3702 PROBE_MESSAGE ?xml version1.0 encodingUTF-8? e:Envelope xmlns:ehttp://www.w3.org/2003/05/soap-envelope xmlns:whttp://schemas.xmlsoap.org/ws/2004/08/addressing xmlns:dhttp://schemas.xmlsoap.org/ws/2005/04/discovery xmlns:dnhttp://www.onvif.org/ver10/network/wsdl e:Header w:MessageIDuuid:84ede3de-8dec-4af0-9d0c-34c3c8c38555/w:MessageID w:Tourn:schemas-xmlsoap-org:ws:2005:04:discovery/w:To w:Actionhttp://schemas.xmlsoap.org/ws/2005/04/discovery/Probe/w:Action /e:Header e:Body d:Probe d:Typesdn:NetworkVideoTransmitter/d:Types /d:Probe /e:Body /e:Envelope def discover_devices(timeout3): sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((, WS_DISCOVERY_PORT)) sock.settimeout(timeout) sock.sendto(PROBE_MESSAGE.encode(), (WS_DISCOVERY_ADDR, WS_DISCOVERY_PORT)) devices [] try: while True: data, addr sock.recvfrom(65535) root ET.fromstring(data) namespaces { a: http://www.w3.org/2003/05/soap-envelope, d: http://schemas.xmlsoap.org/ws/2005/04/discovery } probe_match root.find(.//d:ProbeMatches/d:ProbeMatch, namespaces) if probe_match is not None: xaddrs probe_match.find(d:XAddrs, namespaces) if xaddrs is not None: devices.append({ ip: addr[0], xaddrs: xaddrs.text.strip(), raw_xml: data.decode(utf-8, errorsignore) }) except socket.timeout: pass finally: sock.close() return devices这段代码有几个关键设计。第一SOAP报文里的MessageID必须是UUID格式且每次请求都要重新生成否则部分设备会直接丢弃探测报文。第二绑定端口时必须加SO_REUSEADDR否则多网卡机器上会报地址被占用。第三XAddrs字段是设备Web服务的入口地址它决定了后续ONVIF请求往哪里发格式通常是http://192.168.1.64/onvif/device_service。解析时用ElementTree而不是正则匹配命名空间是因为响应报文的结构相对固定用路径查找更稳妥。抓到的raw_xml字段在调试设备兼容性问题时很有用定位设备不支持某些字段时可以对比不同品牌设备的原始响应差异。3.2 鉴权与媒体服务初始化客户端、拿 ProfileToken、换 RTSP 地址拿到设备Web服务地址后下一步是创建ONVIF客户端并完成鉴权。这里用onvif-zeep库能省掉大量SOAP模板代码。# core/stream_uri.py from onvif import ONVIFCamera def get_rtsp_url(host, port, username, password, profile_index0, stream_typeRTP_Unicast): # 创建ONVIF客户端指定设备地址、端口、账号、密码 camera ONVIFCamera(host, port, username, password) # 获取设备服务能力确认媒体服务是否可用 device_info camera.devicemgmt.GetDeviceInformation() print(f设备型号: {device_info.Manufacturer} {device_info.Model}) # 进入媒体服务获取所有视频编码配置 media_service camera.create_media_service() # 获取profiles列表每个profile对应一套编码参数和码流类型 profiles media_service.GetProfiles() if not profiles: raise RuntimeError(设备未返回任何Profile请检查编码配置) profile_token profiles[profile_index].token print(f使用Profile: {profile_token}, 编码: {profiles[profile_index].Encoding}) # 构造StreamUri请求参数 stream_uri_request media_service.create_type(GetStreamUri) stream_uri_request.ProfileToken profile_token stream_uri_request.StreamSetup { Stream: stream_type, # 码流类型: RTP_Unicast / RTP_Multicast Transport: {Protocol: RTSP} } # 发起请求拿到设备返回的RTSP地址 stream_uri media_service.GetStreamUri(stream_uri_request) rtsp_url stream_uri.Uri print(f设备返回的RTSP地址: {rtsp_url}) # 设备返回的Uri可能不含账号密码需要手动拼接 if username and password: from urllib.parse import urlparse, urlunparse parsed urlparse(rtsp_url) final_url urlunparse(( parsed.scheme, f{username}:{password}{parsed.netloc}, parsed.path, parsed.params, parsed.query, parsed.fragment )) else: final_url rtsp_url return final_url这段代码的流程很清晰先通过ONVIFCamera创建客户端它内部会加载设备WSDL并动态生成所有服务方法。GetDeviceInformation是首次连通性测试如果这里就抛异常说明账号密码或设备地址有问题。GetProfiles返回的ProfileToken是后续所有操作的关键参数。一个摄像头通常有主码流和子码流两组Profile主码流分辨率高适合录像子码流分辨率低适合预览。profile_index参数选0还是1决定了你最终拿到的是哪路视频码流。GetStreamUri方法的参数结构比较特殊StreamSetup是一个嵌套字典内部Transport字段必须显式指定为RTSP协议。这里有个值得注意的细节设备返回的Uri里通常不带账号密码如果你直接拿这个地址去OpenCV拉流大概率会收到401认证失败。我一般会在返回地址的netloc部分手动拼接username:password前缀这就是代码里urlparse那段做的事情。3.3 拉流抓帧OpenCV 从 RTSP 地址读取视频帧并保存截图RTSP地址拿到手之后剩下的问题是怎么稳定拉流、快速抓帧。很多教程直接写cap cv2.VideoCapture(rtsp_url)然后开个循环读帧但在真实项目中这种写法容易翻车——网络抖动会导致read阻塞十几秒甚至永远卡死。这个包里的snapshot模块加了几层防护。# core/snapshot.py import cv2 import time import os def capture_snapshot(rtsp_url, output_path, timeout10, max_attempts3): # 设置OpenCV的RTSP拉流参数关闭自动重连避免卡死 cv2.setUseOptimized(True) cap cv2.VideoCapture() cap.set(cv2.CAP_PROP_OPENCV_FFMPEG_CAPTURE_OPTIONS, rtsp_transport;tcp) cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) ret cap.open(rtsp_url) if not ret: raise RuntimeError(f无法打开RTSP流: {rtsp_url}) frame None start_time time.time() attempt 0 # 循环尝试抓帧直到超时或成功 while time.time() - start_time timeout and attempt max_attempts: ret, frame cap.read() if ret and frame is not None and frame.size 0: break attempt 1 time.sleep(0.5) cap.release() if frame is None: raise RuntimeError(f在{timeout}秒内未读到任何视频帧) # 生成输出目录并保存截图 os.makedirs(os.path.dirname(output_path), exist_okTrue) save_success cv2.imwrite(output_path, frame) if not save_success: raise RuntimeError(f截图保存失败: {output_path}) return output_path这里每一处配置都有实际意义。rtsp_transport;tcp强制使用TCP传输而不走UDP——UDP在弱网环境下丢包严重画面全是马赛克且容易断流TCP传输虽然延迟稍高但稳定性好很多适合截图这种非实时场景。CAP_PROP_BUFFERSIZE设成1是让OpenCV不要缓存太多历史帧确保读到的是最新画面。抓帧循环用了timeout加attempt双保险。有些设备在I帧间隔内read会返回False单纯重试就好有些设备断流后read会阻塞timeout机制保证不会无限等下去。保存用cv2.imwrite而不是PIL因为OpenCV返回的BGR格式直接交给imwrite无需转换省一次拷贝。3.4 让完整流程自动衔接main.py 的调度逻辑把上面三块串起来main.py做的事情就是读配置、发现设备、解析地址、抓帧保存。# main.py import argparse import yaml from core.discovery import discover_devices from core.stream_uri import get_rtsp_url from core.snapshot import capture_snapshot def main(): parser argparse.ArgumentParser(descriptionONVIF设备RTSP取流截图工具) parser.add_argument(--config, defaultconfig.yaml, help配置文件路径) parser.add_argument(--host, help设备IP覆盖配置文件) parser.add_argument(--output, defaultoutput/snapshot.jpg, help截图保存路径) args parser.parse_args() with open(args.config, r, encodingutf-8) as f: config yaml.safe_load(f) host args.host or config[device][host] port config[device].get(port, 80) username config[device][username] password config[device][password] # 未指定IP时先做设备发现 if not host or host auto: print(正在探测局域网中的ONVIF设备...) devices discover_devices(timeout3) if not devices: print(未发现任何设备请检查网络或手动指定IP) return for idx, dev in enumerate(devices[:5]): print(f[{idx}] {dev[ip]} - {dev[xaddrs]}) choice int(input(选择设备序号: )) selected devices[choice] host selected[ip] port 80 print(f目标设备: {host}:{port}) rtsp_url get_rtsp_url(host, port, username, password) print(fRTSP地址: {rtsp_url}) save_path capture_snapshot(rtsp_url, args.output) print(f截图已保存: {save_path}) if __name__ __main__: main()4. 环境搭建与运行照着这几步跑通拿到第一张摄像头截图4.1 安装依赖与验证环境先把环境准备好。建议用Python 3.8到3.10之间的版本我测试时用的是3.9.12。太新的Python版本在zeep库解析WSDL时可能会出现兼容性警告虽然不是致命错误但排查起来会多花时间。# 创建虚拟环境避免污染系统Python python3 -m venv onvif_env source onvif_env/bin/activate # 安装核心依赖 pip install onvif-zeep opencv-python pyyaml requests # 验证安装是否成功 python -c from onvif import ONVIFCamera; print(ONVIF OK) python -c import cv2; print(OpenCV, cv2.__version__)onvif-zeep装完之后第一次调用ONVIFCamera会触发WSDL文件的解析这个过程需要读设备地址所以务必保证网络能通。有些型号的设备WSDL地址是http://ip/onvif/device_service?wsdl如果设备禁用了WSDL访问初始化会直接失败需要先去设备网页后台确认ONVIF服务是开启状态。4.2 配置设备信息并运行第一条命令# config.yaml device: host: auto # 设为auto时自动发现设备也可以直接填IP如192.168.1.64 port: 80 # ONVIF服务端口多数设备用80部分品牌用8080 username: admin password: your_password profile_index: 0 # 0为主码流1为子码流 stream_type: RTP_Unicasthost字段设为auto时程序会先向局域网发组播探测。如果你的设备在另一个网段组播报文过不去这时直接填IP最省事。profile_index对于不熟悉设备编码配置的用户来说可能拿不准我的经验是先填0跑一次如果截图分辨率达不到预期再改成1看子码流。运行命令很直接python main.py --config config.yaml --output output/first_shot.jpg正常输出会依次显示设备型号、ProfileToken、RTSP地址和截图保存路径。整个过程从设备发现到出图一般在5秒内完成如果卡在某一环节多半是鉴权失败或网络不通。4.3 从一行命令到批量抓帧跑通单张截图之后可以顺手试一下批量抓帧。把snapshot模块的调用放到循环里每隔几秒截一张这就是一个最简监控抓帧工具。# examples/batch_snapshot.py import time from core.stream_uri import get_rtsp_url from core.snapshot import capture_snapshot rtsp_url get_rtsp_url(192.168.1.64, 80, admin, your_password) for i in range(10): filename foutput/snap_{time.strftime(%Y%m%d_%H%M%S)}_{i}.jpg capture_snapshot(rtsp_url, filename, timeout5) print(f已保存第{i1}帧 - {filename}) time.sleep(2)这段脚本在测试设备鉴权稳定性的时候很管用。如果中途某一次抓帧失败说明设备连接可能有异常需要检查网络质量或设备最大连接数限制。某些设备默认只允许四路并发连接端口占用满了之后新连接会被拒绝。5. 避坑记录鉴权失败、设备发现不到、RTSP 连不上的翻车现场5.1 ONVIF 鉴权报错时间偏差问题排查现象调用ONVIFCamera初始化或GetDeviceInformation时抛出zeep.exceptions.Fault: NotAuthorized或者Sender not authorized。原因设备端ONVIF服务要求客户端时间和设备时间误差在10秒以内。虚拟机的系统时间如果不准或者设备长期没有NTP同步两边时间差个几分钟很正常此时设备会直接拒绝所有ONVIF请求。解决先确认设备时间和电脑时间。如果设备网页后台支持NTP配置指向同一个NTP服务器如果设备没有NTP功能就把电脑时间手动校准后重试。这个工具包里在utils/time_sync.py里提供了一个思路在鉴权前先通过GetSystemDateAndTime读取设备时间如果偏差超过阈值就自动修正系统时间。实际项目中把这段逻辑集成到启动流程里很有必要。5.2 WS-Discovery 探测不到设备现象运行discover_devices()后返回空列表但设备确实在同一个局域网里用ping也能通。原因一是设备默认关闭了WS-Discovery应答功能部分品牌在出厂固件里把组播探测当作安全风险给屏蔽了二是电脑和设备不在同一个广播域尤其在公司复杂的网络环境里跨VLAN的组播报文无法到达设备三是防火墙拦下了UDP 3702端口的入站包。解决先用nmap -p 3702 --open 192.168.1.0/24确认端口有没有开放。如果设备没有开放3702端口直接放弃自动发现手动填IP走后面的流程。如果端口是开放的但探测不到抓包看看设备是否回了单播响应但被本机防火墙拦了。5.3 RTSP 地址拿到了但 OpenCV 拉流失败现象GetStreamUri成功返回了RTSP地址直接复制到VLC里能播但cv2.VideoCapture.open()返回False。原因最典型的问题是设备返回的RTSP地址需要鉴权而OpenCV的RTSP实现不会自动从URL中提取用户名密码。另一个常见情况是设备默认RTSP端口是554但Url返回的地址里用的是8000这种自定义端口部分设备在双网卡环境下返回了内网IP。解决在get_rtsp_url里手动拼接账号密码到URL中这已经在核心代码里做了。如果拼接后依然失败把OpenCV的传输协议从UDP改成TCP。我的习惯是无论设备返回什么地址都在拉流前强制拼接一遍账号密码并且显式设置rtsp_transport;tcp。5.4 设备返回的 RTSP 地址是 Anonymouse 匿名地址现象GetStreamUri返回的Uri形如rtsp://192.168.1.64:554/stream1看上去正常但设为不带鉴权的URL后OpenCV拉流时提示401 Unauthorized。原因ONVIF标准的GetStreamUri本身不负责鉴权设备返回的地址默认期望客户端在URL里携带账号密码或者使用RTSP的Authorization头。部分厂商在返回的地址里会带上?tokenxxx格式的凭证但大部分厂商就是干干净净一个裸地址。解决沿用5.3的方法手动在netloc位置嵌入账号密码。如果设备对URL里的特殊字符敏感先对密码做URL编码再拼接。5.5 长时间运行时帧中断read 会永久阻塞现象程序运行半小时后cap.read()第一次返回False之后继续调用就彻底卡住CPU占用率清零程序不退出也不报错。原因OpenCV底层的FFmpeg在RTSP连接断开后不会自动重连而是卡在等待状态。这是FFmpeg RTSP实现的一个老毛病网络波动或者设备重启后特别容易触发。解决给read加超时。OpenCV本身没有read超时参数只能在业务层做——每次read前记录时间超过设定值强制cap.release()后重新open。在这种场景下我的做法是把拉流逻辑封装成带重连机制的类内部维护一个状态机检测到超时就自动重新建流。6. 二次开发思路做成多设备轮询的截图服务把单个设备的取流逻辑扩展成多设备轮询服务是这个包最常见的改造方向。我在做一个模拟监控平台X项目时就把这套代码拆成了三部分采集器、调度器、存储层。采集器复用core目录下的模块每个设备一个独立进程互不干扰调度器用生产消费者队列管理采集任务存储层按设备ID和时间戳分目录保存截图。# examples/multi_device_collector.py from concurrent.futures import ThreadPoolExecutor from core.stream_uri import get_rtsp_url from core.snapshot import capture_snapshot devices [ {id: cam_001, host: 192.168.1.64, username: admin, password: pass1}, {id: cam_002, host: 192.168.1.65, username: admin, password: pass2}, ] def collect_one(device): rtsp_url get_rtsp_url(device[host], 80, device[username], device[password]) out_path foutput/{device[id]}/{device[id]}_latest.jpg capture_snapshot(rtsp_url, out_path, timeout8) return f{device[id]} done with ThreadPoolExecutor(max_workers4) as executor: results list(executor.map(collect_one, devices)) for r in results: print(r)多设备并发要注意连接数限制。设备端对并发RTSP连接数通常有限制4到8路不等超过上限的新连接会被拒绝。线程池大小不要超过设备允许的最大连接数保守起见设成4。另外每个设备第一次连接时都要走完整的ONVIF鉴权流程如果希望降低设备压力可以在全局缓存一份每个设备的ONVIFCamera实例和RTSP地址只在首次调用时获取。截图服务做好之后可以把输出路径接到Flask接口上对外提供按设备ID获取最新画面。这样一套基础监控快照系统就成型了。我自己在这类项目的习惯是先跑通单台设备再把取流逻辑做成独立服务最后才接存储和视图层。血泪总结ONVIF踩坑无数最深的坑就是时间同步和RTSP鉴权。从那以后我每次接入新设备都强制走一遍时间校准、端口探测、裸地址拉流三步排查每完成一步打一条日志稳定后基本不会再为接入发愁。希望帮到你。本文还有配套的精品资源点击获取