239 lines
7.4 KiB
Python
239 lines
7.4 KiB
Python
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) |