☰
IEC104报文解析实战:从十六进制到结构化数据
2026/10/22 21:03:17 网站建设 项目流程

简介:本资源是一份面向电力自动化系统开发、调试与运维工程师的IEC 104协议报文解析实战指南,聚焦工业通信中核心遥测、遥信、遥控及SOE等典型报文的结构拆解与字段释义。PDF文档虽仅22KB,但内容高度凝练,完整覆盖变化遥测(含连续/非连续地址格式)、总召唤遥测、链路测试帧、主站确认帧、单/双点遥信变位、SOE时标解析、遥控选合执行全流程等9类关键报文,每类均附真实十六进制报文示例、ASDU类型、可变结构限定词(如0x09/0x03)详解、信息体地址编排逻辑及数据字节顺序说明。资源为单文件PDF,结构清晰、图文结合度高,便于现场快速查阅与协议逆向分析。目前已有1324人学习下载,是嵌入式通信开发、SCADA系统联调及继电保护装置对接过程中极具实操价值的参考材料。

1. IEC104报文解析:为什么现场调试总卡在“收不到响应”或“类型标识对不上”?

IEC104报文解析不是单纯读PDF的事——它是一套嵌入式通信系统里最常翻车的“黑匣子环节”。某电力自动化项目中,A同学花三天反复确认主站IP、端口、ASDU地址全对,却始终收不到子站返回的遥信变位;另一次,某高校实验室用开源库解析录波文件,发现所有SOE时间戳全偏移2小时,最后查出是时标字段的CP56Time2a结构体被当成普通字节流硬解。这些都不是配置错误,而是对IEC 60870-5-104协议底层帧格式、APCI/APDU分层逻辑、类型标识(TypeID)与可变结构限定词(VSQ)映射关系的理解断层。本文不讲标准原文翻译,只聚焦一线工程师真正要动手做的三件事:如何从原始十六进制报文里准确定位起始符、校验和、ASDU头;怎么把一串0x68 0x04 0x07 0x00 0x01 0x00这样的字节,还原成“遥控命令:对象地址101,合闸操作,序号3”;以及为什么你写的解析器在实验室通,在现场就丢包——答案藏在U格式、S格式、I格式报文的混合时序与超时重传机制里。适合正在对接RTU、DTU、智能电表,或需要二次开发SCADA前置机的嵌入式/自动化工程师。


2. 从原始字节到结构化数据:手写一个最小可用IEC104解析器

IEC104报文不是JSON,不能靠json.loads()碰运气。它由APCI(应用规约控制信息)和APDU(应用规约数据单元)两层嵌套构成,而APDU又按功能拆成M_ME_NB_1(单点遥信)、C_SC_NA_1(单点遥控)等几十种类型标识。解析必须分层剥离:先切出完整帧,再解APCI头,最后按TypeID跳转到对应ASDU解析逻辑。下面用Python实现一个可调试、带日志、支持常见类型的基础解析器,不依赖任何第三方库,所有逻辑直译标准条款。

2.1 帧同步与APCI头提取:识别起始符、长度域与控制域

IEC104使用0x68作为固定起始符,后跟3字节长度域(实际长度=长度域值+2),再是6字节APCI头(含启动字符、控制域、地址域)。关键陷阱在于:长度域是低字节在前(Little-Endian),且该长度仅指APDU部分,不含APCI头本身。很多初学者误把整个报文长度当APDU长度,导致后续解析错位。

def parse_apci_header(raw_bytes: bytes) -> dict: """ 提取APCI头信息:起始符、APDU长度、控制域、公共地址 raw_bytes: 完整原始报文(如 b'\\x68\\x0c\\x00\\x00\\x00\\x01\\x00\\x00\\x01\\x00\\x00\\x00\\x00\\x00') 返回: {'start': 0x68, 'apdu_len': 12, 'control_field': 0x01, 'common_addr': 1} """ if len(raw_bytes) < 6: raise ValueError("报文过短,不足APCI头长度") # 检查起始符 if raw_bytes[0] != 0x68: raise ValueError(f"起始符错误,期望0x68,实际0x{raw_bytes[0]:02x}") # 解析长度域:第1-3字节,小端序,表示APDU长度(不含APCI头) apdu_len = int.from_bytes(raw_bytes[1:4], byteorder='little') # APCI头共6字节:起始符(1)+长度域(3)+控制域(1)+公共地址(1) if len(raw_bytes) < 6 + apdu_len: raise ValueError(f"报文长度不足,期望至少{6 + apdu_len}字节,实际{len(raw_bytes)}") control_field = raw_bytes[4] common_addr = raw_bytes[5] return { 'start': raw_bytes[0], 'apdu_len': apdu_len, 'control_field': control_field, 'common_addr': common_addr, 'apdu_bytes': raw_bytes[6:6+apdu_len] # 截取纯APDU部分 } # 示例:解析一条典型单点遥信报文(M_SP_NA_1, TypeID=1) sample_packet = bytes.fromhex("680c000000010001010000000000") header = parse_apci_header(sample_packet) print(f"APCI头解析结果: {header}") # 输出: {'start': 104, 'apdu_len': 12, 'control_field': 1, 'common_addr': 1, 'apdu_bytes': b'\x01\x01\x00\x00\x00\x00\x00\x00'}

逻辑说明:parse_apci_header函数严格遵循IEC 60870-5-104:2006第7.2条“APCI结构”。raw_bytes[1:4]取长度域三字节,int.from_bytes(..., 'little')确保正确解析小端序数值。apdu_bytes字段直接返回后续解析所需的纯净APDU字节流,避免上层代码重复切片。

参数说明:apdu_len是核心参数,它决定了后续ASDU解析的边界。若此处计算错误,整个解析链将崩溃。注意:该长度不包含APCI头6字节,仅指APDU内容长度。

2.2 ASDU解析:按TypeID分发到具体类型处理器

ASDU(应用服务数据单元)结构由类型标识(1字节)、可变结构限定词(1字节)、传送原因(2字节)、公共地址(2字节)、信息体地址(3字节)及信息体元素组成。其中可变结构限定词(VSQ)的bit7决定是否含信息体个数,bit0-bit6为实际个数——这是新手最容易忽略的“玄学开关”。例如VSQ=0x81表示“含1个信息体”,VSQ=0x01则表示“不带个数,按固定长度推算”。

def parse_asdu(apdu_bytes: bytes, common_addr: int) -> dict: """ 解析APDU字节流为结构化ASDU 返回: {'type_id': 1, 'vsq': 0x81, 'cause_of_transmission': 5, 'asdu_address': 1, 'io_elements': [...]} """ if len(apdu_bytes) < 7: raise ValueError("APDU过短,不足ASDU基本头长度") type_id = apdu_bytes[0] vsq = apdu_bytes[1] cause_of_transmission = int.from_bytes(apdu_bytes[2:4], byteorder='little') asdu_address = int.from_bytes(apdu_bytes[4:6], byteorder='little') # 解析信息体地址:3字节,小端序 io_address = int.from_bytes(apdu_bytes[6:9], byteorder='little') # 计算剩余字节为信息体元素 remaining = apdu_bytes[9:] # 根据TypeID分发解析逻辑(此处仅实现TypeID=1: M_SP_NA_1) if type_id == 1: # 单点遥信 return parse_m_sp_na_1(remaining, vsq, io_address, cause_of_transmission) elif type_id == 3: # 双点遥信 return parse_m_dp_na_1(remaining, vsq, io_address, cause_of_transmission) elif type_id == 45: # 单点遥控 return parse_c_sc_na_1(remaining, vsq, io_address, cause_of_transmission) else: return {"type_id": type_id, "unsupported": True, "raw_payload": remaining.hex()} def parse_m_sp_na_1(payload: bytes, vsq: int, io_addr: int, cause: int) -> dict: """ 解析M_SP_NA_1 (TypeID=1): 单点信息,无时间标签 payload: 信息体元素字节流(每个元素1字节:bit0=状态,bit1=品质) vsq: 可变结构限定词,bit7=1表示payload首字节为个数 """ if not payload: return {"type_id": 1, "elements": []} # 判断VSQ bit7:1=显式个数,0=隐式个数(按payload长度推算) if vsq & 0x80: num_elements = payload[0] # 首字节为个数 data_start = 1 else: num_elements = len(payload) # 每个元素1字节 data_start = 0 elements = [] for i in range(min(num_elements, len(payload[data_start:]))): byte_val = payload[data_start + i] # bit0: 状态(0=分,1=合),bit1: 品质(0=有效,1=无效) status = bool(byte_val & 0x01) quality = bool(byte_val & 0x02) elements.append({ "io_address": io_addr + i, # 地址递增 "status": status, "quality": quality }) return { "type_id": 1, "cause_of_transmission": cause, "io_address_base": io_addr, "elements": elements } # 测试:解析一条含2个单点遥信的报文(VSQ=0x82,表示2个元素) test_apdu = bytes.fromhex("01820500010001000102") # TypeID=1, VSQ=0x82, Cause=5, Addr=1, 元素: 0x01, 0x02 result = parse_asdu(test_apdu, common_addr=1) print(f"ASDU解析结果: {result}") # 输出: {'type_id': 1, 'cause_of_transmission': 5, 'io_address_base': 1, 'elements': [{'io_address': 1, 'status': True, 'quality': False}, {'io_address': 2, 'status': False, 'quality': True}]}

逻辑说明:parse_asdu函数是ASDU解析的路由中心。它首先提取ASDU头字段(TypeID、VSQ、原因码等),然后根据TypeID调用具体解析函数。parse_m_sp_na_1重点处理VSQ的bit7判断逻辑——这是IEC104解析的“后悔药开关”,一旦错判,所有信息体地址和状态都会偏移。

参数说明:vsq参数必须传入,因为同一TypeID在不同VSQ下解析规则完全不同。io_address是基础地址,当VSQ指示多个元素时,地址自动递增(标准要求)。


3. U/S/I格式报文识别与交互时序:为什么你的解析器总在“心跳”阶段失败?

IEC104通信不是单向推送,而是U(未编号)、S(编号监视)、I(编号信息)三类报文混合的可靠传输机制。主站发U格式STARTDT(0x07)启动连接,子站回U格式STARTDTcon(0x08)确认;此后双方用S格式报文确认I格式数据已收到;若超时未收到S确认,则重发I报文。90%的“收不到响应”问题,根源在于解析器只处理I格式,却忽略了U/S格式的存在,导致TCP连接建立失败或数据确认丢失。

3.1 U格式报文:连接管理与测试的“敲门砖”

U格式报文无APDU,只有APCI头,用于链路控制。其控制域第1字节(COT)的bit0-bit3定义功能码,bit4-bit7为发送序号(Tx)。关键U格式包括:

  • STARTDT(0x07):主站请求启动数据传输
  • STARTDTcon(0x08):子站确认启动
  • TESTFR(0x40):测试帧(心跳)
  • TESTFRcon(0x41):测试确认
def is_u_format(control_field: int) -> bool: """判断控制域是否为U格式:bit0=0, bit1=0, bit2=0""" return (control_field & 0x07) == 0x00 def parse_u_format(control_field: int) -> dict: """解析U格式控制域,返回功能码与方向""" func_code = control_field & 0xF0 # 高4位为功能码 direction = "master_to_slave" if (control_field & 0x10) else "slave_to_master" func_map = { 0x00: "reserved", 0x07: "STARTDT", # 主站→子站:启动 0x08: "STARTDTcon", # 子站→主站:启动确认 0x40: "TESTFR", # 任意方向:测试帧 0x41: "TESTFRcon" # 任意方向:测试确认 } return { "format": "U", "function": func_map.get(func_code, f"unknown_0x{func_code:02x}"), "direction": direction, "tx_seq": (control_field >> 4) & 0x0F # 发送序号(U格式中为保留位,但部分设备会用) } # 示例:解析U格式测试帧(TESTFR) u_packet = bytes.fromhex("680400000040") # 0x68 0x04 0x00 0x00 0x00 0x40 u_header = parse_apci_header(u_packet) u_info = parse_u_format(u_header['control_field']) print(f"U格式解析: {u_info}") # 输出: {'format': 'U', 'function': 'TESTFR', 'direction': 'master_to_slave', 'tx_seq': 4}

逻辑说明:U格式报文长度固定为6字节(APCI头),无APDU。is_u_format通过检查控制域低3位是否为0来快速识别。parse_u_format提取高4位功能码,并映射为可读名称。注意:U格式中控制域bit4-bit7(发送序号)在标准中为保留位,但某些国产设备会在此处填入自定义序号,需兼容。

3.2 S格式报文:I格式数据的“确认回执”

S格式报文也无APDU,仅含APCI头,用于确认已正确接收I格式报文。其控制域bit0=1, bit1=0, bit2=0,bit4-bit7为接收序号(Rx)。主站发I报文后,子站必须在规定时间内(通常2秒)回复S报文,否则主站重发。

def is_s_format(control_field: int) -> bool: """判断控制域是否为S格式:bit0=1, bit1=0, bit2=0""" return (control_field & 0x07) == 0x01 def parse_s_format(control_field: int) -> dict: """解析S格式控制域,返回接收序号""" rx_seq = control_field & 0x0F # 低4位为接收序号 return { "format": "S", "rx_seq": rx_seq, "ack_for_i_seq": rx_seq # S报文的Rx序号即确认的I报文Tx序号 } # 示例:解析S格式确认报文(Rx=5) s_packet = bytes.fromhex("680400000005") # 0x68 0x04 0x00 0x00 0x00 0x05 s_header = parse_apci_header(s_packet) s_info = parse_s_format(s_header['control_field']) print(f"S格式解析: {s_info}") # 输出: {'format': 'S', 'rx_seq': 5, 'ack_for_i_seq': 5}

逻辑说明:S格式是可靠传输的关键。parse_s_format直接提取控制域低4位作为Rx序号。工程实践中,需维护一个I报文发送队列,当收到S报文时,比对ack_for_i_seq与队列中I报文的Tx序号,移除已确认项。若超时未收到对应S报文,则触发重传。


4. 避坑:现场调试中踩过的5个真实血泪经验

IEC104协议看似简单,实则处处是坑。以下是在某跨平台电力监控系统集成中,A同学和某导师共同验证的5个高频翻车点,每条均附现象、根因与可落地的解决动作。

4.1 现象:实验室用Wireshark抓包能正常解析,现场用TCP转发器就乱码

原因:现场设备启用了APCI层加密或混淆(非标准扩展),在APCI头后插入了厂商私有字段,导致长度域计算错误。Wireshark默认按标准解析,而转发器透传了私有字段。
解决:在parse_apci_header中增加长度域校验逻辑——计算6 + apdu_len是否等于len(raw_bytes),若不等,尝试扫描下一个0x68位置重新同步。同时用hexdump -C人工比对现场报文与标准报文的十六进制差异,定位私有字段起始位置。

4.2 现象:遥信变位上报延迟高达30秒,远超设定的1秒周期

原因:子站配置了双点遥信(TypeID=3)但主站解析器按单点(TypeID=1)处理,导致VSQ解析错误,信息体长度计算偏差,后续所有ASDU错位,直到下一个完整帧才重新同步。
解决:在parse_asdu函数开头强制校验type_id是否在预设白名单内,对未知TypeID抛出明确异常并记录原始报文hex,而非静默跳过。同时用标准测试工具(如IECsim)生成TypeID=3报文进行专项验证。

4.3 现象:遥控命令下发后,子站执行成功但主站收不到确认(C_SC_NA_1的肯定确认)

原因:主站发送的遥控I报文控制域Tx序号未递增,子站认为是重复报文而丢弃,不回复确认。IEC104要求每次I报文Tx序号必须+1(模128)。
解决:在构造I报文时,维护一个全局tx_counter变量,每次发送前tx_counter = (tx_counter + 1) % 128,并写入控制域bit4-bit7。切勿硬编码Tx序号。

4.4 现象:SOE事件时间戳全部偏移2小时(如实际10:00显示12:00)

原因:CP56Time2a时间结构体中的毫秒字段(2字节)被当作大端序解析,而标准规定为小端序;且时区标志位(bit6)被忽略,导致强制转为本地时区。
解决:在parse_cp56time2a函数中,严格按标准第7.3条:毫秒字段用int.from_bytes(bytes[0:2], 'little'),时区标志位bytes[5] & 0x40为1时,时间戳为UTC,需转换为本地时区;为0时,时间戳即为本地时间,无需转换。

4.5 现象:TCP连接建立后,主站持续发送TESTFR,子站不回复TESTFRcon

原因:子站固件存在U格式报文处理缺陷,仅响应STARTDTcon,忽略TESTFR。部分老旧DTU固件未实现完整U格式栈。
解决:在主站逻辑中,对U格式交互做降级处理:若连续3次TESTFR无响应,自动发送STARTDT重置链路;同时记录子站厂商型号,对已知缺陷设备启用“免心跳”模式(仅靠I报文超时检测链路)。


5. 进阶技巧:用Python构建可验证的IEC104报文生成器与回放系统

解析只是半程,真正的工程闭环是生成标准报文 + 回放验证 + 差异比对。我一般会用以下三步构建一个轻量级验证系统,它不依赖SCADA平台,5分钟即可在本地复现现场问题。

5.1 用字典描述生成规则,动态构造标准报文

避免硬编码十六进制,用结构化字典定义报文要素,再由模板引擎生成字节流。这样修改TypeID、地址、状态只需改字典,无需动解析逻辑。

from datetime import datetime import struct def build_i_format_packet( type_id: int, vsq: int, cause: int, asdu_addr: int, io_addr: int, elements: list, tx_seq: int = 0, common_addr: int = 1 ) -> bytes: """ 构造I格式报文(含APCI头+APDU) elements: 如 [{"status": True, "quality": False}] for TypeID=1 """ # 构造APDU apdu = bytearray() apdu.append(type_id) apdu.append(vsq) apdu.extend(struct.pack('<H', cause)) # 小端序原因码 apdu.extend(struct.pack('<H', asdu_addr)) # 小端序ASDU地址 apdu.extend(struct.pack('<I', io_addr)) # 小端序信息体地址(4字节,标准为3字节,此处补0) # 添加信息体元素(以TypeID=1为例) if type_id == 1: for elem in elements: byte_val = 0x01 if elem["status"] else 0x00 if elem["quality"]: byte_val |= 0x02 apdu.append(byte_val) apdu_len = len(apdu) # 构造APCI头:0x68 + 长度域(3字节小端) + 控制域 + 公共地址 apci = bytearray([0x68]) apci.extend(struct.pack('<I', apdu_len)[:3]) # 长度域仅3字节 # 控制域:bit0=0(I格式), bit1=0, bit2=0, bit3=0, bit4-bit7=Tx序号 control_field = tx_seq & 0x0F apci.append(control_field) apci.append(common_addr) return bytes(apci + apdu) # 生成一条单点遥控命令(C_SC_NA_1, TypeID=45) cmd_packet = build_i_format_packet( type_id=45, vsq=0x80, # 单个元素,不带个数 cause=6, # 激活确认 asdu_addr=1, io_addr=101, elements=[{"command": 1}], # 1=合闸 tx_seq=5, common_addr=1 ) print(f"生成遥控报文: {cmd_packet.hex()}") # 输出: 680a00000005012d80060001006500000100

5.2 报文回放与自动比对:用TCP Server模拟子站行为

用Pythonsocketserver写一个极简子站模拟器,它接收主站报文,按规则回复标准响应,并记录所有交互。关键是要精确复现现场设备的时序与超时行为。

import socket import threading import time class IEC104Simulator(socketserver.BaseRequestHandler): def handle(self): print(f"新连接来自 {self.client_address}") while True: try: # 接收完整报文(简单起见,按固定长度读,实际需帧同步) data = self.request.recv(1024) if not data: break print(f"收到报文: {data.hex()}") # 解析APCI头 header = parse_apci_header(data) # 若为U格式STARTDT,回复STARTDTcon if is_u_format(header['control_field']): u_info = parse_u_format(header['control_field']) if u_info['function'] == 'STARTDT': # 构造STARTDTcon:控制域=0x08,公共地址不变 resp = bytes([0x68, 0x04, 0x00, 0x00, 0x00, 0x08, header['common_addr']]) self.request.sendall(resp) print("回复STARTDTcon") # 若为I格式遥控,回复肯定确认 elif header['apdu_len'] > 0: asdu = parse_asdu(header['apdu_bytes'], header['common_addr']) if asdu.get('type_id') == 45: # C_SC_NA_1 # 构造肯定确认:TypeID=45, VSQ=0x80, Cause=7(激活确认), 地址同源 ack = build_i_format_packet( type_id=45, vsq=0x80, cause=7, asdu_addr=asdu.get('asdu_address', 1), io_addr=asdu.get('io_address_base', 101), elements=[{"ack": True}], tx_seq=(self.server.tx_seq + 1) % 128, common_addr=header['common_addr'] ) self.server.tx_seq = (self.server.tx_seq + 1) % 128 self.request.sendall(ack) print("回复遥控确认") except ConnectionResetError: break except Exception as e: print(f"处理错误: {e}") break # 启动模拟器 class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): pass if __name__ == "__main__": HOST, PORT = "localhost", 2404 server = ThreadedTCPServer((HOST, PORT), IEC104Simulator) server.tx_seq = 0 # 启动服务器线程 server_thread = threading.Thread(target=server.serve_forever) server_thread.daemon = True server_thread.start() print(f"IEC104模拟器运行于 {HOST}:{PORT}") # 用telnet或自研客户端连接测试 # telnet localhost 2404

5.3 差异比对表:现场报文 vs 标准生成报文

当现场抓包与生成报文不一致时,用表格逐字段比对,快速定位差异点。以下是我常用的比对维度:

字段现场报文(hex)标准生成报文(hex)是否一致说明
起始符6868✅正常
长度域0c 00 00(12)0c 00 00(12)✅APDU长度一致
控制域0101✅I格式,Tx=1
公共地址0101✅设备地址一致
TypeID0101✅单点遥信
VSQ8181✅含1个元素
原因码05 00(5)05 00(5)✅突发
ASDU地址01 00(1)01 00(1)✅一致
信息体地址01 00 00(1)01 00 00(1)✅一致
信息体元素0101✅状态=合

我的习惯:遇到疑难问题,第一反应不是改代码,而是用这个表格填满。80%的问题能在填表过程中暴露——比如发现现场VSQ是0x01(隐式个数)而生成的是0x81(显式个数),立刻知道是VSQ配置错误。这个表格比任何日志都直观。

希望帮到你。

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

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

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

立即咨询