import sqlite3 import json from datetime import datetime import os import numpy as np import unittest import tempfile curr_path = os.path.dirname(os.path.realpath(__file__)) db_filename = curr_path + '/OutputResults/all_data.db' def get_db(write=False): if write: db = sqlite3.connect(db_filename) else: db = sqlite3.connect('file:%s?mode=ro'%db_filename,uri=True) db.row_factory = sqlite3.Row if write: create_tables(db.cursor()) db.commit() # Save updates from create_tables() return db def read_all_runs_pos_summary(c): c.execute('select * from ALL_RUNS_POS_SUMMARY') summary = {} for row in c.fetchall(): summary[(row['config_ID'],row['run_type'],row['seg_ID'],row['pos_ID'])] = 1 return summary def read_all_runs_nopi_summary(c): c.execute('select * from ALL_RUNS_NOPI_SUMMARY') summary = {} for row in c.fetchall(): summary[row['config_ID']] = 1 return summary def read_latest_runs_summary(c): c.execute('select * from LATEST_RUNS_SUMMARY') summary = {} for row in c.fetchall(): summary[row['config_ID']] = (row['run_date'],row['run_num'],row['rx_num']) return summary def read_configs(c,key_is_ID=True): c.execute('select * from CONFIGS') all_configs = {} for row in c.fetchall(): if key_is_ID: all_configs[row['config_ID']] = (row['filename'],row['short_desc']) else: all_configs[(row['filename'],row['short_desc'])] = row['config_ID'] return all_configs def read_rx_descs(c,key_is_ID=True): c.execute('select * from RX_DESCS') all_rx_descs = {} for row in c.fetchall(): if key_is_ID: all_rx_descs[row['rx_desc_ID']] = row['short_desc'] else: all_rx_descs[row['short_desc']] = row['rx_desc_ID'] return all_rx_descs def read_segs(c,key_is_ID=True): c.execute('select * from SEGS') all_segs = {} for row in c.fetchall(): if key_is_ID: all_segs[row['seg_ID']] = row['desc'] else: all_segs[row['desc']] = row['seg_ID'] return all_segs def read_signals(c,key_is_ID=True): c.execute('select * from SIGNALS') all_signals = {} for row in c.fetchall(): if key_is_ID: all_signals[row['signal_ID']] = row['desc'] else: all_signals[row['desc']] = row['signal_ID'] return all_signals def read_pos_types(c,key_is_ID=True): c.execute('select * from POS_TYPES') all_pos_types = {} for row in c.fetchall(): if key_is_ID: all_pos_types[row['pos_ID']] = row['desc'] else: all_pos_types[row['desc']] = row['pos_ID'] return all_pos_types def read_partials(c,key_is_ID=True): c.execute('select * from PARTIALS') all_partials = {} for row in c.fetchall(): if key_is_ID: all_partials[row['partial_ID']] = row['desc'] else: all_partials[row['desc']] = row['partial_ID'] return all_partials class GenericTable: table_name = "" entries = "" __isfrozen = False def __init__(self): self._all_vars = [] self._all_lines = [] for line in self.entries.split('\n'): var_name = line.split()[0] if var_name.find('(') > 0: # ignore constraints like "UNIQUE(var1,var2)" continue vars(self)[var_name] = None self._all_vars.append(var_name) self._all_lines.append(line) self._freeze() # don't allow new vars def __setattr__(self, key, value): if self.__isfrozen and not hasattr(self, key): raise TypeError( "%r is a frozen class" % self ) object.__setattr__(self, key, value) def _freeze(self): self.__isfrozen = True def insert_into(self,c): c.execute("INSERT INTO %s (%s) VALUES (%s)"% (self.table_name, ','.join(self._all_vars), ','.join('?'*len(self._all_vars))), [vars(self)[k] for k in self._all_vars]) def replace_into(self,c): c.execute("REPLACE INTO %s (%s) VALUES (%s)"% (self.table_name, ','.join(self._all_vars), ','.join('?'*len(self._all_vars))), [vars(self)[k] for k in self._all_vars]) def create_table(self,c): c.execute("CREATE TABLE IF NOT EXISTS %s (%s);"% (self.table_name, self.entries)) # Add any missing columns columns = [i[1] for i in c.execute('PRAGMA table_info(%s)'% self.table_name)] for col_name,col_line in zip(self._all_vars,self._all_lines): if not col_name in columns: print("Adding column: %s"%col_line) c.execute("ALTER TABLE %s ADD %s"% (self.table_name,col_line) ) class ConfigsTable(GenericTable): # Unique scenarios + XML file (e.g., 2019-09-12-Tokyo-Drive01.xml) table_name = "CONFIGS" entries = """config_ID integer UNIQUE, filename text UNIQUE, short_desc text""" # May want to add default for new entries - see NopiResultsTable class RxDescsTable(GenericTable): table_name = "RX_DESCS" # Unique receiver descriptions (e.g., "Top CVS BD930") entries = """rx_desc_ID integer UNIQUE, short_desc text UNIQUE""" # May want to add default for new entries - see NopiResultsTable class SegsTable(GenericTable): # Unique position segments (e.g., "RuralOpen") table_name = "SEGS" entries = """seg_ID integer UNIQUE, desc text UNIQUE""" # May want to add default for new entries - see NopiResultsTable class PosTypesTable(GenericTable): # Unique position types (e.g., "RTK Float") table_name = "POS_TYPES" entries = """pos_ID integer UNIQUE, desc text UNIQUE""" # May want to add default for new entries - see NopiResultsTable class PartialsTable(GenericTable): # Unique partial types (e.g., "Partial") table_name = "PARTIALS" entries = """partial_ID integer UNIQUE, desc text UNIQUE""" # May want to add default for new entries - see NopiResultsTable def create_table(self,c): super().create_table(c) c.execute("""REPLACE INTO PARTIALS VALUES (0,'Full')""") c.execute("""REPLACE INTO PARTIALS VALUES (1,'Partial')""") class SignalsTable(GenericTable): # Unique NoPi signals (e.g., "GPS_L5") table_name = "SIGNALS" entries = """signal_ID integer UNIQUE, desc text UNIQUE""" # May want to add default for new entries - see NopiResultsTable class NopiResultsTable(GenericTable): # NoPi results. Note we have multiple results for each run_num+rx_num. table_name = "NOPI_RESULTS" entries = """run_date timestamp, run_num integer, rx_num integer, rx_desc_ID integer, config_ID integer, signal_ID integer, num_meas integer, carr_std real, code_std real, partial_ID integer DEFAULT 0 NOT NULL""" # partial_ID added later class AgPosResultsTable(GenericTable): # Ag position results table_name = "AG_POS_RESULTS" entries = """run_date timestamp, run_num integer, rx_num integer, rx_desc_ID integer, config_ID integer, custom_fw integer, pos_ID integer, seg_ID integer, num_pos integer, above_6 real, above_10 real, above_50 real, above_80 real, above_2sigma real, partial_ID integer""" # May want to add default for new entries - see NopiResultsTable normal_pos_results_entries = """run_date timestamp, run_num integer, rx_num integer, rx_desc_ID integer, config_ID integer, custom_fw integer, pos_ID integer, seg_ID integer, num_pos integer, cdf_2d_50 real, cdf_2d_68 real, cdf_2d_95 real, cdf_2d_99 real, cdf_2d_100 real, cdf_3d_50 real, cdf_3d_68 real, cdf_3d_95 real, cdf_3d_99 real, cdf_3d_100 real, partial_ID integer NOT NULL DEFAULT 0, cdf_1d_50 real, cdf_1d_68 real, cdf_1d_95 real, cdf_1d_99 real, cdf_1d_100 real""" # May want to add default for new entries - see partial_ID class SbasPosResultsTable(GenericTable): table_name = "SBAS_POS_RESULTS" entries = normal_pos_results_entries class RtkPosResultsTable(GenericTable): table_name = "RTK_POS_RESULTS" entries = normal_pos_results_entries class DgnssPosResultsTable(GenericTable): table_name = "DGNSS_POS_RESULTS" entries = normal_pos_results_entries class RtxPosResultsTable(GenericTable): table_name = "RTX_POS_RESULTS" entries = normal_pos_results_entries dRUN_TABLE = {} dRUN_TYPE = {} dRUN_TYPE['SBAS'] = 0; dRUN_TABLE['SBAS']=SbasPosResultsTable dRUN_TYPE['RTK'] = 1; dRUN_TABLE['RTK']=RtkPosResultsTable dRUN_TYPE['DGNSS'] = 2; dRUN_TABLE['DGNSS']=DgnssPosResultsTable dRUN_TYPE['AG'] = 3; dRUN_TABLE['AG']=AgPosResultsTable dRUN_TYPE['RTX'] = 4; dRUN_TABLE['RTX']=RtxPosResultsTable class ErrLogsTable(GenericTable): # Error log info for each run+receiver. table_name = "ERR_LOGS" entries = """run_date timestamp, run_num integer, rx_num integer, rx_desc_ID integer, config_ID integer, n_logs integer, err_str text, warn_str text, UNIQUE(run_num,rx_num)""" # May want to add default for new entries - see NopiResultsTable class AllRunsPosSummaryTable(GenericTable): # Summary of segments/positions all position runs. Speeds up web UI. table_name = "ALL_RUNS_POS_SUMMARY" entries = """config_ID integer, run_type integer, seg_ID integer, pos_ID integer, UNIQUE(config_ID,run_type,seg_ID,pos_ID)""" # May want to add default for new entries - see NopiResultsTable class LatestRunsSummaryTable(GenericTable): # Summary of latest data. Speeds up web UI. table_name = "LATEST_RUNS_SUMMARY" entries = """config_ID integer UNIQUE, run_date timestamp, run_num integer, rx_num integer""" # May want to add default for new entries - see NopiResultsTable class AllRunsNopiSummaryTable(GenericTable): # Summary of all NoPi runs. Speeds up web UI. table_name = "ALL_RUNS_NOPI_SUMMARY" entries = """config_ID integer UNIQUE""" # May want to add default for new entries - see NopiResultsTable def create_tables(c): ConfigsTable().create_table(c) RxDescsTable().create_table(c) SegsTable().create_table(c) PosTypesTable().create_table(c) PartialsTable().create_table(c) SignalsTable().create_table(c) NopiResultsTable().create_table(c) # Position results. Note we have multiple results for each run_num+rx_num. for run_name in dRUN_TYPE.keys(): dRUN_TABLE[run_name]().create_table(c) ErrLogsTable().create_table(c) AllRunsPosSummaryTable().create_table(c) LatestRunsSummaryTable().create_table(c) AllRunsNopiSummaryTable().create_table(c) def remove_deleted_filesystem_runs( c, all_rx_and_run ): """ c = sqlite cursor all_rx_and_run = all RX # + run # combinations in ResultsQueue/*.xml (e.g., [(0,1000), (0,1001), ...] ) Sometimes a user will delete runs because they have bogus data. We need to detect this and remove the data from the following tables: NOPI_RESULTS, ERR_LOGS, *_POS_RESULTS, LATEST_RUNS_SUMMARY We detect deleted data by looking at what is in the FILESYSTEM_RUNS table and comparing it to all our results tables. """ # Create a temporary table for speed of operations below c.execute("""CREATE TEMPORARY TABLE IF NOT EXISTS FILESYSTEM_RUNS ( rx_num integer, run_num integer, UNIQUE(rx_num,run_num) ); """) c.execute('DELETE FROM FILESYSTEM_RUNS;') c.executemany('REPLACE INTO FILESYSTEM_RUNS VALUES(?,?);',all_rx_and_run) all_table_names = ['NOPI_RESULTS','ERR_LOGS'] for run_name in dRUN_TYPE.keys(): all_table_names.append( "%s_POS_RESULTS"%run_name ) for table_name in all_table_names: to_remove = [] c.execute('SELECT rx_num,run_num from %s EXCEPT select rx_num,run_num from FILESYSTEM_RUNS'%table_name) for row in c.fetchall(): to_remove.append( (row['rx_num'],row['run_num']) ) if len(to_remove) > 0: print(' remove',table_name,to_remove) c.executemany('delete from %s where rx_num=? and run_num=?;'%table_name, to_remove) # If we remove something from LATEST_RUNS_SUMMARY, we must try to # replace it. c.execute('SELECT rx_num,run_num from LATEST_RUNS_SUMMARY EXCEPT select rx_num,run_num from FILESYSTEM_RUNS') to_remove = [] for row in c.fetchall(): to_remove.append( (row['rx_num'],row['run_num']) ) to_add = [] for rx_num, run_num in to_remove: c.execute('SELECT config_ID from LATEST_RUNS_SUMMARY where rx_num=%d and run_num=%d'%(rx_num,run_num)) row = c.fetchone() if row is None: continue config_ID = row['config_ID'] c.execute('select * from ERR_LOGS where config_ID=%d ORDER BY run_num DESC LIMIT 1;'%config_ID) row = c.fetchone() if row is None: continue to_add.append( (row['config_ID'], row['run_date'], row['run_num'], row['rx_num'] ) ) if len(to_remove) > 0: print(' latest_runs remove',to_remove) c.executemany('delete from LATEST_RUNS_SUMMARY where rx_num=? and run_num=?;', to_remove) if len(to_add) > 0: print(' latest_runs add',to_add) c.executemany('insert into LATEST_RUNS_SUMMARY (config_ID,run_date,run_num,rx_num) VALUES (?,?,?,?);', to_add) def insert_errlog(c,run_date,run_num,rx_num,rx_ID,config_ID,data_err_log): tbl = ErrLogsTable() tbl.run_date=run_date tbl.run_num=run_num tbl.rx_num=rx_num tbl.rx_desc_ID=rx_ID tbl.config_ID=config_ID tbl.n_logs=data_err_log['total'] tbl.err_str = json.dumps(data_err_log['errors']) tbl.warn_str = json.dumps(data_err_log['warnings']) tbl.replace_into(c) def insert_nopi(c, all_runs_nopi_summary, all_signals, run_date, run_num, rx_num,rx_ID, config_ID, data_nopi, partial_ID): if len(data_nopi) == 0: return if not config_ID in all_runs_nopi_summary: all_runs_nopi_summary[config_ID] = 1 tbl = AllRunsNopiSummaryTable() tbl.config_ID = config_ID tbl.replace_into(c) for signal_desc,info in data_nopi.items(): if not signal_desc in all_signals: signal_ID = len(all_signals) all_signals[signal_desc] = signal_ID tbl = SignalsTable() tbl.signal_ID = signal_ID tbl.desc = signal_desc tbl.replace_into(c) signal_ID = all_signals[signal_desc] num_meas = info['num'] carr_std = info['car_std'] code_std = info['code_std'] tbl = NopiResultsTable() tbl.run_date=run_date tbl.run_num=run_num tbl.rx_num=rx_num tbl.rx_desc_ID=rx_ID tbl.config_ID=config_ID tbl.signal_ID=signal_ID tbl.num_meas=num_meas tbl.carr_std=carr_std tbl.code_std=code_std tbl.partial_ID=partial_ID tbl.insert_into(c) def insert_stats(c, all_runs_pos_summary, all_segs, all_pos_types, data_stats, run_date, run_num, rx_num, rx_ID, config_ID, run_type, custom_fw, partial_ID): if data_stats is None: return table_name = run_type + '_POS_RESULTS' for seg_name,seg_info in data_stats['seg'].items(): if not seg_name in all_segs: all_segs[seg_name] = len(all_segs) tbl = SegsTable() tbl.seg_ID = all_segs[seg_name] tbl.desc = seg_name tbl.replace_into(c) seg_ID = all_segs[seg_name] if seg_info is None: # The unit completely failed. Insert NULL data. tbl = dRUN_TABLE[run_type]() tbl.run_date = run_date tbl.run_num = run_num tbl.rx_num = rx_num tbl.rx_desc_ID = rx_ID tbl.config_ID = config_ID tbl.custom_fw = custom_fw tbl.pos_ID = all_pos_types['All'] tbl.seg_ID = seg_ID tbl.partial_ID = False tbl.insert_into(c) break for pos_type,pos_info in seg_info.items(): key = pos_type if not key in all_pos_types: all_pos_types[key] = len(all_pos_types) tbl = PosTypesTable() tbl.pos_ID = all_pos_types[key] tbl.desc = key tbl.replace_into(c) pos_ID = all_pos_types[key] key = (config_ID,dRUN_TYPE[run_type],seg_ID,pos_ID) if not key in all_runs_pos_summary: all_runs_pos_summary[key] = 1 tbl = AllRunsPosSummaryTable() tbl.config_ID = config_ID tbl.run_type = dRUN_TYPE[run_type] tbl.seg_ID = seg_ID tbl.pos_ID = pos_ID tbl.replace_into(c) num_pos = pos_info['len'] cdf_1d_50 = None cdf_2d_50 = None cdf_3d_50 = None cdf_1d_68 = None cdf_2d_68 = None cdf_3d_68 = None cdf_1d_95 = None cdf_2d_95 = None cdf_3d_95 = None cdf_1d_99 = None cdf_2d_99 = None cdf_3d_99 = None cdf_1d_100 = None cdf_2d_100 = None cdf_3d_100 = None for n in range(len(pos_info['cdf_vals'])): if pos_info['cdf_vals'][n] == 50: cdf_1d_50 = pos_info['1d'][n] cdf_2d_50 = pos_info['2d'][n] cdf_3d_50 = pos_info['3d'][n] elif pos_info['cdf_vals'][n] == 68: cdf_1d_68 = pos_info['1d'][n] cdf_2d_68 = pos_info['2d'][n] cdf_3d_68 = pos_info['3d'][n] elif pos_info['cdf_vals'][n] == 95: cdf_1d_95 = pos_info['1d'][n] cdf_2d_95 = pos_info['2d'][n] cdf_3d_95 = pos_info['3d'][n] elif pos_info['cdf_vals'][n] == 99: cdf_1d_99 = pos_info['1d'][n] cdf_2d_99 = pos_info['2d'][n] cdf_3d_99 = pos_info['3d'][n] elif pos_info['cdf_vals'][n] == 100: cdf_1d_100 = pos_info['1d'][n] cdf_2d_100 = pos_info['2d'][n] cdf_3d_100 = pos_info['3d'][n] tbl = dRUN_TABLE[run_type]() tbl.run_date = run_date tbl.run_num = run_num tbl.rx_num = rx_num tbl.rx_desc_ID = rx_ID tbl.config_ID = config_ID tbl.custom_fw = custom_fw tbl.pos_ID = pos_ID tbl.seg_ID = seg_ID tbl.num_pos = num_pos tbl.cdf_2d_50 = cdf_2d_50 tbl.cdf_2d_68 = cdf_2d_68 tbl.cdf_2d_95 = cdf_2d_95 tbl.cdf_2d_99 = cdf_2d_99 tbl.cdf_2d_100 = cdf_2d_100 tbl.cdf_3d_50 = cdf_3d_50 tbl.cdf_3d_68 = cdf_3d_68 tbl.cdf_3d_95 = cdf_3d_95 tbl.cdf_3d_99 = cdf_3d_99 tbl.cdf_3d_100 = cdf_3d_100 tbl.partial_ID = partial_ID tbl.cdf_1d_50 = cdf_1d_50 tbl.cdf_1d_68 = cdf_1d_68 tbl.cdf_1d_95 = cdf_1d_95 tbl.cdf_1d_99 = cdf_1d_99 tbl.cdf_1d_100 = cdf_1d_100 tbl.insert_into(c) def insert_ag_stats(c, all_runs_pos_summary, all_segs, all_pos_types, data_stats, run_date, run_num, rx_num, rx_ID, config_ID, run_type, custom_fw, partial_ID): if data_stats is None: return table_name = 'AG_POS_RESULTS' # run_type = 'AG' for seg_name,seg_info in data_stats['seg'].items(): if not seg_name in all_segs: all_segs[seg_name] = len(all_segs) tbl = SegsTable() tbl.seg_ID = all_segs[seg_name] tbl.desc = seg_name tbl.replace_into(c) seg_ID = all_segs[seg_name] if seg_info is None: # The unit completely failed. Insert NULL data. tbl = AgPosResultsTable() tbl.run_date = run_date tbl.run_num = run_num tbl.rx_num = rx_num tbl.rx_desc_ID = rx_ID tbl.config_ID = config_ID tbl.custom_fw = custom_fw tbl.pos_ID = all_pos_types['All'] tbl.seg_ID = seg_ID tbl.insert_into(c) break for pos_type,pos_info in seg_info.items(): key = pos_type if not key in all_pos_types: all_pos_types[key] = len(all_pos_types) tbl = PosTypesTable() tbl.pos_ID = all_pos_types[key] tbl.desc = key tbl.replace_into(c) pos_ID = all_pos_types[key] key = (config_ID,dRUN_TYPE[run_type],seg_ID,pos_ID) if not key in all_runs_pos_summary: all_runs_pos_summary[key] = 1 tbl = AllRunsPosSummaryTable() tbl.config_ID = config_ID tbl.run_type = dRUN_TYPE[run_type] tbl.seg_ID = seg_ID tbl.pos_ID = pos_ID tbl.replace_into(c) num_pos = pos_info['len'] above_6 = None above_10 = None above_50 = None above_80 = None above_2sigma = None for n in range(len(pos_info['threshold'])): if pos_info['threshold'][n] == 6: above_6 = pos_info['2d_above_ctr'][n] elif pos_info['threshold'][n] == 10: above_10 = pos_info['2d_above_ctr'][n] elif pos_info['threshold'][n] == 50: above_50 = pos_info['2d_above_ctr'][n] elif pos_info['threshold'][n] == 80: above_80 = pos_info['2d_above_ctr'][n] elif pos_info['threshold'][n] == -1: above_2sigma = pos_info['2d_above_ctr'][n] tbl = AgPosResultsTable() tbl.run_date = run_date tbl.run_num = run_num tbl.rx_num = rx_num tbl.rx_desc_ID = rx_ID tbl.config_ID = config_ID tbl.custom_fw = custom_fw tbl.pos_ID = pos_ID tbl.seg_ID = seg_ID tbl.num_pos = num_pos tbl.above_6 = above_6 tbl.above_10 = above_10 tbl.above_50 = above_50 tbl.above_80 = above_80 tbl.above_2sigma = above_2sigma tbl.partial_ID = partial_ID tbl.insert_into(c) def add_data_to_db(all_new_data,have_existing_data=True): if all_new_data == [None]: return with get_db(True) as db: c = db.cursor() if have_existing_data: # Delete old data so we don't get multiple copies of what we're putting in. # Only needed for tables without UNIQUE (NOPI_RESULTS, *_POS_RESULTS) for data in all_new_data: run_num = int(data['file_base'].split('-')[1]) rx_num = int(data['file_base'].split('-')[0][2:]) c.execute("delete from NOPI_RESULTS where run_num=? and rx_num=?", (run_num,rx_num)) for run_name in dRUN_TYPE.keys(): table_name = "%s_POS_RESULTS"%run_name c.execute("delete from %s where run_num=? and rx_num=?"%table_name, (run_num,rx_num)) # Add new data all_configs = read_configs(c,key_is_ID=False) all_rxs = read_rx_descs(c,key_is_ID=False) all_segs = read_segs(c,key_is_ID=False) all_pos_types = read_pos_types(c,key_is_ID=False) all_signals = read_signals(c,key_is_ID=False) all_runs_pos_summary = read_all_runs_pos_summary(c) all_runs_nopi_summary = read_all_runs_nopi_summary(c) latest_runs_summary = read_latest_runs_summary(c) for data in all_new_data: run_num = int(data['file_base'].split('-')[1]) rx_num = int(data['file_base'].split('-')[0][2:]) run_date = datetime.strptime(data['date'],'%Y-%m-%d %H:%M:%S') config_desc = data['config']['desc'] config_filename = data['config']['config_filename'].split('/')[-1] if data['do_RTX']: run_type = 'RTX' elif data['do_RTK']: run_type = 'RTK' else: run_type = 'SBAS' custom_fw = 0 if 'custom_fw' in data and data['custom_fw'] == 1: custom_fw = 1 partial_ID = int(data['partial_run']) key = (config_filename, config_desc) if not key in all_configs: # people sometimes change the config_desc. In this # case we don't want a new scenario. We just want to # replace the old desc. idx = np.where([y[0]==config_filename for y in all_configs])[0] if len(idx) > 0: old_key = list(all_configs.keys())[idx[0]] config_ID = all_configs[old_key] del all_configs[old_key] else: config_ID = len(all_configs) all_configs[key] = config_ID tbl = ConfigsTable() tbl.config_ID = all_configs[key] tbl.filename = config_filename tbl.short_desc = config_desc tbl.replace_into(c) config_ID = all_configs[key] key = data['unit_desc'] if not key in all_rxs: all_rxs[key] = len(all_rxs) tbl = RxDescsTable() tbl.rx_desc_ID = all_rxs[key] tbl.short_desc = key tbl.replace_into(c) rx_ID = all_rxs[key] try: latest_date = datetime.strptime(latest_runs_summary[config_ID][0], '%Y-%m-%d %H:%M:%S') except: latest_date = datetime(1900,1,1) if run_date > latest_date and partial_ID == 0: latest_runs_summary[config_ID] = (run_date,run_num,rx_num) tbl = LatestRunsSummaryTable() tbl.config_ID = config_ID tbl.run_date = run_date tbl.run_num = run_num tbl.rx_num = rx_num tbl.replace_into(c) if len(data['stats']) > 0: insert_stats(c, all_runs_pos_summary, all_segs, all_pos_types, data['stats'], run_date, run_num, rx_num, rx_ID, config_ID, run_type, custom_fw, partial_ID) if 'dgnss_stats' in data: insert_stats(c, all_runs_pos_summary, all_segs, all_pos_types, data['dgnss_stats'], run_date, run_num, rx_num, rx_ID, config_ID, 'DGNSS', custom_fw, partial_ID) if 'ag_stats' in data: insert_ag_stats(c, all_runs_pos_summary, all_segs, all_pos_types, data['ag_stats'], run_date, run_num, rx_num, rx_ID, config_ID, 'AG', custom_fw, partial_ID) if 'err_log' in data: insert_errlog( c, run_date, run_num, rx_num, rx_ID, config_ID, data['err_log']) if 'nopi' in data['stats']: insert_nopi( c, all_runs_nopi_summary, all_signals, run_date, run_num, rx_num, rx_ID, config_ID, data['stats']['nopi'],partial_ID) def convert_old_format_to_new_if_needed(): # obsolete pass class Testing(unittest.TestCase): # The tests below mainly check that _freeze() is not violated # in the insert_*() functions and that there are no SQL typos. def test_errlog(self): with get_db(True) as db: db_c = db.cursor() insert_errlog(db_c, datetime(2023,1,1), 0, 1, 2, 0, {'total':3,'errors':['0x1'],'warnings':[]}) db_c.execute('select * from ERR_LOGS') row = db_c.fetchall() self.assertEqual(list(row[0]), ['2023-01-01 00:00:00', 0, 1, 2, 0, 3, '["0x1"]', '[]']) def test_nopi(self): with get_db(True) as db: db_c = db.cursor() summary = {} all_signals = {} data_nopi = {'GPS_L1CA':{'num':1,'car_std':0.1,'code_std':.2}} insert_nopi(db_c, summary, all_signals, datetime(2023,1,1), 1, 2, 3, 4, data_nopi, True) db_c.execute('select * from NOPI_RESULTS') row = db_c.fetchall() self.assertEqual(list(row[0]), ['2023-01-01 00:00:00', 1, 2, 3, 4, 0, 1, 0.1, 0.2, 1]) def test_stats(self): with get_db(True) as db: db_c = db.cursor() summary = {} all_segs = {} all_pos_types = {} data_stats = {'seg':{ 'All':{'All':{'len':5, 'cdf_vals':[50,68,95,99,100], '1d':[1,2,3,4,5], '2d':[6,7,8,9,10], '3d':[11,12,13,14,15] }}, 'Test':None}} insert_stats( db_c, summary, all_segs, all_pos_types, data_stats, datetime(2023,1,1), 1, 2, 3, 4, 'RTK', True, False ) db_c.execute('select * from RTK_POS_RESULTS') row = db_c.fetchall() self.assertEqual(list(row[0]), ['2023-01-01 00:00:00', 1, 2, 3, 4, 1, 0, 0, 5, 6.0, 7.0, 8.0, 9.0, 10.0, 11.0, 12.0, 13.0, 14.0, 15.0, 0, 1.0, 2.0, 3.0, 4.0, 5.0]) self.assertEqual(list(row[1]), ['2023-01-01 00:00:00', 1, 2, 3, 4, 1, 0, 1, None, None, None, None, None, None, None, None, None, None, None, 0, None, None, None, None, None]) def test_ag_stats(self): with get_db(True) as db: db_c = db.cursor() summary = {} all_segs = {} all_pos_types = {} data_stats = {'seg':{ 'All':{'All':{'len':5, 'threshold':[6,10,50,80], '2d_above_ctr':[50,68,95,99,100] }}, 'Test':None}} insert_ag_stats( db_c, summary, all_segs, all_pos_types, data_stats, datetime(2023,1,1), 1, 2, 3, 4, 'RTK', True, True ) db_c.execute('select * from AG_POS_RESULTS') row = db_c.fetchall() self.assertEqual(list(row[0]), ['2023-01-01 00:00:00', 1, 2, 3, 4, 1, 0, 0, 5, 50.0, 68.0, 95.0, 99.0, None, 1]) self.assertEqual(list(row[1]), ['2023-01-01 00:00:00', 1, 2, 3, 4, 1, 0, 1, None, None, None, None, None, None, None]) if __name__ == '__main__': temp_file = tempfile.NamedTemporaryFile(suffix='.db') db_filename = temp_file.name print("Temporary DB:",db_filename) unittest.main(exit=False)