368 lines
18 KiB
Python
368 lines
18 KiB
Python
import struct
|
||
import logging
|
||
from django.db import transaction
|
||
from vdes_utils.vdes_demod_data import VdesDemodData
|
||
from vdes_utils.local_inspection_monitor import LocalInspectionMonitor
|
||
from django.core.cache import cache
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class VdesDataParsing:
|
||
def __init__(self):
|
||
self.hello = "VDES数据解析工具类"
|
||
self.vdes_demod_data_parsing = VdesDemodData() #VDES解调数据解析类
|
||
self.local_inspection_monitor = LocalInspectionMonitor() #地检状态监控数据解析类
|
||
self.payload_fragments = [] # 用于存储长消息分片数据的数组
|
||
|
||
def parse_vdes_data(self, packet_data):
|
||
from system.models import GpsStatus, HistorySession, Contacts, Content, MessageFlag
|
||
from datetime import datetime
|
||
import time
|
||
#判断帧头是否符合琢研
|
||
# print(packet_data.hex())
|
||
if packet_data[:4].decode('ascii') == '####':
|
||
# print("开始解析报文帧")
|
||
packet_type = struct.unpack('>I', packet_data[4:8])[0] #获取报文类型 7为gps， 12为ais
|
||
packet_lenth = struct.unpack('>I', packet_data[8:12])[0]
|
||
packet_content = packet_data[12:]
|
||
print('报文类型：', packet_type)
|
||
if packet_type == 7:
|
||
# print(f"GPS内容长度: {packet_lenth} 字节")
|
||
data = self.get_gps_data(packet_content)
|
||
utc_time = data.get('utc_time', 0)
|
||
EPOCH_DIFF = 1577836800 # 2020-01-01 00:00:00 到 1970-01-01 00:00:00 的秒数差
|
||
try:
|
||
if utc_time:
|
||
# utc_time 是从 1980-01-06 开始的秒数
|
||
unix_timestamp = utc_time + EPOCH_DIFF
|
||
utc_time_dt = datetime.fromtimestamp(unix_timestamp)
|
||
else:
|
||
utc_time_dt = datetime.now()
|
||
except (ValueError, TypeError, OSError):
|
||
utc_time_dt = datetime.now()
|
||
# print(data)
|
||
# print(utc_time_dt)
|
||
#下面代码还应该根据正负判断前端应该显示南北纬东西经
|
||
GpsStatus.objects.create(
|
||
utc_time=utc_time_dt,
|
||
longitude=data.get('longitude', 0),
|
||
latitude=data.get('latitude', 0),
|
||
altitude=data.get('altitude', 0),
|
||
used_satellites=data.get('used_sats', 10),
|
||
gps_satellites=data.get('gps_sats', 6),
|
||
beidou_satellites=data.get('beidou_sats', 4),
|
||
gps_status=data.get('gps_status', 1),
|
||
fix_mode=data.get('position_mode', 3),
|
||
hdop=data.get('hdop', 1.2),
|
||
speed=data.get('speed', 0),
|
||
direction=data.get('direction', 0)
|
||
)
|
||
elif packet_type == 8:
|
||
print(f"地检内容长度: {packet_lenth} 字节")
|
||
data = self.local_inspection_monitor.get_local_inspection_status_data(packet_content)
|
||
cache.set("local_inspection_data", data, timeout=None) #存入redis缓存中，让另一个进程能获取
|
||
|
||
elif packet_type == 12:
|
||
# print(packet_content)
|
||
data = self.get_ais_vdes_data(packet_content)
|
||
if data['message_type'] != 12:
|
||
return
|
||
# print(data)
|
||
if data['demodulator_id'] <= 5: #0-5为VDES解调数据
|
||
# 检查demod_data的前四个字节是否为'#$%'，'#$%'为本系统判断数据（文本图片视频）类型的标识
|
||
if len(data['demod_data']) >= 4 and data['demod_data'][:4] == b'#$%':
|
||
print("检测到特殊标识符'#$%'，打印数据:", data['demod_data'])
|
||
else:
|
||
self.vdes_demod_data_parsing.main(data['demod_data'])
|
||
# print(f"VDES内容长度: {packet_lenth} 字节")
|
||
|
||
# print(f"VDES内容长度: {packet_lenth} 字节")
|
||
# demod_data = self.get_vdes_demod_data(data['demod_data'])
|
||
# print('VDES解调数据:', demod_data)
|
||
# print("----------------------------")
|
||
# time.sleep(0.1)
|
||
# # 此处需要判断消息帧类型，卫星公告牌之类的----------------------------------------
|
||
# combined_data = self.vdes_data_processing(demod_data)
|
||
|
||
|
||
# # 此处需要判断接收的消息数据格式----------------------------------------
|
||
# if combined_data is not None:
|
||
# # print("完整消息:", combined_data)
|
||
# # 保存为图片文件
|
||
# import os
|
||
# from datetime import datetime
|
||
|
||
# # 创建保存目录(如果不存在)
|
||
# save_dir = 'E:\projects\wxwx\projects_job\VDES_Projects\报文模拟解析\地检设备模拟\\tcp_data'
|
||
# os.makedirs(save_dir, exist_ok=True)
|
||
|
||
# # 生成文件名(带时间戳)
|
||
# timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
# filename = os.path.join(save_dir, f'received_image_{timestamp}.jpg') # 假设是JPEG格式
|
||
|
||
# # 写入文件
|
||
# with open(filename, 'wb') as f:
|
||
# f.write(combined_data)
|
||
# print(f"图片已保存到: {filename}")
|
||
# return
|
||
|
||
|
||
|
||
# if combined_data is not None:
|
||
# MMSI = demod_data['Dest_id']
|
||
# print(f"发送者MMSI: {MMSI}")
|
||
# print(f"消息: {combined_data}")
|
||
# # 查询或创建通讯录记录
|
||
# contact, created = Contacts.objects.get_or_create(MMSI=MMSI)
|
||
# Id = contact.Id
|
||
# contact, created = HistorySession.objects.get_or_create(FormId=Id, MMSI=MMSI)
|
||
# content = Content.objects.create(
|
||
# SendId=Id,
|
||
# ReciverId=1,
|
||
# Content="<p>"+combined_data+"</p>",
|
||
# Type=0, # 0=html文本
|
||
# State=1, # 已发送
|
||
# NoCode=str(int(time.time())) , # 你可以放时间戳
|
||
# CreateDateUtc=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
# ReadFlag=0,
|
||
# SoundStatus=0
|
||
# )
|
||
# flag = MessageFlag.objects.first()
|
||
# if flag:
|
||
# flag.has_new_message = True
|
||
# flag.save()
|
||
|
||
|
||
else: #6/7为AIS解调数据
|
||
try:
|
||
demod_data = self.get_ais_demod_data(data['demod_data'])
|
||
# print(demod_data)
|
||
# 处理并存储AIS解调数据
|
||
self.ais_data_processing(demod_data)
|
||
except ValueError as e:
|
||
print(f"跳过不支持的AIS消息: {e}")
|
||
|
||
def get_gps_data(self, packet_content):
|
||
# 解析前40字节的GPS信息
|
||
fields = struct.unpack('>QiiiHHHBBIII', packet_content[:40])
|
||
# 解析卫星状态数据
|
||
satellite_data = []
|
||
remaining_data = packet_content[40:]
|
||
|
||
# 每6字节为一组卫星数据
|
||
for i in range(0, len(remaining_data), 6):
|
||
satellite_bytes = remaining_data[i:i+6]
|
||
if len(satellite_bytes) < 6:
|
||
break # 不足6字节的数据丢弃
|
||
|
||
# 解包卫星数据
|
||
sat_type, prn, snr, elevation, azimuth = struct.unpack('>BBBBH', satellite_bytes)
|
||
|
||
satellite_data.append({
|
||
'satellite_type': sat_type, # 卫星类型
|
||
'prn': prn, # 卫星PRN码
|
||
'snr': snr, # 卫星信噪比(dBHz)
|
||
'elevation': elevation, # 卫星仰角(0-90度)
|
||
'azimuth': azimuth # 卫星方位角(0-359度)
|
||
})
|
||
res = {}
|
||
# 将度分格式转换为十进制度格式
|
||
dm_to_decimal = lambda raw: (1 if raw >= 0 else -1) * (
|
||
int(abs(raw) // 10000000) + (abs(raw) % 10000000) / 10000000 / 60 * 100
|
||
)
|
||
res['utc_time'] = fields[0] # UTC时间 (8字节)
|
||
res['longitude'] = round(dm_to_decimal(fields[1]), 7)
|
||
res['latitude'] = round(dm_to_decimal(fields[2]), 7)
|
||
res['altitude'] = fields[3]/10 # 海拔高度 (4字节，50米)
|
||
res['used_sats'] = fields[4] # 定位使用的卫星数 (2字节)
|
||
res['gps_sats'] = fields[5] # GPS卫星数 (2字节)
|
||
res['beidou_sats'] = fields[6] # 北斗卫星数 (2字节)
|
||
res['gps_status'] = fields[7] # GPS状态 (1字节，单点定位)
|
||
res['position_mode'] = fields[8] # 定位模式 (1字节，3D定位)
|
||
res['hdop'] = fields[9]/10 # 水平精度因子 (4字节，25.5米)
|
||
res['speed'] = fields[10]/100 # 运动速度 (4字节，10.24节)
|
||
res['direction'] = fields[11]/100 # 运动方向 (4字节，359度)
|
||
res['satellite_data'] = satellite_data # 卫星状态数据列表
|
||
return res
|
||
|
||
def get_ais_vdes_data(self, packet_content):
|
||
res ={}
|
||
demod_packet = packet_content[:11]
|
||
demod_data = packet_content[11:]
|
||
fields = struct.unpack('>BBBHHHH', demod_packet)
|
||
res['message_type'] = fields[0]
|
||
res['satellite_id'] = fields[1]
|
||
res['demodulator_id'] = fields[2]
|
||
res['demod_slot'] = fields[3]
|
||
res['receive_delay'] = fields[4]
|
||
snr_packed = fields[5]
|
||
res['frequency_bias'] = fields[6]
|
||
# 解析SNR
|
||
snr_int = (snr_packed >> 8) & 0xFF
|
||
snr_frac = snr_packed & 0xFF
|
||
res['snr'] = snr_int + snr_frac / 100.0
|
||
res['demod_data'] = demod_data
|
||
# bit_string = "".join(f"{byte:08b}" for byte in demod_data)
|
||
# print(bit_string)
|
||
# print(demod_data.hex())
|
||
|
||
return res
|
||
|
||
def get_ais_demod_data(self, bit_string):
|
||
# 如果输入是字节类型，先转换为二进制字符串
|
||
if isinstance(bit_string, bytes):
|
||
bit_string = ''.join(format(byte, '08b') for byte in bit_string)
|
||
pos = 0
|
||
decoded = {}
|
||
|
||
decoded['message_id'] = int(bit_string[pos:pos+6], 2) # 消息ID (6 bits)
|
||
# 检查message_id是否为1、2或3，如果不是则抛出异常 （当前程序只解析了位置报告，还不支持其他ais报文）
|
||
if decoded['message_id'] not in [1, 2, 3]:
|
||
raise ValueError(f"Unsupported message_id: {decoded['message_id']}")
|
||
pos += 6
|
||
decoded['repeat_indicator'] = int(bit_string[pos:pos+2], 2) # 转发指示符 (2 bits)
|
||
pos += 2
|
||
decoded['user_id'] = int(bit_string[pos:pos+30], 2) # 用户ID/MMSI (30 bits)
|
||
pos += 30
|
||
decoded['navigation_status'] = int(bit_string[pos:pos+4], 2) # 导航状态 (4 bits)
|
||
pos += 4
|
||
|
||
# 旋转速率 (8位有符号整数)
|
||
rot_value = int(bit_string[pos:pos+8], 2)
|
||
if rot_value > 127:
|
||
rot_value -= 256
|
||
decoded['rot'] = rot_value
|
||
pos += 8
|
||
|
||
decoded['sog'] = int(bit_string[pos:pos+10], 2) # 地面航速 (10 bits)
|
||
pos += 10
|
||
decoded['position_accuracy'] = int(bit_string[pos], 2) # 位置准确度 (1 bit)
|
||
pos += 1
|
||
|
||
# 经度 (28位)
|
||
longitude_value = int(bit_string[pos:pos+28], 2)
|
||
decoded['longitude'] = longitude_value / (60 * 10000) # 转换为度
|
||
pos += 28
|
||
|
||
# 纬度 (27位)
|
||
latitude_value = int(bit_string[pos:pos+27], 2)
|
||
decoded['latitude'] = latitude_value / (60 * 10000) # 转换为度
|
||
pos += 27
|
||
|
||
decoded['cog'] = int(bit_string[pos:pos+12], 2) # 地面航线 (12 bits)
|
||
pos += 12
|
||
decoded['true_heading'] = int(bit_string[pos:pos+9], 2) # 实际航向 (9 bits)
|
||
pos += 9
|
||
decoded['timestamp'] = int(bit_string[pos:pos+6], 2) # 时戳 (6 bits)
|
||
pos += 6
|
||
decoded['special_maneuver'] = int(bit_string[pos:pos+2], 2) # 特定操纵指示符 (2 bits)
|
||
pos += 2
|
||
decoded['spare'] = int(bit_string[pos:pos+3], 2) # 备用 (3 bits)
|
||
pos += 3
|
||
decoded['raim_flag'] = int(bit_string[pos], 2) # RAIM标志 (1 bit)
|
||
pos += 1
|
||
decoded['comm_state'] = int(bit_string[pos:pos+19], 2) # 通信状态 (19 bits)
|
||
|
||
return decoded
|
||
|
||
def get_vdes_demod_data(self, packed_data):
|
||
"""解码VDES解调信息字段"""
|
||
header_size = struct.calcsize('>B H I B B I H')
|
||
header = packed_data[:header_size]
|
||
Type, payload_size, source_id, satellite_id, session_id, dest_id, fragment_num = \
|
||
struct.unpack('>B H I B B I H', header)
|
||
# 计算负载字节长度进去截取，去除帧尾
|
||
payload = packed_data[header_size:header_size+payload_size - 12]
|
||
# 将UTF-8编码的二进制payload解码为字符串
|
||
try:
|
||
decoded_payload = payload.decode('utf-8')
|
||
except UnicodeDecodeError:
|
||
decoded_payload = payload # 如果解码失败，保留原始二进制数据
|
||
|
||
return {
|
||
'Type': Type,
|
||
'Payload_size': payload_size,
|
||
'Source_id': source_id,
|
||
'Satellite_id': satellite_id,
|
||
'Session_id': session_id,
|
||
'Dest_id': dest_id,
|
||
'Fragment_num': fragment_num,
|
||
'Payload': decoded_payload # 返回解码后的字符串
|
||
}
|
||
|
||
def ais_data_processing(self, demod_data):
|
||
from system.models import ShipLatestStatus
|
||
from datetime import datetime
|
||
# 转换timestamp为datetime对象
|
||
timestamp = demod_data.get('timestamp', 0)
|
||
try:
|
||
timestamp_dt = datetime.fromtimestamp(timestamp) if timestamp else datetime.now()
|
||
except (ValueError, TypeError):
|
||
timestamp_dt = datetime.now()
|
||
|
||
ship_data = {
|
||
'mmsi': str(demod_data.get('user_id', '000000000')),
|
||
'latitude': demod_data.get('latitude', 0),
|
||
'longitude': demod_data.get('longitude', 0),
|
||
'speed': demod_data.get('sog', 0),
|
||
'course': demod_data.get('cog', 0),
|
||
'heading': demod_data.get('true_heading', 0),
|
||
'nav_status': demod_data.get('navigation_status', 0),
|
||
'timestamp': timestamp_dt, # 使用转换后的datetime对象
|
||
'ship_name': '未知',
|
||
'ship_type': 0,
|
||
'draught': 0,
|
||
'destination': '未知'
|
||
}
|
||
with transaction.atomic():
|
||
# 尝试获取已有记录
|
||
ship, created = ShipLatestStatus.objects.get_or_create(
|
||
mmsi=ship_data['mmsi'],
|
||
defaults=ship_data
|
||
)
|
||
|
||
if not created:
|
||
# 更新已有记录
|
||
for field, value in ship_data.items():
|
||
setattr(ship, field, value)
|
||
ship.save()
|
||
|
||
|
||
# def vdes_data_processing(self, demod_data):
|
||
# """处理VDES分片数据"""
|
||
# fragment_num = demod_data['Fragment_num']
|
||
# print(self.payload_fragments)
|
||
# if demod_data['Type'] == 30: # 开始分片
|
||
# self.payload_fragments = [] # 清空数组
|
||
|
||
# elif demod_data['Type'] == 31: # 数据分片
|
||
# # 确保数组足够大以容纳当前分片
|
||
# while len(self.payload_fragments) <= fragment_num:
|
||
# self.payload_fragments.append(None)
|
||
# self.payload_fragments[fragment_num] = demod_data['Payload']
|
||
|
||
# elif demod_data['Type'] == 32: # 结束分片
|
||
# # 拼接所有有效分片
|
||
# combined_payload = ''.join([p for p in self.payload_fragments if p is not None])
|
||
# return combined_payload
|
||
# return None # 如果不是结束分片，返回None
|
||
def vdes_data_processing(self, demod_data):
|
||
"""处理VDES分片数据"""
|
||
fragment_num = demod_data['Fragment_num']
|
||
# print(self.payload_fragments)
|
||
if demod_data['Type'] == 30: # 开始分片
|
||
print("------------------开始——————————————————————————————————————————————")
|
||
self.payload_fragments = [] # 清空数组
|
||
self.payload_fragments.append(demod_data['Payload'])
|
||
|
||
elif demod_data['Type'] == 31: # 数据分片
|
||
self.payload_fragments.append(demod_data['Payload'])
|
||
|
||
elif demod_data['Type'] == 32: # 结束分片
|
||
# 拼接所有有效分片
|
||
print("------------------结束——————————————————————————————————————————————")
|
||
self.payload_fragments.append(demod_data['Payload'])
|
||
combined_payload = b''.join([p for p in self.payload_fragments if p is not None])
|
||
print(len(self.payload_fragments))
|
||
return combined_payload
|
||
return None # 如果不是结束分片，返回None |