电力用户用电信息采集系统通信协议集合与自动化测试工具实践
简介:这份资源面向电力行业技术人员、电表与集中器开发调试人员及智能电网相关专业学生,聚焦用电信息采集系统通信协议的落地应用。包内围绕376.1、376.2、376.3、DL/T 645与Q/GDW 698等主流规约,配套协议解析、调试与测试软件,可帮助读者验证报文格式、排查设备间数据交互问题。资源共89个文件,以37个txt说明与配置文本、8个dll动态库、7个exe可执行程序为主,另含pdf技术文档、doc说明、ini与config配置及数据库文件,压缩包约105.57MB,涵盖协议测试、集中器下行本地接口调试、698解析等模块。目前已有3553人学习下载,适合需要对照标准规约搭建测试环境、理解帧结构与命令码、积累通信排错思路的读者参考使用。
1. 电力用户用电信息采集系统通信协议集合加测试工具:从协议栈到自动化验证的落地路径
电力用户用电信息采集系统里,通信协议从来不是单一一种。一个地市级的采集主站,往下要面对集中器、专变终端、电能表、采集器等多类设备,往上要对接营销系统、计量生产调度平台。设备侧可能跑 DL/T 645、DL/T 698.45,本地通信可能用微功率无线、载波、RS-485,远程通道又涉及 4G、以太网、光纤。协议集合加测试工具,解决的就是“多协议共存时怎么统一验证”的问题。我做过几个采集类项目的联调,最深的体会是:协议本身不难,难的是不同厂家对同一份标准的理解偏差,以及现场没有趁手的测试工具时,排查一个通信故障要耗掉一整天。这套东西适合两类人:一是采集终端或主站的开发测试人员,二是现场运维中需要快速定位通信问题的工程师。如果你正在被“表计不抄、终端不上线、数据对不上”折磨,下面的内容应该能帮你省下不少返工时间。
2. 协议集合怎么拆:DL/T 645、DL/T 698.45 与上行通道的边界
2.1 三类协议在采集链路中的位置
电力用户用电信息采集系统的通信协议,按链路层次可以粗分为三层。第一层是主站与终端之间的远程通信,常见的有 DL/T 698.45、Q/GDW 1376.1,走 TCP 或串口。第二层是终端与电能表之间的本地通信,DL/T 645 是绝对主力,1997 版和 2007 版并存,2007 版又分规约和扩展规约。第三层是终端内部各模块之间的板级通信,比如 UART、I2C、CAN,这些在热词里也频繁出现,但在采集系统语境下,它们更多是终端硬件内部的通信协议,不直接暴露给主站测试。
我一般把协议集合拆成“对外协议”和“对内协议”两堆。对外协议是测试工具要重点覆盖的,因为现场故障八成出在这一层。对内协议在终端研发阶段用逻辑分析仪或示波器抓,测试工具只需要提供串口透传和报文记录能力即可。
DL/T 645 的帧结构是固定的:起始符 68H、地址域 6 字节、控制码、数据长度、数据域、校验和、结束符 16H。2007 版把数据标识从 2 字节扩到 4 字节,控制码也增加了安全认证相关的内容。DL/T 698.45 则面向对象,用 OAD、ROAD、OMD 等标识来寻址,帧头是 68H,但后面跟着长度域和链路控制域,整体比 645 复杂得多。
2.2 协议集合的代码组织方式
如果要把这些协议做成一个可复用的集合,我建议按“编解码器 + 传输适配 + 用例”三层来组织。编解码器只负责字节流和对象之间的转换,传输适配负责串口、TCP、UDP 的收发,用例层写具体的测试场景。下面是一个 DL/T 645 编解码器的 Python 骨架,依赖 pyserial 和构造/解析库。
# dl645_codec.py import struct from dataclasses import dataclass @dataclass class DL645Frame: addr: bytes # 6字节表号,低字节在前 control: int # 控制码 data: bytes # 数据域 # 2007版数据标识为4字节,1997版为2字节 def encode(self) -> bytes: # 起始符 + 地址 + 控制码 + 长度 + 数据 + 校验 + 结束符 body = self.addr + bytes([self.control]) + bytes([len(self.data)]) + self.data checksum = sum(body) & 0xFF return b'\x68' + body + bytes([checksum]) + b'\x16' @classmethod def decode(cls, raw: bytes): if raw[0] != 0x68 or raw[-1] != 0x16: raise ValueError("帧头帧尾错误") addr = raw[1:7] control = raw[7] length = raw[8] data = raw[9:9+length] checksum = raw[9+length] calc = sum(raw[1:9+length]) & 0xFF if calc != checksum: raise ValueError(f"校验和错误: {calc:02X} != {checksum:02X}") return cls(addr, control, data)这段代码里,addr是 6 字节 BCD 码,低字节在前,这是 645 规约的硬性要求。control控制码的 bit7 表示方向,bit6 表示从站应答标志,bit5 表示后续帧标志,bit4 表示有无数据域。data域的长度由第 9 个字节给出,但 2007 版的数据域里前 4 字节是数据标识,后面才是数据内容。校验和是地址域到数据域所有字节的算术和模 256,不包括起始符和结束符。
参数设置上,串口默认 2400bps、8 数据位、1 停止位、偶校验,这是 645 的经典配置。但 2007 版也支持 9600bps 甚至更高,现场遇到抄不到的情况,先确认波特率和校验位,再查表号是否匹配。
2.3 DL/T 698.45 的面向对象编解码要点
698.45 的帧结构比 645 多了一个长度域和链路控制域。帧头 68H 之后是 2 字节长度、2 字节链路控制、1 字节帧头校验,然后是用户数据区。用户数据区里才是真正的应用层报文,包含 OAD、数据单元等。下面是一个简化的 698.45 帧解析片段。
# dl698_codec.py def parse_698_frame(raw: bytes): if raw[0] != 0x68: raise ValueError("非698帧") length = struct.unpack('<H', raw[1:3])[0] link_ctrl = struct.unpack('<H', raw[3:5])[0] # 帧头校验:从长度域到链路控制域逐字节异或 hcs = 0 for b in raw[1:5]: hcs ^= b if hcs != raw[5]: raise ValueError("帧头校验失败") # 用户数据区从第6字节开始,长度为length user_data = raw[6:6+length] # 后续按OAD解析,此处省略具体对象映射 return {"length": length, "link_ctrl": link_ctrl, "user_data": user_data}698.45 的链路控制域里,bit0-3 是帧类型,bit4 是分帧标志,bit5 是扰码标志,bit6-7 是传输方向。帧头校验是异或校验,不是算术和,这一点和 645 不同,写测试工具时容易搞混。用户数据区里的 OAD 是 4 字节,前 2 字节是对象标识,第 3 字节是属性标识,第 4 字节是方法标识。测试工具需要维护一份 OAD 字典,否则解析出来的就是一堆十六进制。
3. 测试工具怎么搭:从串口透传到自动化用例
3.1 最小可用测试工具的四个模块
一个能用的采集协议测试工具,至少要有四个模块:物理层适配、协议编解码、用例执行、结果记录。物理层适配负责打开串口或建立 TCP 连接,协议编解码调用上一章的编解码器,用例执行按预设脚本发帧收帧,结果记录把原始报文和解析结果落盘。
我一般用 Python 的 pyserial 做串口,socket 做 TCP,pytest 做用例管理。下面是一个串口收发的最小示例,带超时和重试。
# transport.py import serial import time class SerialTransport: def __init__(self, port, baudrate=2400, parity='E', timeout=3): self.ser = serial.Serial( port=port, baudrate=baudrate, bytesize=8, parity=parity, stopbits=1, timeout=timeout ) def send_and_recv(self, frame: bytes, retry=2): for i in range(retry + 1): self.ser.reset_input_buffer() self.ser.write(frame) time.sleep(0.05) # 等待从站响应 resp = self.ser.read(256) if resp: return resp raise TimeoutError(f"重试{retry}次后仍无响应")parity='E'是偶校验,645 规约常用。timeout=3表示读超时 3 秒,现场如果信道质量差,可以调到 5 秒。reset_input_buffer在每次发送前清空缓冲区,避免上一帧的残留数据干扰。retry=2表示最多重试两次,加上首次一共三次,这是我在现场总结的比较稳妥的次数,再多会拖慢批量测试速度。
3.2 用 pytest 组织协议一致性用例
测试工具的核心价值在于能批量跑用例。我习惯把每个协议交互写成一个 pytest 用例,用参数化覆盖不同表号、不同数据标识。下面是一个读电能示值的用例。
# test_dl645.py import pytest from dl645_codec import DL645Frame from transport import SerialTransport @pytest.fixture def transport(): return SerialTransport(port='COM3', baudrate=2400) @pytest.mark.parametrize("addr", [ b'\x01\x00\x00\x00\x00\x00', b'\x02\x00\x00\x00\x00\x00', ]) def test_read_energy(transport, addr): # 读当前正向有功总电能,数据标识 00 01 00 00 frame = DL645Frame(addr=addr, control=0x11, data=b'\x00\x01\x00\x00') resp = transport.send_and_recv(frame.encode()) parsed = DL645Frame.decode(resp) assert parsed.control == 0x91, "控制码应为0x91(从站应答+有数据)" assert len(parsed.data) >= 5, "数据域至少包含4字节标识+1字节数据"control=0x11是主站读数据请求,bit7=0 表示主站发出,bit4=1 表示有数据域。从站应答的控制码是0x91,bit7=1 表示从站发出,bit4=1 表示有数据域。数据标识00 01 00 00对应正向有功总电能,这是 645 2007 版的标准标识。断言里检查控制码和数据长度,能快速发现从站是否按规约应答。
参数化里的表号是 BCD 码,b'\x01\x00\x00\x00\x00\x00'表示表号 000000000001。现场如果表号是 12 位,需要按低位在前的方式转换。我见过不少新手直接把表号字符串转成 ASCII 发出去,结果一个都抄不到,血泪经验。
3.3 自动化测试工具的结果记录与回放
测试工具跑完用例,结果要能回放。我一般把每次收发的原始报文按时间戳记到 SQLite 里,字段包括用例名、发送帧、接收帧、耗时、解析结果。下面是一个记录表的建表语句。
CREATE TABLE protocol_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, case_name TEXT NOT NULL, ts DATETIME DEFAULT CURRENT_TIMESTAMP, tx_frame BLOB, rx_frame BLOB, elapsed_ms INTEGER, parsed_json TEXT, status TEXT );tx_frame和rx_frame存原始字节,方便事后用 Wireshark 或自己写的解析器回放。elapsed_ms记录响应时间,现场排查时如果发现某个表响应时间超过 500ms,就要怀疑信道质量或表计处理能力。parsed_json存解析后的结构化数据,方便做统计。status存 pass/fail/timeout,批量跑完直接 group by 就能看出哪些表号有问题。
4. 避坑与排查:协议测试中最容易翻车的五个点
4.1 校验和算错却怪设备不响应
现象:发送读表请求后,从站毫无反应,换一块表也一样。原因:645 的校验和是地址域到数据域的算术和模 256,但很多人把起始符 68H 也算进去了,或者忘了模 256。698.45 的帧头校验是异或,不是算术和,两者搞混就会一直发错误帧。解决:在编解码器里单独写校验函数,用已知正确的报文做单元测试。我一般会拿一份厂家提供的标准报文,先跑通编解码再上现场。
4.2 串口参数不匹配导致乱码
现象:能收到数据,但解析出来全是乱码,或者帧头帧尾对不上。原因:波特率、数据位、停止位、校验位中有一项和表计不一致。645 表计出厂默认 2400bps、偶校验,但有些厂家改成 9600bps、无校验。解决:先用串口助手逐个参数试,或者用测试工具的自动波特率探测功能。我一般会写一个循环,遍历 1200/2400/4800/9600 和偶/无校验的组合,发同一帧看哪个能收到正确响应。
4.3 地址域字节序搞反
现象:单表测试正常,多表批量测试时只有部分表响应。原因:645 的地址域是 BCD 码,低字节在前,高字节在后。如果直接把表号字符串按顺序填进去,表号 000000000001 会变成 01 00 00 00 00 00,这是对的;但表号 000000000012 如果填成 12 00 00 00 00 00,实际表示的是 120000000000,完全错了。解决:写一个表号转换函数,把 12 位十进制字符串按两位一组反转。这个坑我在第一次做批量抄表时踩过,查了一下午才发现是字节序问题。
4.4 698.45 的分帧标志没处理
现象:读长数据时,从站返回多帧,测试工具只解析了第一帧,后续帧被丢弃。原因:698.45 的链路控制域 bit4 是分帧标志,置 1 表示后续还有帧。测试工具如果只收一次就返回,会丢失数据。解决:在传输层加一个循环,检查分帧标志,直到收到分帧标志为 0 的帧再拼接。拼接时注意用户数据区的顺序,不要按接收顺序简单追加,要按帧序号排序。
4.5 测试用例之间相互干扰
现象:单独跑某个用例通过,批量跑时随机失败。原因:串口缓冲区没有清空,上一个用例的残留响应被下一个用例读到。或者多个用例共用同一个串口对象,没有加锁。解决:每个用例开始前 reset_input_buffer,用例之间加 100ms 间隔。如果用 pytest-xdist 并行跑,每个 worker 要分配独立的串口或 TCP 连接,不能共享。
5. 进阶技巧:用协议模糊测试和差分对比提升工具价值
5.1 模糊测试在采集协议里的轻量落地
协议测试工具做到后面,最有价值的不是跑通标准用例,而是能发现厂家实现的边界问题。我一般会加一个轻量模糊测试模块,对控制码、数据长度、数据标识做变异,看从站会不会返回异常帧或者直接死机。下面是一个简单的变异函数。
# fuzz.py import random from dl645_codec import DL645Frame def mutate_frame(frame: DL645Frame, n=1): """对帧的控制码和数据域做随机变异""" mutated = [] for _ in range(n): ctrl = frame.control ^ (1 << random.randint(0, 7)) data = bytearray(frame.data) if data: idx = random.randint(0, len(data)-1) data[idx] = random.randint(0, 255) mutated.append(DL645Frame(frame.addr, ctrl, bytes(data))) return mutated这个函数对控制码的任意一位取反,对数据域的任意一字节随机替换。跑模糊测试时,重点观察从站是否返回异常控制码、是否超时、是否返回长度不符的帧。我遇到过某厂家表计在收到控制码 bit5 置 1 的帧时直接不响应,而标准里这个位是保留位,应该忽略。这种问题在常规用例里发现不了,但现场如果主站发了带保留位的帧,就会导致批量抄表失败。
5.2 差分对比:同一用例跑不同厂家设备
差分对比是我最推荐的一个技巧。同一份测试用例,分别跑 A 厂家和 B 厂家的表计,把响应帧的解析结果做 diff。正常情况下,除了表号和数据值,控制码、数据长度、数据标识的应答格式应该一致。如果发现 A 厂家返回的数据域比 B 厂家多一个字节,或者控制码的某个位不一样,就要查规约原文确认谁对谁错。
我一般会把差分结果输出成表格,按字段对比。下面是一个对比表的示例结构。
| 字段 | A厂家响应 | B厂家响应 | 规约要求 | 判定 |
|---|---|---|---|---|
| 控制码 | 0x91 | 0x91 | 0x91 | 一致 |
| 数据长度 | 5 | 6 | 5 | B厂家多1字节 |
| 数据标识 | 00 01 00 00 | 00 01 00 00 | 00 01 00 00 | 一致 |
| 数据内容 | 4字节 | 5字节 | 4字节 | B厂家多1字节 |
这张表一出来,B 厂家的问题就藏不住了。差分对比的好处是不需要你事先知道正确答案,只要两家实现不一致,就至少有一家有问题。当然,也可能两家都错,所以最终还是要回到规约原文。
5.3 把测试工具做成可持续积累的资产
我做了几年协议测试,最大的教训是:不要每次项目都重写工具。把编解码器、传输适配、用例模板、模糊测试、差分对比做成独立的模块,新项目来了只写设备适配层和用例参数。我现在的习惯是,每遇到一个新的协议变种或厂家私有扩展,就在编解码器里加一个分支,并补一个最小用例。这样工具越用越厚,下次遇到同类问题,直接跑历史用例就能定位。
另外,测试结果一定要落盘并版本化。我一般用 SQLite 加一个简单的 Web 界面,按项目、设备型号、固件版本、测试时间做筛选。现场运维的人拿到一个不抄表的终端,先查历史测试记录,如果同型号同固件版本之前跑过且通过,就重点查现场接线和信道;如果之前就没跑过,那就先跑一遍标准用例。这个习惯帮我省了很多来回沟通的时间。
希望帮到你。
本文还有配套的精品资源,点击获取