# -*- coding: utf-8 -*- import cmath from collections import defaultdict import math from BatchSettingMoudle.BatchSettingReader import StrCut from GNSSCheckMoudle.GNSSBaseInfo import * import os import numpy as np class GNSSDataCheck: def __init__(self,data_path,settings,gnssflag): self.AnubisFile=data_path self.GNSSSys = [] self.SNRFlag = [] self.GNSSFre = defaultdict(list) self.GNSSSNR = defaultdict(list) self.GNSSSNRAVE = {} self.GNSSSNRSTD = defaultdict(float) self.ReadAnubisFile() self.SNRAVESTD() gnssflag[0]=self.GNSSDataJudge(settings) def ReadAnubisFile(self): if os.path.exists(self.AnubisFile): with open(self.AnubisFile, 'r') as file : while True: end_flag=1 buf=file.readline() #观测值信息 if buf.find("#====== Observation types")!=-1: #读取卫星系统 buf=file.readline().split() for i in range(int(buf[3])): self.GNSSSys.append(buf[4+i]) #读取频点 while True: buf=file.readline() if buf.find("=GPSOBS")!=-1 or buf.find("=GALOBS")!=-1 or buf.find("=GLOOBS")!=-1 or buf.find("=BDSOBS")!=-1 or buf.find("=QZSOBS")!=-1 or buf.find("=SBSOBS")!=-1: break for i in range(len(self.GNSSSys)): #GPS if buf.find(Gsyslist[0])!=-1: for i in range(len(GNSSFRE["GPS_L1"])): if buf.find(GNSSFRE["GPS_L1"][i])!=-1: self.GNSSFre["GPS"].append("L1") break for i in range(len(GNSSFRE["GPS_L2"])): if buf.find(GNSSFRE["GPS_L2"][i])!=-1: self.GNSSFre["GPS"].append("L2") break for i in range(len(GNSSFRE["GPS_L5"])): if buf.find(GNSSFRE["GPS_L5"][i])!=-1: self.GNSSFre["GPS"].append("L5") break #GLONASS if buf.find(Gsyslist[1])!=-1: for i in range(len(GNSSFRE["GLO_L1"])): if buf.find(GNSSFRE["GLO_L1"][i])!=-1: self.GNSSFre["GLO"].append("G1") break for i in range(len(GNSSFRE["GLO_L2"])): if buf.find(GNSSFRE["GLO_L2"][i])!=-1: self.GNSSFre["GLO"].append("G2") break for i in range(len(GNSSFRE["GLO_L3"])): if buf.find(GNSSFRE["GLO_L3"][i])!=-1: self.GNSSFre["GLO"].append("G3") break #BDS if buf.find(Gsyslist[2])!=-1: for i in range(len(GNSSFRE["BDS_B1"])): if buf.find(GNSSFRE["BDS_B1"][i])!=-1: self.GNSSFre["BDS"].append("B1") break for i in range(len(GNSSFRE["BDS_B2"])): if buf.find(GNSSFRE["BDS_B2"][i])!=-1: self.GNSSFre["BDS"].append("B2") break for i in range(len(GNSSFRE["BDS_B3"])): if buf.find(GNSSFRE["BDS_B3"][i])!=-1: self.GNSSFre["BDS"].append("B3") break #GALILEO if buf.find(Gsyslist[3])!=-1: for i in range(len(GNSSFRE["GAL_E1"])): if buf.find(GNSSFRE["GAL_E1"][i])!=-1: self.GNSSFre["GAL"].append("E1") break for i in range(len(GNSSFRE["GAL_E5a"])): if buf.find(GNSSFRE["GAL_E5a"][i])!=-1: self.GNSSFre["GAL"].append("E5a") break for i in range(len(GNSSFRE["GAL_E5b"])): if buf.find(GNSSFRE["GAL_E5b"][i])!=-1: self.GNSSFre["GAL"].append("E5b") break #QZSS if buf.find(Gsyslist[4])!=-1: for i in range(len(GNSSFRE["QZSS_L1"])): if buf.find(GNSSFRE["QZSS_L1"][i])!=-1: self.GNSSFre["QZS"].append("L1") break for i in range(len(GNSSFRE["QZSS_L2"])): if buf.find(GNSSFRE["QZSS_L2"][i])!=-1: self.GNSSFre["QZS"].append("L2") break for i in range(len(GNSSFRE["QZSS_L5"])): if buf.find(GNSSFRE["QZSS_L5"][i])!=-1: self.GNSSFre["QZS"].append("L5") break buf=file.readline() #信噪比 SNRFlag=[] snr_end_flag=1 if buf.find("#====== Signal to noise ratio")!=-1: while True: buf=file.readline() if len(buf)!=1: if buf.find("#")==-1 : if buf.find("=")!=-1: buf=StrCut('=', '\n', buf).split() self.GNSSSNR[buf[0]] self.SNRFlag.append(buf[0]) else: for i in range(len(self.SNRFlag)): if buf.find(self.SNRFlag[i])!=-1: buf=buf.split() self.GNSSSNR[self.SNRFlag[i]].append(buf[3]) if self.SNRFlag[i]==self.SNRFlag[-1]: snr_end_flag=0 break if snr_end_flag==0 and len(self.GNSSSNR[self.SNRFlag[-1]])==len(self.GNSSSNR[self.SNRFlag[-2]]): end_flag=0 break if end_flag==0: break else: print("没有找到Anubis结果文件") def SNRAVESTD(self): for i in range(len(self.GNSSSNR)): self.GNSSSNR[self.SNRFlag[i]]=np.array((self.GNSSSNR[self.SNRFlag[i]]),dtype=float) self.GNSSSNRAVE[self.SNRFlag[i]]=np.mean(self.GNSSSNR[self.SNRFlag[i]]) for j in range(len(self.GNSSSNR[self.SNRFlag[i]])): self.GNSSSNRSTD[self.SNRFlag[i]]+=math.pow(self.GNSSSNR[self.SNRFlag[i]][j]-self.GNSSSNRAVE[self.SNRFlag[i]],2) self.GNSSSNRSTD[self.SNRFlag[i]]/=len(self.GNSSSNR[self.SNRFlag[i]]) self.GNSSSNRSTD[self.SNRFlag[i]]=self.GNSSSNRSTD[self.SNRFlag[i]]**0.5 #0=GNSS正常 1=GNSS异常 def GNSSDataJudge(self,settings): with open(settings.ResSavePath, "a") as f: SysHealth=0 GPSFreHealth=0 GLOFreHealth=0 BDSFreHealth=0 GALFreHealth=0 QZSFreHealth=0 SNRHealth=0 f.write("# GNSS异常检测信息："+"\n") f.write(" > 卫星系统检测如下："+"\n") for i in range(len(settings.GNSS_SatSYS)): if settings.GNSS_SatSYS[i] in self.GNSSSys: if settings.GNSS_SatSYS[i]==settings.GNSS_SatSYS[-1]: if SysHealth==0: f.write(" 卫星系统正常"+"\n") continue else: SysHealth=1 f.write(" 缺少"+settings.GNSS_SatSYS[i]+"卫星"+"\n") f.write(" > 卫星频点检测如下："+"\n") f.write(" GPS频点："+"\n") for i in range(len(settings.GNSS_GPSFreq)): if settings.GNSS_GPSFreq[i] in self.GNSSFre["GPS"]: if settings.GNSS_GPSFreq[i]==settings.GNSS_GPSFreq[-1]: if GPSFreHealth==0: f.write(" GPS频点正常"+"\n") continue else: GPSFreHealth=1 f.write(" 缺少GPS_"+settings.GNSS_GPSFreq[i]+"频点"+"\n") f.write(" GLO频点："+"\n") for i in range(len(settings.GNSS_GLOFreq)): if settings.GNSS_GLOFreq[i] in self.GNSSFre["GLO"]: if settings.GNSS_GLOFreq[i]==settings.GNSS_GLOFreq[-1]: if GLOFreHealth==0: f.write(" GLO频点正常"+"\n") continue else: GLOFreHealth=1 f.write(" 缺少GLO_"+settings.GNSS_GLOFreq[i]+"频点"+"\n") f.write(" BDS频点："+"\n") for i in range(len(settings.GNSS_BDSFreq)): if settings.GNSS_BDSFreq[i] in self.GNSSFre["BDS"]: if settings.GNSS_BDSFreq[i]==settings.GNSS_BDSFreq[-1]: if BDSFreHealth==0: f.write(" BDS频点正常"+"\n") continue else: BDSFreHealth=1 f.write(" 缺少BDS_"+settings.GNSS_BDSFreq[i]+"频点"+"\n") f.write(" GAL频点："+"\n") for i in range(len(settings.GNSS_GALFreq)): if settings.GNSS_GALFreq[i] in self.GNSSFre["GAL"]: if settings.GNSS_GALFreq[i]==settings.GNSS_GALFreq[-1]: if GALFreHealth==0: f.write(" GAL频点正常"+"\n") continue else: GALFreHealth=1 f.write(" 缺少GAL_"+settings.GNSS_GALFreq[i]+"频点"+"\n") f.write(" QZS频点："+"\n") for i in range(len(settings.GNSS_QZSFreq)): if settings.GNSS_QZSFreq[i] in self.GNSSFre["QZS"]: if settings.GNSS_QZSFreq[i]==settings.GNSS_QZSFreq[-1]: if QZSFreHealth==0: f.write(" QZS频点正常"+"\n") continue else: QZSFreHealth=1 f.write(" 缺少QZS_"+settings.GNSS_QZSFreq[i]+"频点"+"\n") f.write(" > 信噪比检测如下："+"\n") for i in range(len(self.SNRFlag)): if self.GNSSSNRAVE[self.SNRFlag[i]]>=settings.GNSS_MinAveSNR and self.GNSSSNRSTD[self.SNRFlag[i]]<=settings.GNSS_MaxSNRStd: # if self.SNRFlag[i]==self.SNRFlag[-1]: # f.write(" 信噪比正常"+"\n") continue else: SNRHealth=1 if self.GNSSSNRAVE[self.SNRFlag[i]]settings.GNSS_MaxSNRStd: f.write(" "+self.SNRFlag[i]+":SNR_STD过大"+"\n") if SNRHealth==0: f.write(" 信噪比正常"+"\n") if SysHealth==1 or GLOFreHealth==1 or BDSFreHealth==1 or GALFreHealth==1 or QZSFreHealth==1 or SNRHealth==1: print("GNSS异常") return 1 else: print("GNSS正常") return 0