434 lines
16 KiB
Python
434 lines
16 KiB
Python
|
import struct
|
||
|
|
from enum import Enum
|
||
|
|
from typing import Dict, Any
|
||
|
|
|
||
|
|
class VDESMessageType(Enum):
|
||
|
|
"""VDES报文类型枚举"""
|
||
|
|
DEBUG_COMMAND = 0
|
||
|
|
LVDS_DATA = 1
|
||
|
|
CAN_DATA = 2
|
||
|
|
AD9361_CONTROL = 3
|
||
|
|
WAVE_IP_CONTROL = 4
|
||
|
|
TEST_PARAM_CONTROL = 5
|
||
|
|
TEST_WAVE_UPLOAD = 6
|
||
|
|
GPS_DATA = 7
|
||
|
|
STATUS_MONITOR = 8
|
||
|
|
VDES_PARAM_CONTROL = 9
|
||
|
|
VDES_UPLOAD_DATA = 10
|
||
|
|
VDES_COLLECT_DATA = 11
|
||
|
|
VDES_DEMOD_DATA = 12
|
||
|
|
NOTIFICATION = 13
|
||
|
|
|
||
|
|
class VDESUploadDataType(Enum):
|
||
|
|
"""VDES上注数据类型枚举"""
|
||
|
|
LONG_DATA_PARAM = 0
|
||
|
|
LONG_DATA_CONTENT = 1
|
||
|
|
LONG_DATA_END = 2
|
||
|
|
SHORT_MESSAGE = 3
|
||
|
|
EMERGENCY_MESSAGE = 4
|
||
|
|
|
||
|
|
class VDESTCPParser:
|
||
|
|
"""VDES TCP协议解析器"""
|
||
|
|
|
||
|
|
HEADER_MAGIC = b'####'
|
||
|
|
MAX_PACKET_SIZE = 32768
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self._init_parsers()
|
||
|
|
|
||
|
|
def _init_parsers(self):
|
||
|
|
"""初始化各类型报文的解析方法"""
|
||
|
|
self._parsers = {
|
||
|
|
VDESMessageType.DEBUG_COMMAND: self._parse_debug_command,
|
||
|
|
VDESMessageType.LVDS_DATA: self._parse_lvds_data,
|
||
|
|
VDESMessageType.CAN_DATA: self._parse_can_data,
|
||
|
|
VDESMessageType.AD9361_CONTROL: self._parse_ad9361_control,
|
||
|
|
VDESMessageType.WAVE_IP_CONTROL: self._parse_wave_ip_control,
|
||
|
|
VDESMessageType.TEST_PARAM_CONTROL: self._parse_test_param_control,
|
||
|
|
VDESMessageType.TEST_WAVE_UPLOAD: self._parse_test_wave_upload,
|
||
|
|
VDESMessageType.GPS_DATA: self._parse_gps_data,
|
||
|
|
VDESMessageType.STATUS_MONITOR: self._parse_status_monitor,
|
||
|
|
VDESMessageType.VDES_PARAM_CONTROL: self._parse_vdes_param_control,
|
||
|
|
VDESMessageType.VDES_UPLOAD_DATA: self._parse_vdes_upload_data,
|
||
|
|
VDESMessageType.VDES_COLLECT_DATA: self._parse_vdes_collect_data,
|
||
|
|
VDESMessageType.VDES_DEMOD_DATA: self._parse_vdes_demod_data,
|
||
|
|
VDESMessageType.NOTIFICATION: self._parse_notification,
|
||
|
|
}
|
||
|
|
|
||
|
|
def parse_packet(self, data: bytes) -> Dict[str, Any]:
|
||
|
|
"""
|
||
|
|
解析TCP报文
|
||
|
|
:param data: 原始报文数据
|
||
|
|
:return: 解析后的字典结果
|
||
|
|
"""
|
||
|
|
if len(data) < 12:
|
||
|
|
raise ValueError(f"Packet too short ({len(data)} bytes), minimum 12 bytes required")
|
||
|
|
|
||
|
|
# 验证报文头
|
||
|
|
header = data[:4]
|
||
|
|
if header != self.HEADER_MAGIC:
|
||
|
|
raise ValueError(f"Invalid packet header: {header.hex()}")
|
||
|
|
|
||
|
|
# 解析报文类型和长度
|
||
|
|
msg_type, content_len = struct.unpack('>II', data[4:12])
|
||
|
|
content = data[12:12+content_len]
|
||
|
|
|
||
|
|
# 验证内容长度
|
||
|
|
if len(content) != content_len:
|
||
|
|
raise ValueError(f"Content length mismatch: expected {content_len}, got {len(content)}")
|
||
|
|
|
||
|
|
# 获取报文类型枚举
|
||
|
|
try:
|
||
|
|
msg_type_enum = VDESMessageType(msg_type)
|
||
|
|
except ValueError:
|
||
|
|
return {
|
||
|
|
'type': msg_type,
|
||
|
|
'type_name': f'UNKNOWN_{msg_type}',
|
||
|
|
'content': content,
|
||
|
|
'raw': data
|
||
|
|
}
|
||
|
|
|
||
|
|
# 调用对应的解析方法
|
||
|
|
parser = self._parsers.get(msg_type_enum)
|
||
|
|
if parser:
|
||
|
|
return {
|
||
|
|
'type': msg_type_enum,
|
||
|
|
'type_name': msg_type_enum.name,
|
||
|
|
'content': parser(content),
|
||
|
|
'raw': data
|
||
|
|
}
|
||
|
|
else:
|
||
|
|
return {
|
||
|
|
'type': msg_type_enum,
|
||
|
|
'type_name': msg_type_enum.name,
|
||
|
|
'content': content,
|
||
|
|
'raw': data
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_debug_command(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析调试指令报文(类型0)"""
|
||
|
|
if len(content) < 1:
|
||
|
|
raise ValueError("Debug command too short")
|
||
|
|
|
||
|
|
cmd_type = content[0]
|
||
|
|
descriptions = {
|
||
|
|
0: "初始化地检",
|
||
|
|
1: "测试波形上注完成",
|
||
|
|
2: "LVDS数据开始上注",
|
||
|
|
3: "LVDS初始化",
|
||
|
|
4: "AD9361重新校准",
|
||
|
|
5: "GPS初始化"
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
'command_type': cmd_type,
|
||
|
|
'description': descriptions.get(cmd_type, f"未知指令类型: {cmd_type}")
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_lvds_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析LVDS数据报文(类型1)"""
|
||
|
|
return {
|
||
|
|
'data': content,
|
||
|
|
'length': len(content)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_can_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析CAN数据报文(类型2)"""
|
||
|
|
return {
|
||
|
|
'data': content,
|
||
|
|
'length': len(content)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_ad9361_control(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析AD9361参数控制报文(类型3)"""
|
||
|
|
if len(content) != 13:
|
||
|
|
raise ValueError("AD9361 control message must be 13 bytes")
|
||
|
|
|
||
|
|
tx_lo, tx_rate, tx_atten = struct.unpack('>QIB', content)
|
||
|
|
return {
|
||
|
|
'tx_lo_freq': tx_lo,
|
||
|
|
'tx_sample_rate': tx_rate,
|
||
|
|
'tx_attenuation': tx_atten
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_wave_ip_control(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析波形发射IP核控制报文(类型4)"""
|
||
|
|
if len(content) < 20:
|
||
|
|
raise ValueError("Wave IP control message too short")
|
||
|
|
|
||
|
|
# 解析固定部分
|
||
|
|
channel, cic_enable, sample_rate, wave_type = struct.unpack('>BBIB', content[:7])
|
||
|
|
freq_offset, sweep_type, sweep_bw, sweep_speed = struct.unpack('>iBII', content[7:20])
|
||
|
|
|
||
|
|
return {
|
||
|
|
'channel': channel,
|
||
|
|
'cic_enable': bool(cic_enable),
|
||
|
|
'sample_rate': sample_rate,
|
||
|
|
'wave_type': wave_type,
|
||
|
|
'freq_offset': freq_offset,
|
||
|
|
'sweep_type': sweep_type,
|
||
|
|
'sweep_bandwidth': sweep_bw,
|
||
|
|
'sweep_speed': sweep_speed
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_test_param_control(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析发送测试参数控制报文(类型5)"""
|
||
|
|
if len(content) != 9:
|
||
|
|
raise ValueError("Test param control message must be 9 bytes")
|
||
|
|
|
||
|
|
test_type, test_count, test_interval = struct.unpack('>BII', content)
|
||
|
|
return {
|
||
|
|
'test_type': test_type,
|
||
|
|
'test_count': test_count,
|
||
|
|
'test_interval_ms': test_interval
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_test_wave_upload(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析发送测试波形上注报文(类型6)"""
|
||
|
|
return {
|
||
|
|
'wave_data': content,
|
||
|
|
'length': len(content)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_gps_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析GPS解析数据报文(类型7)"""
|
||
|
|
if len(content) < 36:
|
||
|
|
raise ValueError("GPS data message too short")
|
||
|
|
|
||
|
|
fields = struct.unpack('>QiiihhhBBIBII', content[:36])
|
||
|
|
|
||
|
|
return {
|
||
|
|
'utc_time': fields[0],
|
||
|
|
'longitude': fields[1],
|
||
|
|
'latitude': fields[2],
|
||
|
|
'altitude': fields[3],
|
||
|
|
'used_satellites': fields[4],
|
||
|
|
'gps_satellites': fields[5],
|
||
|
|
'beidou_satellites': fields[6],
|
||
|
|
'gps_status': fields[7],
|
||
|
|
'position_mode': fields[8],
|
||
|
|
'hdop': fields[9],
|
||
|
|
'speed': fields[10],
|
||
|
|
'direction': fields[11]
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_status_monitor(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析地检状态监控报文(类型8)"""
|
||
|
|
if len(content) < 53:
|
||
|
|
raise ValueError("Status monitor message too short")
|
||
|
|
|
||
|
|
# 解析前38字节
|
||
|
|
fields = struct.unpack('>BBIIBBBBIIIBIII', content[:38])
|
||
|
|
|
||
|
|
result = {
|
||
|
|
'cpu0_usage': fields[0],
|
||
|
|
'cpu1_usage': fields[1],
|
||
|
|
'version': fields[2],
|
||
|
|
'uptime': fields[3],
|
||
|
|
'gps_status': fields[4],
|
||
|
|
'pps_lock_status': fields[5],
|
||
|
|
'ad9361_status': fields[6],
|
||
|
|
'ad9361_calibration_factor': fields[7],
|
||
|
|
'ad9361_tx_lo': fields[8],
|
||
|
|
'ad9361_tx_rate': fields[9],
|
||
|
|
'ad9361_tx_atten': fields[10],
|
||
|
|
'ad9361_rx_lo': fields[11],
|
||
|
|
'ad9361_rx_rate': fields[12],
|
||
|
|
'ad9361_rx_gain': fields[13],
|
||
|
|
'vdes_enable': fields[14],
|
||
|
|
}
|
||
|
|
|
||
|
|
# 解析剩余部分
|
||
|
|
pos = 38
|
||
|
|
more_fields = struct.unpack('>BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB', content[pos:pos+15])
|
||
|
|
|
||
|
|
result.update({
|
||
|
|
'vdes_filter_enable': more_fields[0],
|
||
|
|
'rac_limit_enable': more_fields[1],
|
||
|
|
'vdes_agc_enable': more_fields[2],
|
||
|
|
'sat_delay_comp_enable': more_fields[3],
|
||
|
|
'upper_sat_id': more_fields[4],
|
||
|
|
'upper_sat_cqi': more_fields[5],
|
||
|
|
'upper_sat_amp': more_fields[6],
|
||
|
|
'lower_sat_id': more_fields[7],
|
||
|
|
'lower_sat_cqi': more_fields[8],
|
||
|
|
'lower_sat_power': more_fields[9],
|
||
|
|
'current_sat_id': more_fields[10],
|
||
|
|
'current_slot': more_fields[11],
|
||
|
|
'current_subframe': more_fields[12],
|
||
|
|
'vdes_tx_link_id': more_fields[13],
|
||
|
|
'vdes_tx_channel': more_fields[14],
|
||
|
|
'vdes_tx_freq': more_fields[15],
|
||
|
|
'vdes_buffer1_status': more_fields[16],
|
||
|
|
'vdes_buffer2_status': more_fields[17],
|
||
|
|
'vdes_demod1_enable': more_fields[18],
|
||
|
|
'vdes_demod1_link_id': more_fields[19],
|
||
|
|
'vdes_demod1_channel': more_fields[20],
|
||
|
|
'vdes_demod1_freq': more_fields[21],
|
||
|
|
'vdes_demod2_enable': more_fields[22],
|
||
|
|
'vdes_demod2_link_id': more_fields[23],
|
||
|
|
'vdes_demod2_channel': more_fields[24],
|
||
|
|
'vdes_demod2_freq': more_fields[25],
|
||
|
|
'vdes_demod3_enable': more_fields[26],
|
||
|
|
'vdes_demod3_link_id': more_fields[27],
|
||
|
|
'vdes_demod3_channel': more_fields[28],
|
||
|
|
'vdes_demod3_freq': more_fields[29],
|
||
|
|
'vdes_demod4_enable': more_fields[30],
|
||
|
|
'vdes_demod4_link_id': more_fields[31],
|
||
|
|
'vdes_demod4_channel': more_fields[32],
|
||
|
|
'vdes_demod4_freq': more_fields[33],
|
||
|
|
'ais_demod1_enable': more_fields[34],
|
||
|
|
'ais_demod1_freq': more_fields[35],
|
||
|
|
'ais_demod2_enable': more_fields[36],
|
||
|
|
'ais_demod2_freq': more_fields[37],
|
||
|
|
})
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def _parse_vdes_param_control(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析VDES协议参数控制报文(类型9)"""
|
||
|
|
if len(content) < 18:
|
||
|
|
raise ValueError("VDES param control message too short")
|
||
|
|
|
||
|
|
fields = struct.unpack('>BBBBIBBBiBIBBBiB', content[:18])
|
||
|
|
|
||
|
|
return {
|
||
|
|
'vdes_enable': fields[0],
|
||
|
|
'filter_enable': fields[1],
|
||
|
|
'rac_limit_enable': fields[2],
|
||
|
|
'agc_enable': fields[3],
|
||
|
|
'sat_delay_comp_enable': fields[4],
|
||
|
|
'mmsi': fields[5],
|
||
|
|
'default_tx_link_id': fields[6],
|
||
|
|
'default_tx_channel': fields[7],
|
||
|
|
'default_tx_freq': fields[8],
|
||
|
|
'tx_delay': fields[9],
|
||
|
|
'tx_power_atten': fields[10],
|
||
|
|
'demod1_link_id': fields[11],
|
||
|
|
'demod1_channel': fields[12],
|
||
|
|
'demod1_freq': fields[13],
|
||
|
|
'rx_delay': fields[14],
|
||
|
|
'rx_gain': fields[15],
|
||
|
|
'clear_buffer1': fields[16],
|
||
|
|
'clear_buffer2': fields[17],
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_vdes_upload_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析VDES上注数据报文(类型10)"""
|
||
|
|
if len(content) < 1:
|
||
|
|
raise ValueError("VDES upload data message too short")
|
||
|
|
|
||
|
|
data_type = content[0]
|
||
|
|
try:
|
||
|
|
data_type_enum = VDESUploadDataType(data_type)
|
||
|
|
except ValueError:
|
||
|
|
data_type_enum = None
|
||
|
|
|
||
|
|
result = {'data_type': data_type}
|
||
|
|
|
||
|
|
if data_type_enum == VDESUploadDataType.LONG_DATA_PARAM:
|
||
|
|
if len(content) < 6:
|
||
|
|
raise ValueError("Long data param message too short")
|
||
|
|
target_mmsi, priority = struct.unpack('>IB', content[1:6])
|
||
|
|
result.update({
|
||
|
|
'target_mmsi': target_mmsi,
|
||
|
|
'priority': priority
|
||
|
|
})
|
||
|
|
elif data_type_enum == VDESUploadDataType.LONG_DATA_CONTENT:
|
||
|
|
result['data'] = content[1:]
|
||
|
|
elif data_type_enum == VDESUploadDataType.SHORT_MESSAGE:
|
||
|
|
if len(content) < 6:
|
||
|
|
raise ValueError("Short message too short")
|
||
|
|
msg_type, target_mmsi = struct.unpack('>BI', content[1:6])
|
||
|
|
msg_data = content[6:]
|
||
|
|
result.update({
|
||
|
|
'message_type': msg_type,
|
||
|
|
'target_mmsi': target_mmsi,
|
||
|
|
'message_data': msg_data
|
||
|
|
})
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def _parse_vdes_collect_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析VDES采集波形报文(类型11)"""
|
||
|
|
if len(content) != 3584:
|
||
|
|
raise ValueError("VDES collect data must be 3584 bytes")
|
||
|
|
return {
|
||
|
|
'wave_data': content,
|
||
|
|
'length': len(content)
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_vdes_demod_data(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析VDES/AIS解调数据报文(类型12)"""
|
||
|
|
if len(content) < 8:
|
||
|
|
raise ValueError("VDES demod data message too short")
|
||
|
|
|
||
|
|
msg_type, sat_id, demod_id, slot, delay, snr = struct.unpack('>BBBHHh', content[:8])
|
||
|
|
demod_data = content[8:]
|
||
|
|
|
||
|
|
return {
|
||
|
|
'message_type': msg_type,
|
||
|
|
'satellite_id': sat_id,
|
||
|
|
'demodulator_id': demod_id,
|
||
|
|
'slot': slot,
|
||
|
|
'delay_us': delay,
|
||
|
|
'snr_db': snr / 256, # 转换为浮点数
|
||
|
|
'demod_data': demod_data
|
||
|
|
}
|
||
|
|
|
||
|
|
def _parse_notification(self, content: bytes) -> Dict[str, Any]:
|
||
|
|
"""解析消息通知报文(类型13)"""
|
||
|
|
if len(content) < 1:
|
||
|
|
raise ValueError("Notification message too short")
|
||
|
|
|
||
|
|
notif_type = content[0]
|
||
|
|
descriptions = {
|
||
|
|
0: "VDES协议开始执行",
|
||
|
|
1: "VDES协议停止执行",
|
||
|
|
2: "VDES功能初始化",
|
||
|
|
3: "VDES开始采集波形",
|
||
|
|
4: "VDES正在采集波形",
|
||
|
|
5: "VDES停止采集波形",
|
||
|
|
6: "VDES收到上行数据",
|
||
|
|
7: "VDES与卫星建立会话成功",
|
||
|
|
8: "VDES与卫星结束会话",
|
||
|
|
9: "VDES发送上行测试波形",
|
||
|
|
10: "VDES上行测试完毕",
|
||
|
|
11: "VDES发送上行数据",
|
||
|
|
12: "VDES解调数据"
|
||
|
|
}
|
||
|
|
|
||
|
|
return {
|
||
|
|
'notification_type': notif_type,
|
||
|
|
'description': descriptions.get(notif_type, f"未知通知类型: {notif_type}"),
|
||
|
|
'additional_data': content[1:] if len(content) > 1 else None
|
||
|
|
}
|
||
|
|
|
||
|
|
# 使用示例
|
||
|
|
if __name__ == "__main__":
|
||
|
|
parser = VDESTCPParser()
|
||
|
|
|
||
|
|
# 示例1: 解析调试指令报文
|
||
|
|
debug_cmd = b'####\x00\x00\x00\x00\x00\x00\x00\x01\x02'
|
||
|
|
try:
|
||
|
|
result = parser.parse_packet(debug_cmd)
|
||
|
|
print("调试指令报文解析结果:")
|
||
|
|
print(f"类型: {result['type_name']}")
|
||
|
|
print(f"内容: {result['content']}")
|
||
|
|
except ValueError as e:
|
||
|
|
print(f"解析错误: {e}")
|
||
|
|
|
||
|
|
# 示例2: 解析GPS数据报文
|
||
|
|
gps_data = b'####\x00\x00\x00\x07\x00\x00\x00\x24' + \
|
||
|
|
b'\x00\x00\x00\x00\x00\x00\x00\x01' + \
|
||
|
|
b'\x00\x00\x00\x02\x00\x00\x00\x03' + \
|
||
|
|
b'\x00\x00\x00\x04\x00\x05\x00\x06' + \
|
||
|
|
b'\x00\x07\x01\x02\x00\x00\x00\x08' + \
|
||
|
|
b'\x00\x00\x00\x09\x00\x00\x00\x0A'
|
||
|
|
try:
|
||
|
|
result = parser.parse_packet(gps_data)
|
||
|
|
print("\nGPS数据报文解析结果:")
|
||
|
|
print(f"类型: {result['type_name']}")
|
||
|
|
print(f"内容: {result['content']}")
|
||
|
|
except ValueError as e:
|
||
|
|
print(f"解析错误: {e}")
|