299 lines
11 KiB
Python
299 lines
11 KiB
Python
from dataclasses import dataclass
|
||
from typing import Optional
|
||
import struct
|
||
|
||
@dataclass
|
||
class Frame:
|
||
"""帧数据结构"""
|
||
command: int # 命令码 (4字节)
|
||
length: int # 数据长度 (4字节)
|
||
frame_idx: int # 帧序号 (4字节)
|
||
channels: list # 17个通道数据 (每个4字节)
|
||
checksum: int # 校验和 (1字节)
|
||
|
||
class FrameFormat:
|
||
"""帧格式描述类"""
|
||
# 帧头魔数 (4字节)
|
||
FRAME_HEADER = bytes([0xAA, 0xAA, 0xAA, 0xAA])
|
||
|
||
# 各字段长度定义
|
||
HEADER_SIZE = 4
|
||
COMMAND_SIZE = 4
|
||
LENGTH_SIZE = 4
|
||
INDEX_SIZE = 4
|
||
CHECKSUM_SIZE = 1
|
||
CHANNEL_SIZE = 4 # 每个通道的字节数
|
||
CHANNEL_COUNT = 17 # 通道数量
|
||
|
||
# 帧固定部分的长度（不包含可变长度的数据部分）
|
||
FIXED_SIZE = HEADER_SIZE + COMMAND_SIZE + LENGTH_SIZE + INDEX_SIZE + CHECKSUM_SIZE
|
||
|
||
# 数据部分的长度（帧IDX + 通道数据）
|
||
DATA_SIZE = INDEX_SIZE + (CHANNEL_COUNT - 1) * CHANNEL_SIZE # 帧IDX + (通道数-1)个通道数据
|
||
|
||
@classmethod
|
||
def generateFrameData(cls, frame_idx: int, command: int = None, channel_values: list = None) -> bytes:
|
||
"""
|
||
生成测试用的帧数据
|
||
Args:
|
||
frame_idx: 帧序号
|
||
command: 命令码，如果为None则使用frame_idx作为命令码
|
||
channel_values: 通道数据列表，如果为None则生成随机数据
|
||
Returns:
|
||
bytes: 完整的帧数据
|
||
"""
|
||
# 使用frame_idx作为默认命令码
|
||
if command is None:
|
||
command = frame_idx
|
||
|
||
# 生成随机通道数据
|
||
if channel_values is None:
|
||
import random
|
||
channel_values = []
|
||
for _ in range(cls.CHANNEL_COUNT): # 使用类常量确保17个通道
|
||
# 生成-1000到1000之间的随机浮点数，转换为整数存储
|
||
value = int(random.uniform(-1000, 1000) * 1000) # 放大1000倍以保持精度
|
||
channel_values.append(value)
|
||
elif len(channel_values) != cls.CHANNEL_COUNT:
|
||
raise ValueError(f"通道数据数量必须为 {cls.CHANNEL_COUNT}，当前为 {len(channel_values)}")
|
||
|
||
# 构建数据部分（帧IDX + 通道数据）
|
||
data = struct.pack("<I", frame_idx) # 帧IDX
|
||
for value in channel_values:
|
||
data += struct.pack("<i", value) # 通道数据
|
||
|
||
# 使用build_frame方法构建完整帧
|
||
frame_data = cls.build_frame(command, frame_idx, data)
|
||
|
||
# 验证生成的帧是否有效
|
||
if not cls.parse_frame(frame_data):
|
||
print(f"\n帧验证失败，帧数据:")
|
||
print(f"帧头: {frame_data[:cls.HEADER_SIZE].hex().upper()}")
|
||
print(f"命令码: {struct.unpack('<I', frame_data[cls.HEADER_SIZE:cls.HEADER_SIZE+cls.COMMAND_SIZE])[0]}")
|
||
print(f"长度: {struct.unpack('<I', frame_data[cls.HEADER_SIZE+cls.COMMAND_SIZE:cls.HEADER_SIZE+cls.COMMAND_SIZE+cls.LENGTH_SIZE])[0]}")
|
||
print(f"帧序号: {struct.unpack('<I', frame_data[cls.HEADER_SIZE+cls.COMMAND_SIZE+cls.LENGTH_SIZE:cls.HEADER_SIZE+cls.COMMAND_SIZE+cls.LENGTH_SIZE+cls.INDEX_SIZE])[0]}")
|
||
print(f"校验和: {frame_data[-1]:02X}")
|
||
print(f"完整帧数据: {frame_data.hex().upper()}")
|
||
raise ValueError("生成的帧验证失败")
|
||
|
||
return frame_data
|
||
|
||
@classmethod
|
||
def parse_frame(cls, data: bytes) -> Optional[Frame]:
|
||
"""
|
||
解析帧数据
|
||
Args:
|
||
data: 完整的帧数据
|
||
Returns:
|
||
Frame: 解析后的帧对象，如果解析失败返回None
|
||
"""
|
||
try:
|
||
# 检查数据长度
|
||
if len(data) < cls.FIXED_SIZE + cls.DATA_SIZE:
|
||
print(f"数据长度不足: {len(data)} < {cls.FIXED_SIZE + cls.DATA_SIZE}")
|
||
return None
|
||
|
||
# 检查帧头
|
||
if data[:cls.HEADER_SIZE] != cls.FRAME_HEADER:
|
||
print(f"帧头不匹配: {data[:cls.HEADER_SIZE].hex()} != {cls.FRAME_HEADER.hex()}")
|
||
return None
|
||
|
||
# 解析固定字段
|
||
offset = cls.HEADER_SIZE
|
||
command = struct.unpack("<I", data[offset:offset+cls.COMMAND_SIZE])[0]
|
||
offset += cls.COMMAND_SIZE
|
||
|
||
length = struct.unpack("<I", data[offset:offset+cls.LENGTH_SIZE])[0]
|
||
offset += cls.LENGTH_SIZE
|
||
|
||
# 提取帧IDX（第一个数据项）
|
||
frame_idx = struct.unpack("<I", data[offset:offset+4])[0]
|
||
offset += 4
|
||
|
||
# 提取17个通道数据
|
||
channels = []
|
||
try:
|
||
for _ in range(17):
|
||
if offset + 4 > len(data):
|
||
print("错误：数据长度不足以读取所有通道")
|
||
return None
|
||
channel_value = struct.unpack("<I", data[offset:offset+4])[0]
|
||
channels.append(channel_value)
|
||
offset += 4
|
||
except Exception as e:
|
||
print(f"解析通道数据失败: {e}")
|
||
return None
|
||
|
||
# 确保还有校验和字节
|
||
if offset >= len(data):
|
||
print("错误：数据长度不足以读取校验和")
|
||
return None
|
||
|
||
# 提取校验和
|
||
checksum = data[offset]
|
||
|
||
# 验证校验和
|
||
if not cls.verify_checksum(data[:offset], checksum):
|
||
print("校验和验证失败")
|
||
return None
|
||
|
||
return Frame(
|
||
command=command,
|
||
length=length,
|
||
frame_idx=frame_idx,
|
||
channels=channels,
|
||
checksum=checksum
|
||
)
|
||
|
||
except Exception as e:
|
||
print(f"解析帧时出错: {e}")
|
||
return None
|
||
|
||
@classmethod
|
||
def build_frame(cls, command: int, index: int, data: bytes) -> bytes:
|
||
"""
|
||
构建帧数据
|
||
Args:
|
||
command: 命令码
|
||
index: 帧索引
|
||
data: 数据内容
|
||
Returns:
|
||
bytes: 完整的帧数据
|
||
"""
|
||
# 构建帧数据（不包含校验和）
|
||
frame_data = (
|
||
cls.FRAME_HEADER +
|
||
struct.pack("<I", command) +
|
||
struct.pack("<I", len(data) + cls.CHECKSUM_SIZE) + # 修正：长度应该是数据长度加上校验和长度
|
||
data # 修正：直接使用传入的数据，不需要再次打包帧IDX
|
||
)
|
||
|
||
# 计算并添加校验和
|
||
checksum = cls.calculate_checksum(frame_data)
|
||
return frame_data + bytes([checksum])
|
||
|
||
@staticmethod
|
||
def calculate_checksum(data: bytes) -> int:
|
||
"""
|
||
计算校验和 - 不包含校验码本身
|
||
Args:
|
||
data: 需要计算校验和的数据（不包含校验码）
|
||
Returns:
|
||
int: 校验和值
|
||
"""
|
||
# 简单的字节累加校验
|
||
return sum(data) & 0xFF
|
||
|
||
@classmethod
|
||
def verify_checksum(cls, data: bytes, checksum: int) -> bool:
|
||
"""
|
||
验证校验和
|
||
Args:
|
||
data: 数据内容（不包含校验码）
|
||
checksum: 校验和值
|
||
Returns:
|
||
bool: 校验结果
|
||
"""
|
||
return cls.calculate_checksum(data) == checksum
|
||
|
||
@classmethod
|
||
def get_frame_length(cls, data: bytes) -> Optional[int]:
|
||
"""
|
||
从数据中获取帧总长度
|
||
Args:
|
||
data: 至少包含帧头和长度字段的数据
|
||
Returns:
|
||
int: 帧总长度，如果数据不足返回None
|
||
"""
|
||
min_size = cls.HEADER_SIZE + cls.COMMAND_SIZE + cls.LENGTH_SIZE
|
||
if len(data) < min_size:
|
||
return None
|
||
|
||
if data[:cls.HEADER_SIZE] != cls.FRAME_HEADER:
|
||
return None
|
||
|
||
length = struct.unpack("<I", data[cls.HEADER_SIZE+cls.COMMAND_SIZE:min_size])[0]
|
||
return length + cls.FIXED_SIZE
|
||
|
||
if __name__ == "__main__":
|
||
# 测试数据（十六进制字符串）
|
||
test_data_hex = "AAAAAAAA0200000049000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000F3"
|
||
print(len(test_data_hex))
|
||
# 将十六进制字符串转换为字节数据
|
||
test_data = bytes.fromhex(test_data_hex)
|
||
print(f"测试数据长度: {len(test_data)} 字节")
|
||
print(f"测试数据: {test_data.hex(' ').upper()}")
|
||
|
||
# 解析过程调试
|
||
print("\n解析过程:")
|
||
|
||
# 1. 检查数据长度
|
||
print(f"1. 检查数据长度: {len(test_data)} >= {FrameFormat.FIXED_SIZE}")
|
||
|
||
# 2. 检查帧头
|
||
frame_header = test_data[:FrameFormat.HEADER_SIZE]
|
||
print(f"2. 检查帧头: {frame_header.hex().upper()} == {FrameFormat.FRAME_HEADER.hex().upper()}")
|
||
|
||
# 3. 解析命令码
|
||
offset = FrameFormat.HEADER_SIZE
|
||
command_bytes = test_data[offset:offset+FrameFormat.COMMAND_SIZE]
|
||
command = struct.unpack("<I", command_bytes)[0]
|
||
print(f"3. 命令码: 0x{command:08X} (bytes: {command_bytes.hex().upper()})")
|
||
|
||
# 4. 解析长度
|
||
offset += FrameFormat.COMMAND_SIZE
|
||
length_bytes = test_data[offset:offset+FrameFormat.LENGTH_SIZE]
|
||
length = struct.unpack("<I", length_bytes)[0]
|
||
print(f"4. 长度: {length} (bytes: {length_bytes.hex().upper()})")
|
||
|
||
# 5. 提取帧IDX
|
||
offset += FrameFormat.LENGTH_SIZE # 更新offset到帧IDX位置
|
||
frame_idx_bytes = test_data[offset:offset+4]
|
||
frame_idx = struct.unpack("<I", frame_idx_bytes)[0]
|
||
print(f"5. 帧IDX: {frame_idx}")
|
||
offset += 4
|
||
|
||
# 6. 提取17个通道数据
|
||
print("6. 通道数据:")
|
||
channels = []
|
||
try:
|
||
for i in range(17):
|
||
channel_bytes = test_data[offset:offset+4]
|
||
if len(channel_bytes) != 4:
|
||
print(f" 警告：通道 {i+1} 数据不完整，期望4字节，实际{len(channel_bytes)}字节")
|
||
break
|
||
channel_value = struct.unpack("<I", channel_bytes)[0]
|
||
channels.append(channel_value)
|
||
print(f" 通道 {i+1}: 0x{channel_value:08X}")
|
||
offset += 4
|
||
except Exception as e:
|
||
print(f" 错误：解析通道数据失败: {e}")
|
||
print(f" 当前偏移量: {offset}")
|
||
print(f" 剩余数据: {test_data[offset:].hex().upper()}")
|
||
|
||
# 7. 提取校验和
|
||
if offset < len(test_data):
|
||
checksum = test_data[offset]
|
||
print(f"7. 校验和: 0x{checksum:02X}")
|
||
else:
|
||
print("7. 错误：无法读取校验和，数据不足")
|
||
|
||
# 8. 验证校验和
|
||
# 计算校验和时只使用校验码之前的数据
|
||
calculated_checksum = FrameFormat.calculate_checksum(test_data[:offset])
|
||
print(f"8. 校验和验证: 计算值=0x{calculated_checksum:02X}, 实际值=0x{checksum:02X}")
|
||
|
||
# 解析帧
|
||
frame = FrameFormat.parse_frame(test_data)
|
||
if frame:
|
||
print("\n解析成功!")
|
||
print(f"命令码: 0x{frame.command:08X}")
|
||
print(f"长度: {frame.length}")
|
||
print(f"帧IDX: {frame.frame_idx}")
|
||
print("通道数据:")
|
||
for i, value in enumerate(frame.channels):
|
||
print(f"通道 {i+1}: 0x{value:08X}")
|
||
print(f"校验和: 0x{frame.checksum:02X}")
|
||
else:
|
||
print("\n解析失败!")
|