############################################################################### # Copyright (c) 2011 - 2026 Trimble Inc. # $Id: NoPiUT_Common_Meas.py,v 1.19 2024/06/18 21:35:49 acartmel Exp $ ############################################################################### # # NoPiUT_Common_Meas.py # # Miscellaneous collection of basic NoPi diffs_combo file handling functions # common to NoPi_Display and NoPi_Stats_Checker # ############################################################################### import sys import os import datetime import string import shlex import gc from numpy import * from NoPiUT_Const import * ############################################################################### # Define required constants ############################################################################### # Data invalid value DATA_INVALID = 0xFFFF ############################################################################### # Check if specified data path is valid. Also, append the directory path with # '/' if it is missing one ############################################################################### def data_path_valid( data_path ) : # Check validity of user specified path to data directory if ( not( os.path.isdir( data_path ) ) ) : print('Data Path specified is invalid. Please verify data path') print('Terminating execution') sys.exit() # If data path is valid, ensure it is terminated with a slash else : end_slash = data_path[ len( data_path ) - 1 ] if ( ( end_slash != '/' ) & ( end_slash != '\\' ) ) : data_path = data_path + '/' return( data_path ) ############################################################################### # Load a NoPi results file for a single signal combo # Loads into global variables ############################################################################### def load_combo_file( diffs_summ, fname ) : # Don't use loadtxt as there seems to be a memory usage issue # on Linux # din.nbytes = 521MB # top reports python consuming ~3.2GB # din = loadtxt( fname ) # Load the file as a N,1 array # Reshape as needed to account for the number of columns in the data gc.collect() num_data_cols = get_num_data_cols( diffs_summ ) data_in = fromfile( fname, dtype=float, count=-1, sep=' ' ) data_in = data_in.reshape( int(len( data_in ) / num_data_cols), num_data_cols ) # Recover the flag bits # Read as float, convert to int data_flags = zeros( len( data_in ), int ) data_flags += data_in[:, num_data_cols - 1].astype(int) return( data_in, data_flags ) ############################################################################### # Return number of data columns in a diffs_combo file. Determined based # on file format specified in the diffs summary file. # FILE_FORMAT_EXTENDED has 9 columns # FILE_FORMAT_EXTENDED_BASE_ROVER has 12 columns ############################################################################### def get_num_data_cols( diffs_summ ) : # Default extended data num_data_cols = 9 ext_format = diffs_summ.file_format - 1 if ( (ext_format & ~cNoPiConst.DIFFS_FMT_VLD_MASK) == 0 ): if ( ext_format & cNoPiConst.DIFFS_FMT_BASE_ROVER ) : # Inc. Az/El/CNo at rover num_data_cols += 3 if ( ext_format & cNoPiConst.DIFFS_FMT_DOPPLER ) : # Inc. D.D. Doppler num_data_cols += 1 else : print('Unsupported file format specified in summary file') sys.exit() return( num_data_cols ) ############################################################################### # Return the column to access in the data array to access the flags data # type # This is dependent on the code in NoPi that generates the data ############################################################################### def get_flags_column( diffs_summ ) : num_data_cols = get_num_data_cols( diffs_summ ) return( num_data_cols - 1 ) ############################################################################### # Return the flags mask required to test the validity of NoPi data for the # requested measurement type # This is dependent on the code in NoPi that generates the data and flags ############################################################################### def get_flags_mask( meas_type ) : if ( meas_type == cNoPiConst.MEAS_TYPE_DD_CARR ) : flags_msk = cNoPiConst.DIFFS_FLG_CARR_VLD elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_CODE ) : flags_msk = cNoPiConst.DIFFS_FLG_CODE_VLD elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_DOPP ) : flags_msk = cNoPiConst.DIFFS_FLG_DOPP_VLD elif ( meas_type == cNoPiConst.MEAS_TYPE_SD_CNO ) : flags_msk = cNoPiConst.DIFFS_FLG_CNO_VLD else : print('Unknown measurement type') sys.exit() return( flags_msk ) ############################################################################### # Return the column to access in the data array for the requested measurement # type # This is dependent on the code in NoPi that generates the data ############################################################################### def get_data_column( diffs_summ, meas_type ) : ext_format = diffs_summ.file_format - 1 if ( meas_type == cNoPiConst.MEAS_TYPE_DD_CARR ) : data_col = 7 elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_CODE ) : data_col = 6 elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_DOPP and (ext_format & cNoPiConst.DIFFS_FMT_DOPPLER) ) : data_col = 8 elif ( meas_type == cNoPiConst.MEAS_TYPE_SD_CNO ) : data_col = 5 else : print('Unknown measurement type') sys.exit() # Increment data column by 3 if file includes rover Az/El/CNo if ( ext_format & cNoPiConst.DIFFS_FMT_BASE_ROVER ) : data_col += 3 return( data_col ) ############################################################################### # Return base/rover elevation data column ############################################################################### def get_elev_data_column( station_name ) : if ( station_name == 'Base' ) : elev_data_col = 3 elif ( station_name == 'Rover' ) : elev_data_col = 6 else : print('Invalid station name: Must be \'Base\' or \'Rover\' ') sys.exit return( elev_data_col ) ############################################################################### # Return base/rover raw C/No data column ############################################################################### def get_cno_data_column( station_name ) : if ( station_name == 'Base' ) : cno_data_col = 4 elif ( station_name == 'Rover' ) : cno_data_col = 7 else : print('Invalid station name: Must be \'Base\' or \'Rover\' ') sys.exit return( cno_data_col ) ############################################################################### # Return the scale factor to apply to the NoPi data for the requested # measurement type # This is dependent on the code in NoPi that generates the data and the desired # display units (see get_meas_units() too) ############################################################################### def get_data_scale( meas_type ) : if ( meas_type == cNoPiConst.MEAS_TYPE_DD_CARR ) : scale = 1000.0 elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_CODE ) : scale = 1.0 elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_DOPP ) : scale = 1.0 elif ( meas_type == cNoPiConst.MEAS_TYPE_SD_CNO ) : scale = 1.0 / 10.0 else : print('Unknown measurement type') sys.exit() return( scale ) ############################################################################### # Ref or test sat_type/sub_type are stored in four different objects in # diffs_summ # This function returns the combo name as a single string ############################################################################### def construct_combo_name( diffs_summ, index ) : combo_name = '%s %s %s %s' % \ ( diffs_summ.combo[ index ].ref_sat_type, \ diffs_summ.combo[ index ].ref_sub_type, \ diffs_summ.combo[ index ].tst_sat_type, \ diffs_summ.combo[ index ].tst_sub_type ) return( combo_name ) ############################################################################### # Return the short name of the requested measurement type Used in the file # naming ############################################################################### def get_meas_name( meas_type ) : if ( meas_type == cNoPiConst.MEAS_TYPE_DD_CARR ) : mname = 'carr' elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_CODE ) : mname = 'code' elif ( meas_type == cNoPiConst.MEAS_TYPE_DD_DOPP ) : mname = 'dopp' elif ( meas_type == cNoPiConst.MEAS_TYPE_SD_CNO ) : mname = 'cno' else : print('Unknown measurement type') sys.exit() return( mname ) ############################################################################### # Return the display satellite system name string from the Stinger SAT_TYPE_ # string that's used describe the signal combo ############################################################################### def convert_sat_type( sat_type ) : if ( sat_type == 'SAT_TYPE_GPS' ) : sname = 'GPS' elif ( sat_type == 'SAT_TYPE_SBAS' ) : sname = 'SBAS' elif ( sat_type == 'SAT_TYPE_GLN' ) : sname = 'GLONASS' elif ( sat_type == 'SAT_TYPE_GAL' ) : sname = 'Galileo' elif ( sat_type == 'SAT_TYPE_QZSS' ) : sname = 'QZSS' elif ( sat_type == 'SAT_TYPE_BDS' ) : sname = 'Beidou' elif ( sat_type == 'SAT_TYPE_IRNSS' ) : sname = 'IRNSS' elif ( sat_type == 'SAT_TYPE_XONA' ) : sname = 'Xona' else : sname = 'Unknown' return( sname ) ############################################################################### # Return the display signal name string from the Stinger SUB_TYPE_ string # that's used to describe the signal combo ############################################################################### def convert_sub_type( sub_type ) : return( sub_type[ 8 : len( sub_type ) ] ) ############################################################################### # The following functions are common to NoPi Stats Generator/Stats # Concat/Stats Display ############################################################################### # Class containing optional command line arguments passed when invoking # NoPiSG_Mult_Day.py or NoPiSC_Process.py. The command line is parsed and # assigned to appropriate objects ############################################################################### class cl_options: def __init__( self ) : self.baseline_file = '' self.start_date = DATA_INVALID self.end_date = DATA_INVALID self.num_days = DATA_INVALID self.reprocess_data = False ############################################################################### # List of metrics evaluated using NoPi_Stats_Checker. # Please update this function to add/delete any metric. # The names are case insensitive within NoPi_Stats_Checker / NoPi_Stats_Display ############################################################################### def get_metrics_list( ) : metric_names = [ 'Mean', \ 'Median', \ 'Std', \ 'MADN', \ 'Skewness', \ 'Three_Sigma', \ 'Rob_Three_Sigma' ] return( metric_names ) ############################################################################### # Return column number for a particular metric. Corresponds to the # format of the stats_combo / acq_stats_combo files ############################################################################### def get_metrics_column( metric ) : if ( metric.upper( ) == 'MEAN' ) : col_num = 6 elif ( metric.upper( ) == 'MEDIAN' ) : col_num = 7 elif ( metric.upper( ) == 'STD' ) : col_num = 8 elif ( metric.upper( ) == 'MADN' ) : col_num = 9 elif ( metric.upper( ) == 'SKEWNESS' ) : col_num = 10 elif( metric.upper( ) == 'THREE_SIGMA' ) : col_num = 11 elif( metric.upper( ) == 'ROB_THREE_SIGMA' ) : col_num = 12 return( col_num ) ############################################################################### # Generate a string of NaNs to append to the concatenated stats file. Follows # the same format as used in each row of a stats_combo file ############################################################################### def stats_file_num_cols( ) : num_cols = 13 return( num_cols ) ############################################################################### # This function parses optional command line arguments. Specific to # multi-day processing scripts. # Supports four options: # -b: File containing names of baselines to be processed # -s: Processing start date # -e: Processing end date # -n: Number of days to process ############################################################################### def parse_multi_day_options( ) : user_opts = cl_options() argv_parsed = 0 for argc in range( 2, len( sys.argv ) ) : tmp_str = sys.argv[argc] # Check for all possible command line options # Parse User specified baseline file. if '-b' in tmp_str : # If there is a white space between -b and , # filename will be the next argv if ( len( tmp_str ) == 2 ) : user_opts.baseline_file = sys.argv[argc + 1] else : user_opts.baseline_file = tmp_str[2:] argv_parsed += 1 # Parse start date if found. Include the either string and not # just the numbers. This simplifies the str2obj_date() function elif '-s' in tmp_str : if ( len( tmp_str[2:] ) == 8 and tmp_str[2:].isdigit() ) : user_opts.start_date = tmp_str[0:] argv_parsed += 1 else : print('Start date format mismatch!!! Expected format: yyyymmdd') # Parse end date if found elif '-e' in tmp_str : if ( len( tmp_str[2:] ) == 8 and tmp_str[2:].isdigit() ) : user_opts.end_date = tmp_str[0:] argv_parsed += 1 else : print('End date format mismatch!!! Expected format: yyyymmdd') # Parse number of days to process elif '-n' in tmp_str : if ( tmp_str[2:].isdigit() and int( tmp_str[2:] ) > 0 ) : user_opts.num_days = int( tmp_str[2:] ) argv_parsed += 1 print('Num days to process: ' + str( user_opts.num_days )) else : print('Number of days cannot be parsed' ) # Ignore existing stats_combo file. Reprocess the entire dataset elif '--clean' in tmp_str : user_opts.reprocess_data = True # Flag a warning for all other command line options elif '-' in tmp_str : print('Invalid command line argument: %s' % tmp_str) # If optional command line arguments were passed and could be # parsed, flag an error and terminate execution if ( len( sys.argv ) > 2 and argv_parsed == 0 ) : print('Command line arguments cannot be parsed.') print('Terminating execution') print('----------------------------------------------------') show_multi_day_help() sys.exit() return( user_opts ) ############################################################################### # Parse either the user specified or default baselines file ############################################################################### def parse_baselines_file( user_opts, legacy_baselines = True ) : baseline_name = [] baseline_dir = [] norm_file_name = [] start_dates = [] end_dates = [] file_path = os.path.abspath(sys.argv[0]) file_dir = os.path.dirname(file_path) top_dir = os.path.dirname(file_dir) fname = top_dir + '/NoPi_Utils/' if ( legacy_baselines ) : fname = fname + 'NoPi_Legacy_Baselines.cfg' else : fname = fname + 'NoPi_Oct2017_Baselines.cfg' # User specified baseline file exists and can be read if ( os.access( user_opts.baseline_file, os.R_OK ) ) : fname = user_opts.baseline_file print('Using as baseline file: %s' % fname) # Else use the default baseline file else : if ( len( user_opts.baseline_file ) > 0 ) : print('Unable to read baseline file: %s' % user_opts.baseline_file) else : print('Baseline file unspecified') print('Using default baseline file: %s' % fname) fin = open( fname, 'rt' ) file_lines = fin.readlines() fin.close() for line in file_lines : tmp_line = shlex.split( line, True ) if ( len( tmp_line ) > 0 ) : [real_name, dir_name, norm_file, start_date, end_date] = line.split( ',' ) baseline_name.append( real_name.strip() ) baseline_dir.append( dir_name.strip() ) norm_file_name.append( norm_file.strip() ) start_dates.append( start_date.strip() ) end_dates.append( end_date.strip() ) if ( len( baseline_name ) == 0 ) : print('Baseline file: %s may be empty or contains only comments' % fname) print('Please verify the file. Terminating execution') sys.exit() return( baseline_name, baseline_dir, norm_file_name, start_dates, end_dates ) ############################################################################### # This function evaluates and returns appropriate start and end date to loop # over. If inadequate or no command line arguments are specified, default dates # are returned. The default start date is 365 days prior to the current date. # # Possible Input Combination:Three bits can have 8 combinations. Will enumerate # them here and match them to each of the four cases in the code below. # -------------------------------------------------------------------------- # -s|-e|-n # 0| 0| 0 : Case 4 # 0| 0| 1 : Case 3 # 0| 1| 0 : Case 4 # 0| 1| 1 : Case 3 # 1| 0| 0 : Start_date is user specified and end date is the current day # 1| 0| 1 : Case 2 # 1| 1| 0 : Case 1 # 1| 1| 1 : Case 1 # -------------------------------------------------------------------------- # ############################################################################### def get_date_range( user_opts ) : today = datetime.date.today() # If user has specified start and/or end dates, convert it from a string # to a datetime object if ( user_opts.start_date != DATA_INVALID ) : start_date = str2obj_date( user_opts.start_date ) if ( user_opts.end_date != DATA_INVALID ) : end_date = str2obj_date( user_opts.end_date ) else : end_date = today # There are a total of eight different combinations possible for the three # command line options: -s, -e and -n. The following lines determine the # appropriate start and end dates. # Case 1: # If start and end dates are specified, ignore the -n option. Use the # specified dates subject to necessary validity checks. if ( user_opts.start_date != DATA_INVALID and user_opts.end_date != DATA_INVALID ) : if ( end_date < start_date ) : print('WARNING!!! End date cannot be prior to start date') print('Start date will be 365 days prior to end date') start_date = end_date - datetime.timedelta( days = 365 ) # Case 2: # If only -s and not -e was parsed, the end date will be the smaller of # the current day or num_days from start date. elif ( user_opts.start_date != DATA_INVALID ) : if( user_opts.num_days != DATA_INVALID and user_opts.num_days <= ( today - start_date ).days ) : end_date = start_date + datetime.timedelta( days = user_opts.num_days ) # Case 3: # If only -n was parsed, the start date will be num_days prior to end # date. End date is either the current day or a user specified end date as # determined above elif( user_opts.num_days != DATA_INVALID ) : start_date = end_date - datetime.timedelta( days = user_opts.num_days ) # Case 4: # Default case to cover all other cases. End_date is determined above else : start_date = end_date - datetime.timedelta( days = 365 ) # Provide user with feedback about processing start and end dates print('Based on command line arguments specified:') print('Start date is: %s' % start_date.strftime("%Y%m%d")) print('End date is: %s' % end_date.strftime("%Y%m%d")) return( start_date, end_date ) ############################################################################### # This function converts a date in string format to a datetime object # # NOTE: The date string can be : # 1. Eight characters long with date specified in yyyymmdd format # 2. Include either -s or -e as its first two characters. The date in yyyymmdd # format is parsed from this 10 character string and converted to a datetime # object ############################################################################### def str2obj_date( date ) : today = datetime.date.today() if ( len( date ) == 10 ) : year = int( date[2:6] ) month = int( date[6:8] ) day = int( date[8: ] ) elif ( len( date ) == 8 ) : year = int( date[0:4] ) month = int( date[4:6] ) day = int( date[6: ] ) else : print('Invalid date') sys.exit() obj_date = datetime.date( year, month, day ) if ( obj_date > today ) : print('Input date: %s is later than today' % date) print('Changing date to today') obj_date = today return( obj_date ) ############################################################################### # Return data path based on current date and baseline. Expects as inputs, both # the baseline name and baseline directory name. Both are specified in a # baseline.cfg file. # # Displays the current date if enabled. This is required only when invoking # Stats Generator. Helps monitor processing progress. ############################################################################### def get_cur_day_path( date, data_path, baseline, dir_name, disp_date ) : cur_date = date.strftime("%Y%m%d") # Display current processing date. Will be displayed only when # running Stats Generator if disp_date : print('----------------------------------------------------------------------' ) print('Processing %s Date: %s' % ( baseline, cur_date )) cur_year = date.year cur_day_path = data_path + str( cur_year ) + '/' if cur_year >= 2022: cur_mon = str(date.month) if date.month < 10 and len(cur_mon) == 1: cur_mon = '0' + cur_mon cur_day_path += cur_mon + '/' cur_day_path += cur_date if ( dir_name == '' ) : cur_day_path += '/' else : cur_day_path += '_' + dir_name + '/' return( cur_date, cur_day_path ) ############################################################################### # Show a few lines of help if command line parameters specific to multi-day # processing scripts cannot be decoded correctly ############################################################################### def show_multi_day_help() : print('Expected command line is:') print('python