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}")