import struct import math def create_ais_message(): """创建AIS消息1的二进制数据""" # 初始化二进制位列表 bits = [] # 添加各个字段（按照协议顺序） # 消息ID (6 bits) message_id = 1 # AIS消息1 bits.append(format(message_id, '06b')) # 转发指示符 (2 bits) repeat_indicator = 0 # 默认 bits.append(format(repeat_indicator, '02b')) # 用户ID/MMSI (30 bits) user_id = 123456789 # 示例MMSI bits.append(format(user_id, '030b')) # 导航状态 (4 bits) navigation_status = 8 # 航行中 bits.append(format(navigation_status, '04b')) # 旋转速率 (8 bits) # 注意：旋转速率有特殊编码规则 rot = -128 # 无旋转信息（默认） bits.append(format(rot & 0xFF, '08b')) # 使用8位补码表示 # 地面航速SOG (10 bits) sog = 102 # 10.2节 (102 * 0.1) bits.append(format(min(sog, 1023), '010b')) # 最大1023 # 位置准确度 (1 bit) position_accuracy = 1 # 高精度(>10m) bits.append(str(position_accuracy)) # 经度 (28 bits) # 以1/10,000分钟为单位，东经为正 longitude = 120.5 # 东经120.5度 longitude_min = int(longitude * 60 * 10000) bits.append(format(longitude_min & 0x0FFFFFFF, '028b')) # 28位 # 纬度 (27 bits) # 以1/10,000分钟为单位，北纬为正 latitude = 30.5 # 北纬30.5度 latitude_min = int(latitude * 60 * 10000) bits.append(format(latitude_min & 0x07FFFFFF, '027b')) # 27位 # 地面航线COG (12 bits) cog = 900 # 90.0度 (900 * 0.1) bits.append(format(min(cog, 3600), '012b')) # 最大3600 # 实际航向 (9 bits) true_heading = 90 # 90度 bits.append(format(min(true_heading, 511), '09b')) # 最大511 # 时戳 (6 bits) timestamp = 30 # UTC秒 bits.append(format(min(timestamp, 63), '06b')) # 特定操纵指示符 (2 bits) special_maneuver = 0 # 不可用 bits.append(format(special_maneuver, '02b')) # 备用 (3 bits) spare = 0 bits.append(format(spare, '03b')) # RAIM标志 (1 bit) raim_flag = 0 # 未使用 bits.append(str(raim_flag)) # 通信状态 (19 bits) comm_state = 0 # 默认 bits.append(format(comm_state, '019b')) # 合并所有位 bit_string = ''.join(bits) # 将位字符串转换为字节 byte_count = (len(bit_string)) // 8 ais_data = bytearray() for i in range(byte_count): byte_str = bit_string[i*8:(i+1)*8] ais_data.append(int(byte_str, 2)) return ais_data def create_ais_demod_packet(): """创建AIS解调数据报文（类型12）的内容""" # 消息类型 (1字节) msg_type = 12 # 卫星ID (1字节) satellite_id = 1 # 解调器编号 (1字节) - 4表示AIS解调器1 demodulator_id = 4 # 解调时隙 (2字节) slot = 1000 # 接收延迟 (2字节) 单位微秒 delay = 500 # 500微秒 # 信噪比 (2字节) # 高8位整数部分，低8位小数部分 snr = 25.5 # dB integer_part = int(snr) fractional_part = int((snr - integer_part) * 256) snr_value = (integer_part << 8) | fractional_part # 解调数据 (AIS消息) demod_data = create_ais_message() # 打包为二进制 (大端字节序) content = struct.pack( '>BBBHh', # B:uint8, H:uint16, h:int16 msg_type, satellite_id, demodulator_id, slot, snr_value ) # 添加解调数据 content += demod_data return content def create_tcp_packet(packet_type, content): """创建符合VDES TCP协议的报文""" # 报文帧头 (4字节) header = b'####' # 报文类型 (4字节，大端) type_field = struct.pack('>I', packet_type) # 报文长度 (4字节，大端，内容长度) length = len(content) length_field = struct.pack('>I', length) # 构建完整报文 packet = header + type_field + length_field + content return packet def parse_ais_message(ais_data): """解析AIS消息1的二进制数据""" # 将字节转换为二进制字符串 bit_string = ''.join(f'{b:08b}' for b in ais_data) # 解析各个字段 pos = 0 message_id = int(bit_string[pos:pos+6], 2); pos += 6 repeat_indicator = int(bit_string[pos:pos+2], 2); pos += 2 user_id = int(bit_string[pos:pos+30], 2); pos += 30 nav_status = int(bit_string[pos:pos+4], 2); pos += 4 rot = int(bit_string[pos:pos+8], 2) rot = rot if rot < 128 else rot - 256 # 转换为有符号数; pos += 8 sog = int(bit_string[pos:pos+10], 2) / 10.0; pos += 10 pos_acc = int(bit_string[pos:pos+1], 2); pos += 1 longitude = int(bit_string[pos:pos+28], 2) longitude = longitude if longitude < 0x8000000 else longitude - 0x10000000 # 转换为有符号数 longitude = longitude / 10000.0 / 60.0; pos += 28 latitude = int(bit_string[pos:pos+27], 2) latitude = latitude if latitude < 0x4000000 else latitude - 0x8000000 # 转换为有符号数 latitude = latitude / 10000.0 / 60.0; pos += 27 cog = int(bit_string[pos:pos+12], 2) / 10.0; pos += 12 true_heading = int(bit_string[pos:pos+9], 2); pos += 9 timestamp = int(bit_string[pos:pos+6], 2); pos += 6 special_maneuver = int(bit_string[pos:pos+2], 2); pos += 2 spare = int(bit_string[pos:pos+3], 2); pos += 3 raim_flag = int(bit_string[pos:pos+1], 2); pos += 1 comm_state = int(bit_string[pos:pos+19], 2); pos += 19 return { 'message_id': message_id, 'repeat_indicator': repeat_indicator, 'user_id': user_id, 'nav_status': nav_status, 'rot': rot, 'sog': sog, 'pos_acc': pos_acc, 'longitude': longitude, 'latitude': latitude, 'cog': cog, 'true_heading': true_heading, 'timestamp': timestamp, 'special_maneuver': special_maneuver, 'spare': spare, 'raim_flag': raim_flag, 'comm_state': comm_state } # 生成AIS解调报文 ais_demod_content = create_ais_demod_packet() print(ais_demod_content) ais_demod_packet = create_tcp_packet(12, ais_demod_content) # 打印二进制报文 print("\n\nAIS解调报文字节长度:", len(ais_demod_packet)) print("二进制表示:") print(' '.join(f'{b:02x}' for b in ais_demod_packet)) # 打印字段解释 print("\n报文结构解析:") print(f"帧头: {ais_demod_packet[:4].decode('ascii')}") print(f"报文类型: {struct.unpack('>I', ais_demod_packet[4:8])[0]} (12=VDES/AIS解调数据)") print(f"内容长度: {struct.unpack('>I', ais_demod_packet[8:12])[0]} 字节") # 解析内容部分 header_part = ais_demod_packet[12:19] # 前8字节是固定头部 msg_type, sat_id, demod_id, slot, snr_value = struct.unpack('>BBBHh', header_part) # 解析信噪比 snr_int = snr_value >> 8 snr_frac = snr_value & 0xFF snr = snr_int + snr_frac / 256.0 # AIS数据部分 ais_data = ais_demod_packet[20:] print("\nAIS解调内容解析:") print(f"消息类型: {msg_type}") print(f"卫星ID: {sat_id}") print(f"解调器编号: {demod_id} (4=AIS解调器1)") print(f"解调时隙: {slot}") # print(f"接收延迟: {delay} μs") print(f"信噪比: {snr:.1f} dB") print(f"AIS数据长度: {len(ais_data)} 字节") # 打印AIS数据的十六进制表示 print("AIS消息十六进制:", ' '.join(f'{b:02x}' for b in ais_data)) ais_info = parse_ais_message(ais_data) print(ais_info)