☰
GB28181摄像机模拟器实战:从SIP注册到RTP推流
2026/10/15 20:24:07 网站建设 项目流程

简介:这是一份面向视频监控开发与测试人员的GB28181摄像机模拟工具,用于在无真实硬件的情况下验证国标平台的注册、心跳、视频请求与目录检索等交互流程。资源围绕SIP协议展开,涵盖Invite邀请的发起与响应、设备向平台注册并上报设备ID与位置信息、周期性心跳保活,以及平台请求实时视频时通过RTP/RTCP回传码流等关键环节,适合需要调试GB28181兼容性、搭建联调环境的中高级开发者。压缩包为rar格式,共5个文件,包含可执行程序、动态链接库、xml配置与txt说明文档,整体约423KB,体积轻便,解压后即可按说明配置SIP服务器地址与设备参数。目前已有977人学习下载,可帮助读者快速理解国标信令流程、掌握模拟器配置要点,并在软件层面完成监控系统的功能验证与排错。

1. 从一台不存在的摄像机说起:GB28181 模拟到底在模拟什么

做视频监控平台对接的工程师,大概率都遇到过这种局面:平台侧代码写完了,信令交互逻辑也捋顺了,可手头就是没有一台能听话的国标摄像机。真机要么在客户现场,要么在仓库里吃灰,要么型号太老、要么固件太新,总之你没法随时让它注册、推流、响应云台指令。这时候“GB28181 摄像机模拟”就成了刚需——用软件造出一台符合 GB/T 28181 规范的虚拟设备,让它像真机一样向平台注册、心跳、上报目录、响应 INVITE 并推送 PS 流。

它解决的核心问题不是“播放视频”,而是“让平台以为有一台真机在线”。适合三类人:一是做平台侧开发、需要稳定测试信令链路的;二是做设备侧固件、想对照标准验证自己实现的;三是做集成和运维、要复现现场故障却拿不到设备的。模拟器不是玩具,它得把 SIP 注册、目录查询、INVITE 协商、RTP 打包这几条链路都跑通,否则平台一个 401 或 488 就把你打回原形。下面按我实际搭过的一套最小可用方案,从协议骨架讲到能跑起来的代码和参数。

2. 拆开 GB28181 模拟器的四根支柱:SIP、目录、INVITE、RTP

2.1 为什么 SIP 注册是模拟器的第一道生死线

GB28181 的设备接入本质是 SIP 协议上的扩展。摄像机作为 SIP UA,向平台(SIP 服务器)发起 REGISTER,平台返回 401 带 WWW-Authenticate 头,设备用摘要认证重新 REGISTER,成功后平台记录设备在线。模拟器如果这一步过不去,后面全是空谈。

常见做法是用现成 SIP 栈,比如 Python 的pyVoIP或直接基于socket手写 SIP 报文。我一般会先手写一遍 REGISTER,把 Authorization 头的计算搞明白,再换库。因为摘要认证的response计算涉及 username、realm、nonce、uri、method、password 的 MD5 拼接,错一个字符就是 401 循环。

import hashlib def sip_digest(username, realm, password, method, uri, nonce): # HA1 = MD5(username:realm:password) ha1 = hashlib.md5(f"{username}:{realm}:{password}".encode()).hexdigest() # HA2 = MD5(method:uri) ha2 = hashlib.md5(f"{method}:{uri}".encode()).hexdigest() # response = MD5(HA1:nonce:HA2) response = hashlib.md5(f"{ha1}:{nonce}:{ha2}".encode()).hexdigest() return response

这段是摘要认证的核心。realm和nonce从平台 401 响应里取,uri通常就是sip:平台ID@平台IP:端口。注意method必须大写,uri要和 REGISTER 行里的 URI 完全一致,否则平台算出来的 response 对不上。参数上,设备 ID 一般用 20 位国标编码,前 8 位是行政区划,中间 10 位是行业编码,后 2 位是设备类型,模拟时随便编但要在平台侧配好对应关系。

2.2 目录查询与心跳:让平台觉得你“活着且有货”

注册成功后,平台会发 MESSAGE 请求查询设备目录(Catalog),模拟器要回 200 OK 并随后发 MESSAGE 上报目录 XML。目录里至少要有通道信息:通道 ID、名称、状态、父设备 ID。心跳则是设备定时发 MESSAGE 给平台,平台回 200 OK,超时未收到就标记离线。

import time import xml.etree.ElementTree as ET def build_catalog_xml(device_id, channels): root = ET.Element("Response") cmd = ET.SubElement(root, "CmdType") cmd.text = "Catalog" sn = ET.SubElement(root, "SN") sn.text = str(int(time.time())) dev = ET.SubElement(root, "DeviceID") dev.text = device_id sum_num = ET.SubElement(root, "SumNum") sum_num.text = str(len(channels)) dev_list = ET.SubElement(root, "DeviceList") for ch in channels: item = ET.SubElement(dev_list, "Item") ET.SubElement(item, "DeviceID").text = ch["id"] ET.SubElement(item, "Name").text = ch["name"] ET.SubElement(item, "Status").text = "ON" return ET.tostring(root, encoding="unicode")

这里SN是序列号,平台用它匹配请求和响应,每次要递增或至少不同。SumNum是通道总数,DeviceList里每个Item是一个通道。心跳间隔常见 60 秒,但测试时可以调到 10 秒加快验证。注意 XML 的根节点是Response还是Notify取决于你是回复查询还是主动上报,别搞混。

2.3 INVITE 协商:SDP 里藏着的编码和端口陷阱

平台要预览视频时,会发 INVITE 给设备,SDP 里带平台接收 RTP 的 IP、端口和期望的编码。模拟器要回 200 OK,SDP 里带上自己的发送 IP、端口和编码,然后开始推流。这里最容易翻车的是:平台 SDP 里的m=video端口是平台收流端口,设备 SDP 里的端口是设备发流端口,两者不能填反。

# 平台发来的 INVITE SDP 片段(示意) m=video 30000 RTP/AVP 96 a=rtpmap:96 PS/90000 a=recvonly

设备回复的 SDP 里m=video端口填自己准备发流的端口,a=sendonly,编码用 PS(Program Stream),payload type 通常 96。如果平台要求 TCP 被动或主动,还要在 SDP 里加a=setup:passive或a=setup:active,并在对应方向建立 TCP 连接。UDP 模式下直接往平台 IP 和端口发 RTP 包即可。

2.4 RTP 打包:PS 流不是随便塞进去就能播

GB28181 规定视频流封装成 PS 格式再拆成 RTP 包。PS 流有固定的起始码00 00 01 BA,里面包 PES 包,PES 里再包 H.264 或 H.265 的 NALU。模拟器如果只是把裸 H.264 塞进 RTP,平台解不出来,表现为“信令通了但黑屏”。

常见做法是用 FFmpeg 把一段本地 MP4 转成 PS 流,再按 RTP 打包发送。或者直接用ffmpeg -re -i test.mp4 -c:v copy -f mpegts udp://平台IP:端口推 TS 流,但 GB28181 要的是 PS over RTP,不是 TS over UDP。更稳妥的是用ffmpeg -re -i test.mp4 -c:v libx264 -f mpegts -输出到管道,再用代码拆 PS 包打 RTP。参数上,SSRC 要唯一,时间戳按 90kHz 递增,序列号每包加一。

3. 用 Python 搭一台最小可注册的模拟摄像机:从零到平台在线

3.1 环境准备与依赖选择:别一上来就上重框架

我试过用osip2、pjsip这类 C 库,编译麻烦,调试也重。后来固定用 Python 的socket+threading手写 SIP 报文,配合ffmpeg子进程推流。依赖只有标准库和 ffmpeg,干净。

# 环境准备 python3 -m venv venv source venv/bin/activate # 不需要额外 pip 包,标准库够用 # 确认 ffmpeg 可用 ffmpeg -version

ffmpeg用来生成测试视频流,如果手头有 H.264 裸流文件也可以直接用。Python 版本建议 3.8 以上,hashlib和socket的 API 稳定。

3.2 注册与心跳的完整代码骨架

下面是一个能跑通注册和心跳的最小骨架,省略了 INVITE 处理,先确保平台能看到设备在线。

import socket import threading import time import hashlib import re class GBDevice: def __init__(self, device_id, platform_ip, platform_port, password): self.device_id = device_id self.platform_ip = platform_ip self.platform_port = platform_port self.password = password self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind(("0.0.0.0", 0)) # 随机本地端口 self.local_port = self.sock.getsockname()[1] self.nonce = None self.realm = None self.register_ok = False def _md5(self, s): return hashlib.md5(s.encode()).hexdigest() def _build_register(self, with_auth=False): uri = f"sip:{self.platform_id}@{self.platform_ip}:{self.platform_port}" call_id = f"{int(time.time())}@{self.device_id}" from_tag = str(int(time.time())) cseq = 1 lines = [ f"REGISTER {uri} SIP/2.0", f"Via: SIP/2.0/UDP {self.local_ip}:{self.local_port};rport", f"From: <sip:{self.device_id}@{self.platform_ip}>;tag={from_tag}", f"To: <sip:{self.device_id}@{self.platform_ip}>", f"Call-ID: {call_id}", f"CSeq: {cseq} REGISTER", f"Contact: <sip:{self.device_id}@{self.local_ip}:{self.local_port}>", "Max-Forwards: 70", "Expires: 3600", "User-Agent: GBSim", ] if with_auth and self.nonce: response = self._md5( f"{self._md5(f'{self.device_id}:{self.realm}:{self.password}')}:" f"{self.nonce}:{self._md5(f'REGISTER:{uri}')}" ) auth = ( f'Authorization: Digest username="{self.device_id}", ' f'realm="{self.realm}", nonce="{self.nonce}", ' f'uri="{uri}", response="{response}", algorithm=MD5' ) lines.append(auth) lines.append("Content-Length: 0") lines.append("") lines.append("") return "\r\n".join(lines).encode() def register(self): self.sock.sendto(self._build_register(False), (self.platform_ip, self.platform_port)) data, _ = self.sock.recvfrom(4096) text = data.decode(errors="ignore") if "401" in text: self.realm = re.search(r'realm="([^"]+)"', text).group(1) self.nonce = re.search(r'nonce="([^"]+)"', text).group(1) self.sock.sendto(self._build_register(True), (self.platform_ip, self.platform_port)) data, _ = self.sock.recvfrom(4096) text = data.decode(errors="ignore") if "200" in text: self.register_ok = True print("注册成功") return self.register_ok def heartbeat_loop(self): while True: if self.register_ok: # 构造 MESSAGE 心跳,此处省略 XML 细节 pass time.sleep(60)

_build_register里Via头的rport让平台把响应发回源端口,Contact里的 IP 和端口要填模拟器实际可达的地址。Expires是注册有效期,平台可能返回更短的值,以平台为准。register方法先发无认证请求,收到 401 后提取realm和nonce,再发带认证的请求。注意recvfrom是阻塞的,实际用的时候要开线程或设超时。

3.3 参数怎么调:设备 ID、端口、心跳间隔的取值逻辑

设备 ID 用 20 位数字,前 8 位可以随便编一个行政区划码,中间 10 位用行业编码,最后 2 位用131表示摄像机。平台侧一般会配一个设备白名单或自动注册,测试时把模拟器的 ID 填进去就行。

本地端口用bind(("0.0.0.0", 0))让系统随机分配,避免和已有服务冲突。如果平台要求设备用固定端口,就改成具体端口号,但要确保防火墙放行。

心跳间隔默认 60 秒,测试时改成 10 秒能快速验证平台是否正常响应。但别改太小,比如 1 秒,平台可能限流或日志刷爆。注册有效期Expires设 3600,平台如果返回 200 且带Expires头,以平台返回的为准,模拟器要在到期前重新注册。

4. 避坑与排查:模拟器上线失败的五个血泪现场

4.1 注册一直 401:摘要认证的 response 算错了

现象:模拟器发 REGISTER,平台反复回 401,换密码也没用。

原因:response计算时uri和 REGISTER 行里的 URI 不一致,或者method没大写,或者realm/nonce提取时带了多余空格。

解决:把uri打印出来和 REGISTER 行逐字符对比,确保完全一致。realm和nonce用正则提取后strip()一下。另外确认密码是平台侧配的那个,不是设备 ID。

4.2 平台显示在线但目录是空的

现象:注册成功,平台设备列表里能看到设备,但点开没有通道。

原因:模拟器没响应 Catalog 查询,或者响应的 XML 格式不对,比如根节点用了Notify而不是Response,或者SumNum和实际Item数量对不上。

解决:抓包看平台发的 MESSAGE 里CmdType是不是Catalog,然后对照标准 XML 模板逐字段检查。SN要回显平台请求里的值,不能自己乱编。

4.3 INVITE 后平台不回 ACK,视频黑屏

现象:模拟器收到 INVITE,回了 200 OK,但平台不发 ACK,也不收流。

原因:200 OK 的 SDP 里m=video端口填成了平台收流端口,或者a=sendonly写成了a=recvonly,平台以为设备要收流而不是发流。

解决:设备 SDP 里m=video端口填自己发流的端口,方向属性用a=sendonly。如果平台要求 TCP,检查a=setup角色是否匹配。

4.4 RTP 包发出去了但平台解不出画面

现象:Wireshark 能看到 RTP 流,但平台播放器一直转圈或黑屏。

原因:PS 流封装不对,比如直接发了裸 H.264 NALU,没有 PS 头;或者 RTP 时间戳不递增,平台丢弃。

解决:用 FFmpeg 生成标准 PS 流,确认每个 RTP 包前有 PS 起始码。时间戳按 90kHz 递增,序列号每包加一。SSRC 要唯一且和 SDP 里一致。

4.5 心跳超时被平台踢下线

现象:设备在线一段时间后变成离线,日志显示心跳超时。

原因:心跳间隔大于平台配置的超时时间,或者心跳 MESSAGE 的SN没递增导致平台认为是重包。

解决:把心跳间隔调到平台超时时间的一半,比如平台 120 秒超时,心跳设 60 秒。SN每次加一,别复用。

5. 进阶技巧:用抓包和日志把模拟器调成“真机”

5.1 抓包定位信令断点:看 401、200、INVITE 的往返

模拟器调不通的时候,别猜,直接抓包。在模拟器所在机器上tcpdump -i any -w gb.pcap port 5060,然后用 Wireshark 打开。重点看三个地方:REGISTER 的 401 响应里realm和nonce是什么;200 OK 的Contact头里平台期望的设备地址是什么;INVITE 的 SDP 里平台收流端口和编码是什么。抓包能看到平台实际发的报文,比看文档快。

5.2 用真实 PS 流替换测试流:FFmpeg 生成与循环推流

测试流用 FFmpeg 生成一段带时间戳的 PS 流,循环推给模拟器。命令如下:

# 生成 10 秒测试视频并转为 PS 流,输出到管道 ffmpeg -re -f lavfi -i testsrc=size=1280x720:rate=25 -t 10 \ -c:v libx264 -preset ultrafast -tune zerolatency \ -f mpegts - | python3 ps_sender.py

ps_sender.py从标准输入读 PS 流,拆成 RTP 包发给平台。-re表示按实时速率读取,-t 10是 10 秒,testsrc是测试图案。这样平台看到的是动态画面,方便验证时间戳和序列号。

5.3 多通道模拟:一个进程管多个设备 ID 的线程模型

一个模拟器进程可以模拟多台摄像机,每个设备 ID 一个线程,各自维护 SIP 会话和 RTP 发送。线程间共享平台地址和密码,但Call-ID、From tag、SSRC要独立。用threading.Thread起多个GBDevice实例即可。注意端口不要冲突,每个设备用不同的本地端口或让系统随机分配。

5.4 验证清单:平台侧看到在线、目录、预览、云台四步全过

调完之后按这个清单过一遍:平台设备列表显示在线;点开设备能看到通道目录;点预览能出画面;发云台指令模拟器能回 200 OK。四步全过,模拟器就算合格。我一般会把抓包文件和日志留一份,下次平台升级或换型号时对照,省得重新踩坑。

希望帮到你。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询