import struct import datetime import random import bitarray import pyproj class pkt_base: def __init__(self, pkt_type=0, pkt_len=0): self.pkt_header = b'####' self.pkt_type = struct.pack('!I', pkt_type) self.pkt_len = struct.pack('!I', pkt_len) self.pkt_payload = b'' def get_packet(self): return self.pkt_header + self.pkt_type + self.pkt_len + self.pkt_payload class pkt_gps(pkt_base): def __init__(self, longitude, latitude, altitude, speed, heading): super().__init__(pkt_type=7) self.gpstime = int(datetime.datetime.now().timestamp() - 315964782) # Timestamp in seconds since 1980.1.6 self.lon = int(longitude*10000000) # Longitude in degrees * 10^7 self.lat = int(latitude*10000000) # Latitude in degrees * 10^7 self.alt = int(altitude) # Altitude in millimeters self.sates_used = random.randint(2, 6) # Number of satellites used self.sates_gps = random.randint(2, 6) # Number of satellites GPS in view self.sates_bd = random.randint(2, 6) # Number of satellites BD in view self.gps_status = random.randint(0, 4) # gps status: 0=No fix, 1=2D fix, 2=3D fix, 3=DGPS, 4=RTK float, 5=RTK fixed self.mode = random.randint(1, 3) # mode: 1=none, 2=2D, 3=3D self.hdop = random.randint(100, 100000) # HDOP (Horizontal Dilution of Precision) in 0.1m self.speed = int(speed*100) # Speed in 0.01 knot self.heading = int(heading) # Heading in degrees def _fill_payload(self): self.pkt_payload = struct.pack('!QIIIHHHBBIII', self.gpstime, self.lon, self.lat, self.alt, self.sates_used, self.sates_gps, self.sates_bd, self.gps_status, self.mode, self.hdop, self.speed, self.heading) def get_packet(self): self._fill_payload() self.pkt_len = struct.pack('!I', len(self.pkt_payload)) return super().get_packet() class pkt_demod(pkt_base): def __init__(self): super().__init__(pkt_type=12) self.msg_type = 12 # 消息类型 self.sat_id = random.randint(0, 255) # 卫星ID self.demod_id = 0 # 解调器编号 self.time_slot = random.randint(10, 2249) # 解调时隙 self.recv_delay = random.randint(0, 65535) # 接收延迟 self.snr = random.randint(-128, 127) # 信噪比 def _fill_payload(self): self.pkt_payload = b'' self.pkt_payload = struct.pack('!BBBHHh', self.msg_type, self.sat_id, self.demod_id, self.time_slot, self.recv_delay, self.snr) class pkt_ais(pkt_demod): def __init__(self): super().__init__() self.demod_id = random.randint(4, 5) class pkt_ais_msg_1(pkt_ais): def __init__(self, mmsi, sog, lon, lat, heading): super().__init__() self.msg_id = 1 # message ID self.repeat_indicator = random.randint(0, 3) # Repeat indicator self.mmsi = mmsi # Maritime Mobile Service Identity self.nav_status = 0 # Navigation status self.rot = 0 # Rate of turn self.sog = int(sog * 10) # Speed over ground in 0.1 knot self.pos_acc = random.randint(0, 1) # Position accuracy self.lon = int(lon * 600000) # Longitude in 1/10000 min self.lat = int(lat * 600000) # Latitude in 1/10000 min self.cog = int(heading * 10) # Course over ground in 0.1 degree self.heading = int(heading) # Heading in degrees self.time_stamp = 0 # Time stamp self.manoeuvre_indicator = random.randint(0, 3) # Manuever indicator self.spare = 0 # Spare bits self.raim_flag = random.randint(0, 1) # RAIM flag self.comm_state = random.randint(0, 3) # Communication state def _fill_payload(self): super()._fill_payload() demod_data = bitarray.bitarray() demod_data += bitarray.bitarray(f"{self.msg_id:06b}") # Demodulation flag demod_data += bitarray.bitarray(f"{self.repeat_indicator:02b}") # Repeat indicator demod_data += bitarray.bitarray(f"{self.mmsi:030b}") # Maritime Mobile Service Identity demod_data += bitarray.bitarray(f"{self.nav_status:04b}") # Navigation status demod_data += bitarray.bitarray(f"{self.rot:08b}") # Rate of turn demod_data += bitarray.bitarray(f"{self.sog:010b}") # Speed over ground in 0.1 knot demod_data += bitarray.bitarray(f"{self.pos_acc:01b}") # Position accuracy demod_data += bitarray.bitarray(f"{self.lon:028b}") # Longitude in 1/10000 min demod_data += bitarray.bitarray(f"{self.lat:027b}") # Latitude in 1/10000 min demod_data += bitarray.bitarray(f"{self.cog:012b}") # Course over ground in 0.1 degree demod_data += bitarray.bitarray(f"{self.heading:09b}") # Heading in degrees demod_data += bitarray.bitarray(f"{self.time_stamp:06b}") # Time stamp demod_data += bitarray.bitarray(f"{self.manoeuvre_indicator:02b}") # Manoeuvre indicator demod_data += bitarray.bitarray(f"{self.spare:03b}") # Spare bits demod_data += bitarray.bitarray(f"{self.raim_flag:01b}") # RAIM flag demod_data += bitarray.bitarray(f"{self.comm_state:019b}") # Communication state if len(demod_data) != 168: raise ValueError("ais_msg_1 length is not 168 bits", len(demod_data)) self.pkt_payload += demod_data.tobytes() self.pkt_len = struct.pack('!I', len(self.pkt_payload)) def get_packet(self, distance=0): if distance > 0: # Use WGS84 ellipsoid. g = pyproj.Geod(ellps='WGS84') # Forward transformation - returns longitude, latitude, back azimuth of terminus points lon_new, lat_new, _ = g.fwd(self.lon / 600000, self.lat / 600000, self.cog, distance) self.lon = int(lon_new * 600000) self.lat = int(lat_new * 600000) self._fill_payload() return super().get_packet() class pkt_ais_msg_2(pkt_ais_msg_1): def __init__(self, mmsi, sog, lon, lat, heading): super().__init__(mmsi, sog, lon, lat, heading) self.msg_id = 2 # message ID class pkt_ais_msg_3(pkt_ais_msg_1): def __init__(self, mmsi, sog, lon, lat, heading): super().__init__(mmsi, sog, lon, lat, heading) self.msg_id = 3 # message ID class pkt_vdes(pkt_base): def __init__(self): self.pkt_payload = b'' def _fill_payload(self): # This method should be overridden in subclasses to fill the payload raise NotImplementedError("Subclasses should implement this method") def get_packet(self): self._fill_payload() self.pkt_len = struct.pack('!I', len(self.pkt_payload)) return super().get_packet()