如何在Python中解析UDP数据报中长度未知的C结构体

核心问题分析 #

你遇到的报错根源是格式字符串与实际数据结构不匹配：

• 错误地将C#的ulong（8字节）用Python struct的L（4字节无符号长整型）解析
• 格式字符串中多余了2个float和1个bool，导致计算出的所需字节数（52）与实际数据长度不符
• 未处理可变长度的CurrentSampleStruct循环解析逻辑

正确解析步骤 #

1. 先解析固定长度的头部结构体，获取NumberOfValues参数
2. 根据NumberOfValues循环解析每个样本结构体
3. 最后解析尾部的终止符

修正后的代码 #

import socket
import datetime
import struct

# Configs
UDP_IP = "<IP OF SOURCE>"
UDP_PORT = <UDP_PORT>  # datagram port

# 创建UDP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((UDP_IP, UDP_PORT))

# 启动信息
timestamp = datetime.datetime.now().time()
print(f"启动时间: {timestamp}")
print("-------------------------------------------")

# 测试读取5个数据包
for i in range(5):
    data, senderaddr = sock.recvfrom(10240000)
    print(f"\n第{i+1}个数据包长度: {len(data)}字节")

    # 1. 解析头部结构体
    # 格式字符串说明: <表示小端字节序，对应C# struct的默认字节序
    header_format = '<hhQffh??fff'
    header_size = struct.calcsize(header_format)
    header = struct.unpack(header_format, data[:header_size])
    
    # 提取头部字段
    (msg_id, msg_version, ping_time, start_range, sample_interval, 
     num_values, is_vessel_relative, is_valid_bottom_data,
     bottom_vel_north, bottom_vel_east, vessel_heading) = header
    
    # 转换PingTime为可读时间（Kongsberg通常用毫秒级时间戳）
    ping_datetime = datetime.datetime.fromtimestamp(ping_time / 1000.0)

    # 2. 解析可变长度的样本数据
    sample_format = '<ff?'
    sample_size = struct.calcsize(sample_format)
    samples = []
    current_pos = header_size
    
    for idx in range(num_values):
        sample_data = data[current_pos:current_pos+sample_size]
        vel_north, vel_east, valid_data = struct.unpack(sample_format, sample_data)
        samples.append({
            "序号": idx+1,
            "北向速度": vel_north,
            "东向速度": vel_east,
            "数据有效": valid_data
        })
        current_pos += sample_size

    # 3. 解析尾部终止符
    terminator = struct.unpack('<h', data[current_pos:current_pos+2])[0]

    # 打印解析结果
    print("=== 头部信息 ===")
    print(f"消息ID: {msg_id}")
    print(f"消息版本: {msg_version}")
    print(f"Ping时间: {ping_datetime} (原始戳: {ping_time})")
    print(f"起始距离: {start_range}")
    print(f"采样间隔: {sample_interval}")
    print(f"样本数量: {num_values}")
    print(f"相对船坐标系: {is_vessel_relative}")
    print(f"底部数据有效: {is_valid_bottom_data}")
    print(f"底部北向速度: {bottom_vel_north}")
    print(f"底部东向速度: {bottom_vel_east}")
    print(f"船首向: {vessel_heading}")

    print("\n=== 样本数据 ===")
    for sample in samples:
        print(f"样本{sample['序号']}: 北向={sample['北向速度']}, 东向={sample['东向速度']}, 有效={sample['数据有效']}")

    print(f"\n=== 终止符 ===")
    print(f"终止符值: {terminator}")

关键细节说明 #

• 字节序：使用<指定小端字节序，匹配C# struct的默认内存布局（若解析结果异常，可尝试换成大端>）
• 类型对应：
  • C# ulong → Python struct Q（8字节无符号长整型）
  • C# short → h（2字节短整型）
  • C# float → f（4字节单精度浮点）
  • C# bool → ?（1字节布尔值）
• 内存对齐：若解析样本时出现数据偏移，可能是C# struct启用了4字节对齐，此时需将样本格式字符串改为<ffxx?（xx表示2字节填充），对应样本长度变为12字节，需重新计算sample_size

内容的提问来源于stack exchange，提问作者Julian Race