4. 解析SECS消息格式

任务目标

实现一个SECS-I/II消息解析器，能够从原始字节流中解析出消息头（Message Header）与数据项（Data Item），并正确识别SECS-II的数据格式编码（Format Code）。本案例聚焦于SECS-II消息体的标准格式，不涉及HSMS/SECS-I的传输层组包。

必要环境与依赖

使用Python 3.8+，仅依赖标准库struct和datetime，无需第三方包。所有代码在单个文件中完成。

SECS消息格式要点

  • 消息头（Message Header）：10字节固定长度，包含Session ID（2字节）、Header Byte 1（Stream/Function）、Header Byte 2（Wait Bit）、Message Length（4字节）等字段。本案例解析器聚焦于从完整SECS-II消息帧中提取消息头信息。
  • 数据项（Data Item）：SECS-II消息体由数据项组成，每个数据项包含：格式字节（Format Byte，高4位为格式编码，低4位为长度字节数）、长度字节（1/2/4字节，表示数据长度）、数据内容。
  • 格式编码（Format Code）：常见编码包括：0x00（List）、0x10（Binary）、0x20（Boolean）、0x30（ASCII）、0x40（JIS-8）、0x50（2字节有符号整数）、0x60（4字节有符号整数）、0x70（8字节有符号整数）、0x80（4字节浮点）、0x90（8字节浮点）。

完整实现代码

import struct
from datetime import datetime

class SECSMessage:
    """解析后的SECS消息对象"""
    def __init__(self, session_id, stream, function, wait_bit, body_bytes):
        self.session_id = session_id
        self.stream = stream
        self.function = function
        self.wait_bit = wait_bit
        self.body_bytes = body_bytes
        self.data_items = []

    def __repr__(self):
        return (f"SECSMessage(SessionID=0x{self.session_id:04X}, "
                f"S{self.stream}F{self.function}, WaitBit={self.wait_bit}, "
                f"DataItems={len(self.data_items)})")

def parse_header(header_bytes):
    """解析10字节SECS消息头"""
    if len(header_bytes) != 10:
        raise ValueError(f"消息头必须为10字节，实际{len(header_bytes)}字节")
    session_id = struct.unpack('>H', header_bytes[0:2])[0]
    stream = (header_bytes[2] & 0x7F) >> 1
    wait_bit = header_bytes[2] & 0x01
    function = header_bytes[3]
    length = struct.unpack('>I', header_bytes[6:10])[0]
    return session_id, stream, function, wait_bit, length

def parse_data_item(data, offset=0):
    """解析单个数据项，返回(格式码, 值, 新偏移量)"""
    if offset >= len(data):
        raise ValueError("数据偏移越界")
    fmt_byte = data[offset]
    fmt_code = fmt_byte & 0xF0
    len_bytes = fmt_byte & 0x0F
    offset += 1

    if len_bytes == 0:
        length = 0
    elif len_bytes == 1:
        length = data[offset]
        offset += 1
    elif len_bytes == 2:
        length = struct.unpack('>H', data[offset:offset+2])[0]
        offset += 2
    elif len_bytes == 4:
        length = struct.unpack('>I', data[offset:offset+4])[0]
        offset += 4
    else:
        raise ValueError(f"非法长度字节数: {len_bytes}")

    if offset + length > len(data):
        raise ValueError(f"数据长度越界: 需要{length}字节，剩余{len(data)-offset}字节")

    raw = data[offset:offset+length]
    offset += length

    # 根据格式码解码
    if fmt_code == 0x00:  # List
        value = []
        pos = 0
        while pos < length:
            child, pos = parse_data_item(raw, pos)
            value.append(child)
    elif fmt_code == 0x10:  # Binary
        value = list(raw)
    elif fmt_code == 0x20:  # Boolean
        value = [bool(b) for b in raw]
    elif fmt_code == 0x30:  # ASCII
        value = raw.decode('ascii', errors='replace')
    elif fmt_code == 0x40:  # JIS-8
        value = raw.decode('euc-jp', errors='replace')
    elif fmt_code == 0x50:  # 2字节有符号整数
        value = list(struct.unpack(f'>{length//2}h', raw)) if length >= 2 else []
    elif fmt_code == 0x60:  # 4字节有符号整数
        value = list(struct.unpack(f'>{length//4}i', raw)) if length >= 4 else []
    elif fmt_code == 0x70:  # 8字节有符号整数
        value = list(struct.unpack(f'>{length//8}q', raw)) if length >= 8 else []
    elif fmt_code == 0x80:  # 4字节浮点
        value = list(struct.unpack(f'>{length//4}f', raw)) if length >= 4 else []
    elif fmt_code == 0x90:  # 8字节浮点
        value = list(struct.unpack(f'>{length//8}d', raw)) if length >= 8 else []
    else:
        raise ValueError(f"未知格式码: 0x{fmt_code:02X}")

    return (fmt_code, value, offset)

def parse_secs_message(full_message):
    """解析完整SECS消息（含10字节消息头 + 消息体）"""
    if len(full_message) < 10:
        raise ValueError("消息长度不足10字节")
    session_id, stream, function, wait_bit, body_len = parse_header(full_message[:10])
    body = full_message[10:10+body_len]
    if len(body) < body_len:
        raise ValueError(f"消息体不完整: 声明{body_len}字节，实际{len(body)}字节")

    msg = SECSMessage(session_id, stream, function, wait_bit, body)
    if body_len > 0:
        pos = 0
        while pos < body_len:
            fmt_code, value, pos = parse_data_item(body, pos)
            msg.data_items.append((fmt_code, value))
    return msg

# ========== 验证示例 ==========
if __name__ == '__main__':
    # 构造 S1F1 (会话ID=0x0001, WaitBit=0)
    header = struct.pack('>HBBHI', 0x0001, 0x02, 0x01, 0, 0)
    msg = parse_secs_message(header)
    print(f"解析结果: {msg}")
    assert msg.session_id == 1 and msg.stream == 1 and msg.function == 1

    # 构造 S6F11 带数据: List[ASCII"MDLN", ASCII"12345"]
    body = bytes([0x01, 0x02])  # List, 2字节长度
    body += struct.pack('>H', 2 + 2 + 5 + 2 + 5)  # List总长度
    body += bytes([0x31, 0x04]) + b'MDLN'  # ASCII "MDLN"
    body += bytes([0x31, 0x05]) + b'12345'  # ASCII "12345"
    header2 = struct.pack('>HBBHI', 0x0001, 0x0C, 0x0B, 0, len(body))
    msg2 = parse_secs_message(header2 + body)
    print(f"解析结果: {msg2}")
    print(f"数据项: {msg2.data_items}")
    assert msg2.stream == 6 and msg2.function == 11
    assert msg2.data_items[0][0] == 0x00  # List
    assert msg2.data_items[0][1][0][1] == 'MDLN'
    assert msg2.data_items[0][1][1][1] == '12345'
    print("全部断言通过，解析正确。")

运行与验证方法

  1. 将上述代码保存为 secs_parser.py。
  2. 在终端执行 python secs_parser.py。
  3. 观察输出：应打印两条解析结果，第一条为无数据项的S1F1消息，第二条为带List数据项的S6F11消息。
  4. 程序末尾的assert全部通过时，输出“全部断言通过，解析正确。”，退出码为0；否则抛出AssertionError。

通过判定标准：程序无异常退出，且控制台输出两条解析结果与最终断言通过信息，即表示解析器正确完成了消息头解析、数据项格式识别与值解码。

常见问题

  • 长度字节数（Length Bytes）：格式字节的低4位表示后续长度字段占用的字节数（0/1/2/4），不是数据长度本身。解析时需先读取长度字段，再按该长度截取数据内容。
  • List嵌套：List格式（0x00）的数据内容为子数据项序列，需递归解析直到消费完List声明的总长度。
  • 字节序：SECS-II标准使用大端字节序（Big-Endian），所有多字节整数和浮点数均按大端解析。
← 上一章 📖 返回目录 下一章 →