import struct import socket import time import random from random import uniform, randint import math import threading from vdes_message_unpack import VdesMessageUnpack vdes_message_queue = [] message_queue_lock = threading.Lock() SHIP_MMSIS = [random.randint(336295939, 336295969) for _ in range(10)] SHIP_STATUS = {} # 生成包含30艘船信息的列表（MMSI、纬度、经度、航向） SHIP_LIST = [ ( 336295939 + i, # MMSI号 round(random.uniform(30.0, 34.0), 6), round(random.uniform(123.5, 126.0), 6), random.randint(0, 359) ) for i in range(30) ] # 初始化船只状态 for ship in SHIP_LIST: mmsi, init_lat, init_lon, heading = ship SHIP_STATUS[mmsi] = { 'current_lat': init_lat, 'current_lon': init_lon, 'heading': heading, 'speed': 280.24 #加速100倍用于演示 } YANGTZE_ESTUARY_LAT = 31.23 YANGTZE_ESTUARY_LON = 123.87 RANGE_NM = 5 # 5海里范围 current_lat = YANGTZE_ESTUARY_LAT # 当前纬度 current_lon = YANGTZE_ESTUARY_LON # 当前经度 own_heading = 90 # 固定航向 (50度) def generate_random_position(): """生成船只位置(基于当前位置和航向计算)""" global current_lat, current_lon, own_heading # 速度为10.24节(来自GPS报文设置) speed_knots = 300.24 #加速100倍用于演示 # 转换为公里/小时 (1节 = 1.852公里/小时) speed_kmh = speed_knots * 1.852 # 转换为公里/秒 (发送间隔为1秒) speed_kmps = speed_kmh / 3600 # 计算1秒内移动的距离(公里) distance_km = speed_kmps * 1 # 将航向转换为弧度 heading_rad = math.radians(own_heading) # 计算纬度和经度变化 # 1度纬度 ≈ 111公里 delta_lat = distance_km / 111.0 * math.cos(heading_rad) # 1度经度 ≈ 111公里 * cos(纬度) delta_lon = distance_km / (111.0 * math.cos(math.radians(current_lat))) * math.sin(heading_rad) # 更新当前位置 current_lat += delta_lat current_lon += delta_lon # 转换为度分格式*10^7 longitude = int(current_lon * 10000000) latitude = int(current_lat * 10000000) return longitude, latitude def generate_ship_position(mmsi): """根据船只MMSI生成当前位置""" global SHIP_STATUS ship = SHIP_STATUS[mmsi] #获取全局船只中的一个进行修改 speed_knots = ship['speed'] speed_kmh = speed_knots * 1.852 speed_kmps = speed_kmh / 3600 distance_km = speed_kmps * 1 # 航向转弧度 heading_rad = math.radians(ship['heading']) # 计算经纬度变化 delta_lat = distance_km / 111.0 * math.cos(heading_rad) delta_lon = distance_km / (111.0 * math.cos(math.radians(ship['current_lat']))) * math.sin(heading_rad) # 更新船只状态 ship['current_lat'] += delta_lat ship['current_lon'] += delta_lon # 返回新位置 return ship['current_lon'], ship['current_lat'] 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 create_gps_packet(): """创建GPS解析数据报文(类型7)的内容""" # 模拟GPS数据 (表11定义) lon, lat = generate_random_position() utc_time = 1680000000 # UTC时间 (8字节) longitude = lon # 经度 (4字节，东经120.5度) latitude = lat # 纬度 (4字节，北纬30.5度) altitude = 50 # 海拔高度 (4字节，50米) used_sats = 10 # 定位使用的卫星数 (2字节) gps_sats = 8 # GPS卫星数 (2字节) beidou_sats = 7 # 北斗卫星数 (2字节) gps_status = 1 # GPS状态 (1字节，单点定位) position_mode = 3 # 定位模式 (1字节，3D定位) hdop = 2550 # 水平精度因子 (4字节，25.5米) speed = 1024 # 运动速度 (4字节，10.24节) direction = own_heading # 运动方向 (4字节，359度) # 打包为二进制 (大端字节序) content = struct.pack( '>QiiiHHHBBIII', # 格式: Q(8),i(4),i(4),i(4),H(2),H(2),H(2),B(1),B(1),I(4),I(4),I(4) utc_time, longitude, latitude, altitude, used_sats, gps_sats, beidou_sats, gps_status, position_mode, hdop, speed, direction ) return content def create_ais_vdes_info_packet(demod_content, Demodulator_id): """创建解调信息报文(类型12)的内容""" # 模拟解调信息数据 message_type = 12 # 消息类型 (1字节) satellite_id = 1 # 卫星ID (1字节) demodulator_id = Demodulator_id # 解调器编号 (1字节) demod_slot = 1234 # 解调时隙 (2字节) receive_delay = 5678 # 接收延迟 (2字节，单位us) snr = 25.75 # 信噪比 (2字节，dB) # 处理信噪比 (高8位整数部分，低8位小数部分) snr_int = int(snr) snr_frac = int((snr - snr_int) * 100) # 转换为0-99的小数部分 snr_packed = (snr_int & 0xFF) << 8 | (snr_frac & 0xFF) # 打包为二进制 (大端字节序) content = struct.pack( '>BBBHHH', # 格式: B(1),B(1),B(1),H(2),H(2),h(2) message_type, satellite_id, demodulator_id, demod_slot, receive_delay, snr_packed ) return content + demod_content def encode_ais_demod_fields(SHIP_MMSI, lat, lon, heading): """编码AIS解调信息字段""" # 初始化二进制位列表 bits = [] # 示例数据 message_id = 1 # 消息ID (6 bits) repeat_indicator = 0 # 转发指示符 (2 bits) user_id = SHIP_MMSI # 用户ID/MMSI (30 bits) navigation_status = 8 # 导航状态 (4 bits) - 航行中 rot = -128 # 旋转速率 (8 bits) - 无信息 sog = 102 # 地面航速 (10 bits) - 10.2节 position_accuracy = 1 # 位置准确度 (1 bit) - 高 longitude, latitude = lon, lat # 经纬度 cog = heading # 地面航线 (12 bits) - 90.0° true_heading = heading # 实际航向 (9 bits) - 90° timestamp = 30 # 时戳 (6 bits) - UTC秒 special_maneuver = 0 # 特定操纵指示符 (2 bits) spare = 0 # 备用 (3 bits) raim_flag = 0 # RAIM标志 (1 bit) comm_state = 0 # 通信状态 (19 bits) # 添加各个字段（按照协议顺序） bits.append(format(message_id, '06b')) # 消息ID bits.append(format(repeat_indicator, '02b')) # 转发指示符 bits.append(format(user_id, '030b')) # 用户ID bits.append(format(navigation_status, '04b')) # 导航状态 # 旋转速率 (8位有符号整数) rot_bits = rot & 0xFF bits.append(format(rot_bits, '08b')) # 地面航速 (10位) bits.append(format(min(sog, 1023), '010b')) # 位置准确度 (1位) bits.append(str(position_accuracy)) # 经度 (28位) # 转换为1/10,000分钟单位 longitude_min = int(longitude * 60 * 10000) bits.append(format(longitude_min & 0x0FFFFFFF, '028b')) # 纬度 (27位) latitude_min = int(latitude * 60 * 10000) bits.append(format(latitude_min & 0x07FFFFFF, '027b')) # 地面航线 (12位) bits.append(format(min(cog, 3600), '012b')) # 实际航向 (9位) bits.append(format(min(true_heading, 511), '09b')) # 时戳 (6位) bits.append(format(min(timestamp, 63), '06b')) # 特定操纵指示符 (2位) bits.append(format(special_maneuver, '02b')) # 备用 (3位) bits.append(format(spare, '03b')) # RAIM标志 (1位) bits.append(str(raim_flag)) # 通信状态 (19位) bits.append(format(comm_state, '019b')) # 合并所有位 bit_string = ''.join(bits) # 验证总位数 total_bits = len(bit_string) if total_bits != 168: print(f"警告: 总位数应为168，实际为{total_bits}") # 将位字符串转换为字节 binary_int = int(bit_string, 2) # 将二进制字符串转换为整数 byte_length = (len(bit_string) + 7) // 8 # 计算需要的字节数 binary_data = binary_int.to_bytes(byte_length, 'big') # 转换为字节 return binary_data def encode_vdes_demod_fields(Type, Fragment_num, MMSI, Payload): """编码VDES解调信息字段""" source_id = 12345678 # 信源电台ID satellite_id = 1 # 卫星ID session_id = 100 # 会话ID dest_id = MMSI # 目的地电台ID fragment_num = Fragment_num # 片段编号 binary_payload = Payload # 载荷内容 # 计算载荷大小 (字段3到8的大小) payload_size = 4 + 1 + 1 + 4 + 2 + len(binary_payload) # 打包为二进制 (大端字节序) content = struct.pack( '>B H I B B I H', # 格式: B(1),H(2),I(4),B(1),B(1),I(4),H(2) Type, # 类型 (1字节) payload_size, # 载荷大小 (2字节) source_id, # 信源电台ID (4字节) satellite_id, # 卫星ID (1字节) session_id, # 会话ID (1字节) dest_id, # 目的地电台ID (4字节) fragment_num # 片段编号 (2字节) ) return content + binary_payload # 添加载荷内容 def print_binary(data): """打印二进制数据""" print("十六进制表示:", data.hex()) print("二进制表示:", ' '.join(format(byte, '08b') for byte in data)) def handle_client(conn, addr): """处理客户端连接的线程函数""" print(f"客户端连接来自: {addr}") unpacker = VdesMessageUnpack() mmsi: int fragment_num = 0 # 创建发送线程 send_thread = threading.Thread(target=send_messages, args=(conn,)) send_thread.daemon = True send_thread.start() # 接收消息 while True: data = conn.recv(65536) if data: # print(f"收到VDES客户端消息: {data.hex()}") try: frame_type, message = unpacker.unpack_tcp_packet(data) mmsi = unpacker.get_mmsi() if frame_type == 0: # 起始片段 Type = 30 elif frame_type == 1: # 延续片段 Type = 31 elif frame_type == 2: # 结束片段 Type = 32 vdes_demod_content = encode_vdes_demod_fields(Type, fragment_num, mmsi, message) vdes_data = create_ais_vdes_info_packet(vdes_demod_content, 0) vdes_packet = create_tcp_packet(12, vdes_data) fragment_num += 1 # 将消息添加到全局队列 # with message_queue_lock: vdes_message_queue.append(vdes_packet) if not unpacker.get_unpack_statu(): fragment_num = 0 # print("解析消息完成") except ValueError as e: print(f"报文解析错误: {e}") def send_messages(conn): """持续发送消息的线程函数""" try: while True: # 检查并发送队列中的VDES消息 # with message_queue_lock: # print(11111111111111111111111) # print(vdes_message_queue) if vdes_message_queue: print(vdes_message_queue) packet = vdes_message_queue.pop(0) conn.sendall(packet) print(f"发送VDES报文: {packet.hex()}") time.sleep(0.5) # 模拟发送间隔 # 生成并发送AIS报文 # selected_ship = random.choice(SHIP_LIST) # mmsi = selected_ship[0] # lon, lat = generate_ship_position(mmsi) # heading = SHIP_STATUS[mmsi]['heading'] # ais_demod_content = encode_ais_demod_fields(mmsi, lat, lon, heading) # ais_data = create_ais_vdes_info_packet(ais_demod_content, 4) # ais_packet = create_tcp_packet(12, ais_data) # print(f"发送AIS报文: {ais_packet.hex()}") # conn.sendall(ais_packet) # # 生成并发送GPS报文 # gps_content = create_gps_packet() # gps_packet = create_tcp_packet(7, gps_content) # conn.sendall(gps_packet) # time.sleep(2) except (ConnectionResetError, BrokenPipeError): print("客户端断开连接") def send_data_continuously(): """启动Socket服务端""" host = '127.0.0.1' port = 10 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s.bind((host, port)) s.listen(1) print(f"Socket服务端已启动，监听 {host}:{port}") try: while True: conn, addr = s.accept() client_thread = threading.Thread(target=handle_client, args=(conn, addr)) client_thread.daemon = True client_thread.start() except KeyboardInterrupt: print("\n服务端已停止") except Exception as e: print(f"服务端错误: {e}") if __name__ == "__main__": send_data_continuously()