153 lines
5.8 KiB
Python
153 lines
5.8 KiB
Python
|
||
import random
|
||
import struct
|
||
|
||
def generate_ais_message1():
|
||
"""生成AIS Message1位置报告 (M.1371-5 Annex8 3.1)"""
|
||
# 随机位置
|
||
lat, lon = 124, 30
|
||
|
||
# 转换为AIS格式
|
||
lat_ais = int((lat + 90) * 60000)
|
||
lon_ais = int((lon + 180) * 60000)
|
||
|
||
# 随机选择一艘船
|
||
mmsi = 200000000
|
||
COMM_STATES = [
|
||
{"sync_state": 0, "slot_timeout": 0, "sub_message": 0},
|
||
{"sync_state": 1, "slot_timeout": 1, "sub_message": 1},
|
||
{"sync_state": 2, "slot_timeout": 2, "sub_message": 2},
|
||
{"sync_state": 3, "slot_timeout": 3, "sub_message": 3}
|
||
]
|
||
# 随机通信状态
|
||
comm_state = random.choice(COMM_STATES)
|
||
comm_state_field = (
|
||
(comm_state["sync_state"] << 17) |
|
||
(comm_state["slot_timeout"] << 14) |
|
||
comm_state["sub_message"]
|
||
)
|
||
|
||
# 构建Message1 (168位/21字节)
|
||
msg = struct.pack('>I', (1 << 26) | (random.randint(0, 3) << 24) | (mmsi >> 6))
|
||
msg += struct.pack('>I', ((mmsi & 0x3F) << 26) | (random.randint(0, 15) << 22))
|
||
msg += struct.pack('>I', (random.randint(0, 255) << 24) | (random.randint(0, 1023) << 14))
|
||
msg += struct.pack('>I', (random.randint(0, 1) << 31) | (lon_ais & 0xFFFFFFF))
|
||
msg += struct.pack('>I', (lat_ais << 4) | (random.randint(0, 4095) >> 8))
|
||
msg += struct.pack('>I', ((random.randint(0, 4095) & 0xFF) << 24) | (random.randint(0, 3599) << 12))
|
||
msg += struct.pack('>I', (random.randint(0, 511) << 23) | (random.randint(0, 63) << 17) | (comm_state_field & 0x1FFFF))
|
||
|
||
return msg
|
||
gam = generate_ais_message1()
|
||
print(gam)
|
||
print(type(gam), len(gam))
|
||
def decode_ais_message1(data):
|
||
"""解码AIS Message1位置报告 (M.1371-5 Annex8 3.1)"""
|
||
# if len(data) != 21:
|
||
# raise ValueError("AIS消息1应为21字节")
|
||
|
||
# 将数据转换为整数列表
|
||
parts = [int.from_bytes(data[i:i+4], 'big') for i in range(0, 20, 4)]
|
||
parts.append(int.from_bytes(data[20:], 'big') << 24) # 最后一个字节
|
||
|
||
# 解码字段
|
||
decoded = {}
|
||
|
||
# 第一部分：消息ID + 转发指示符 + MMSI高24位
|
||
decoded['message_id'] = (parts[0] >> 26) & 0x3F
|
||
decoded['repeat_indicator'] = (parts[0] >> 24) & 0x03
|
||
decoded['user_id'] = ((parts[0] & 0x00FFFFFF) << 6)
|
||
|
||
# 第二部分：MMSI低6位 + 导航状态 + 旋转速率 + SOG高4位
|
||
decoded['user_id'] |= (parts[1] >> 26) & 0x3F
|
||
decoded['nav_status'] = (parts[1] >> 22) & 0x0F
|
||
rot_value = (parts[1] >> 14) & 0xFF
|
||
decoded['rot'] = rot_value if rot_value < 128 else rot_value - 256
|
||
decoded['sog'] = ((parts[1] & 0x00003F00) >> 8) << 6
|
||
|
||
# 第三部分：SOG低6位 + 位置准确度 + 经度高21位
|
||
decoded['sog'] |= (parts[2] >> 26) & 0x3F
|
||
decoded['pos_acc'] = (parts[2] >> 25) & 0x01
|
||
decoded['longitude'] = ((parts[2] & 0x01FFFFFF) << 7)
|
||
|
||
# 第四部分：经度低7位 + 纬度
|
||
decoded['longitude'] |= (parts[3] >> 25) & 0x7F
|
||
decoded['latitude'] = parts[3] & 0x07FFFFFF
|
||
|
||
# 第五部分：COG + 实际航向 + 时戳高3位
|
||
decoded['cog'] = (parts[4] >> 20) & 0x0FFF
|
||
decoded['true_heading'] = (parts[4] >> 11) & 0x01FF
|
||
decoded['timestamp'] = ((parts[4] & 0x00000700) >> 8) << 3
|
||
|
||
# 第六部分：时戳低3位 + RAIM标志 + 通信状态高10位
|
||
decoded['timestamp'] |= (parts[5] >> 29) & 0x07
|
||
decoded['raim'] = (parts[5] >> 24) & 0x01
|
||
decoded['comm_state'] = ((parts[5] & 0x00FFFFFF) >> 14) << 9
|
||
|
||
# 第七部分：通信状态低9位
|
||
decoded['comm_state'] |= (parts[6] >> 23) & 0x1FF
|
||
|
||
# 添加spare字段（协议中为保留位，固定为0）
|
||
decoded['spare'] = 0
|
||
|
||
return decoded
|
||
|
||
# 示例数据（从您的生成函数中获取）
|
||
data = b'\x06/\xaf\x08\x03@\x00\x00\x11\x93\x80\x00\x80\xc0B\xc0\x0c>\xc4\x00*\x8e\xa0\x00\xba\x02@\x01'
|
||
|
||
# 解码数据
|
||
decoded = decode_ais_message1(data)
|
||
|
||
# 打印解码结果
|
||
print("解码结果:")
|
||
for field, value in decoded.items():
|
||
print(f"{field}: {value}")
|
||
|
||
# # 生成AIS消息（使用您的函数）
|
||
# data = generate_ais_message1()
|
||
# print("生成的二进制数据:", data)
|
||
|
||
# # 解码数据
|
||
# decoded = decode_ais_message1(data)
|
||
|
||
# # 打印格式化结果
|
||
# print("\n解码结果:")
|
||
# print(f"消息ID: {decoded['message_id']}")
|
||
# print(f"转发指示符: {decoded['repeat_indicator']}")
|
||
# print(f"用户ID (MMSI): {decoded['user_id']}")
|
||
|
||
# # 导航状态描述
|
||
# nav_status_desc = {
|
||
# 0: "发动机使用中", 1: "锚泊", 2: "未操纵", 3: "有限适航性",
|
||
# 4: "受船舶吃水限制", 5: "系泊", 6: "搁浅", 7: "从事捕捞",
|
||
# 8: "航行中", 15: "未规定(默认)"
|
||
# }
|
||
# print(f"导航状态: {decoded['nav_status']} ({nav_status_desc.get(decoded['nav_status'], '未知状态')})")
|
||
|
||
# # 旋转速率描述
|
||
# rot = decoded['rot']
|
||
# if rot == -128:
|
||
# print("旋转速率: -128 (无可用旋转信息)")
|
||
# elif rot == 127:
|
||
# print("旋转速率: 127 (右旋超过5º/30s)")
|
||
# elif rot == -127:
|
||
# print("旋转速率: -127 (左旋超过5º/30s)")
|
||
# else:
|
||
# print(f"旋转速率: {rot}")
|
||
|
||
# # 位置准确度描述
|
||
# print(f"位置准确度: {decoded['pos_acc']} ({'高(>10m)' if decoded['pos_acc'] else '低(<10m)'})")
|
||
|
||
# # 转换为实际单位
|
||
# longitude_deg = decoded['longitude'] / (60 * 10000)
|
||
# latitude_deg = decoded['latitude'] / (60 * 10000)
|
||
# sog_knots = decoded['sog'] / 10.0
|
||
# cog_deg = decoded['cog'] / 10.0
|
||
|
||
# print(f"经度: {decoded['longitude']} (1/10,000 min) ≈ {longitude_deg:.5f}°")
|
||
# print(f"纬度: {decoded['latitude']} (1/10,000 min) ≈ {latitude_deg:.5f}°")
|
||
# print(f"地面航速: {decoded['sog']} (1/10节) ≈ {sog_knots:.1f}节")
|
||
# print(f"地面航线: {decoded['cog']} (1/10°) ≈ {cog_deg:.1f}°")
|
||
# print(f"实际航向: {decoded['true_heading']}°")
|
||
# print(f"时戳: {decoded['timestamp']}秒")
|
||
# print(f"RAIM标志: {decoded['raim']} ({'RAIM正在使用' if decoded['raim'] else 'RAIM未使用'})")
|
||
# print(f"通信状态: {decoded['comm_state']}") |