# import os from os import chdir, system from pathlib import Path from datetime import datetime, timedelta # from os.path import exists from re import sub import numpy as np # from glob import glob # from tools import get_sat_freq, read_gnss_v2, dt2float, lagrange, inv_quat_rotate # from tools import calculate_tec, calc_bore_sight from traceback import extract_tb from step0 import prepare_file from step1 import gather_data from step4 import calc_tec from sys import argv if __name__ == '__main__': if len(argv) == 3: working_leo = argv[2] nProcess = int(argv[1]) elif len(argv) == 2: working_leo = '' nProcess = int(argv[1]) else: working_leo = '' nProcess = 0 working_dir = Path(__file__).absolute().parent data_dir = Path('/data3/xinjiaz/data_GPSRO/planetiq_scdr/GNOMES_PYXIS_L1A_SpWx') chdir(working_dir) root_out = working_dir / f'L1_extract_{working_leo}' logfile = working_dir / 'log' / f'progress_rt_{working_leo}_{datetime.now():%Y%m}.txt' def log_message(msg): with open(logfile, 'a') as fid: fid.write(f"{datetime.now():%H:%M:%S} {msg}\n") log_message('- checking progress...') progfile = working_dir / f'progress_rt_{working_leo}.npz' if progfile.exists(): with np.load(progfile, allow_pickle=True) as fid: tarfiles = fid['tarfiles'] progress = fid['progress'] timetag = fid['timetag'] else: tarfiles = np.full((0,),Path.cwd()) progress = np.array([],dtype='int8') timetag = np.array([],dtype=datetime) def update_progress(pnum,state): progress[pnum] = state timetag[pnum] = datetime.now() def save_progress(): np.savez(f'progress_rt_{working_leo}.npz', tarfiles = tarfiles, progress = progress, timetag = timetag, ) idx = np.logical_and(progress == 0, datetime.now() - timetag > timedelta(minutes=60)) update_progress(idx, -1) if np.sum(idx) > 0: log_message(f"-- {np.sum(idx)} timeout processes were terminated.") idx = progress == -2 update_progress(idx, -1) if np.sum(idx) > 0: log_message(f"-- {np.sum(idx)} error processes were terminated.") if np.any(progress==0): log_message("-- Still progressing.") else: manual_date = len(working_leo)>0 # set True to manually set dates to process nloop = 0 while True: new_proc = False nloop += 1 if manual_date: # manually assign dates to process process_dates = [datetime(2026, 3, 10) + timedelta(days=x) for x in range(10)] else: dt_today = datetime.today() process_dates = [dt_today + timedelta(days=x) for x in range(-2,2)] # 3 recent days root_files = np.hstack([list(data_dir.glob(f'{x:%Y-%m-%d}/*.tar')) for x in process_dates]) idx = np.array([tarfile in root_files for tarfile in tarfiles], dtype=bool) tarfiles = tarfiles[idx] progress = progress[idx] timetag = timetag[idx] if np.any(idx==False): log_message(f'- {np.sum(idx==False):d} files are removed from the queue.') idx = np.array([root_file not in tarfiles for root_file in root_files], dtype=bool) tarfiles = np.concatenate((tarfiles, root_files[idx])) progress = np.concatenate((progress, np.full((np.sum(idx),),-1,dtype='int8'))) timetag = np.concatenate((timetag, np.full((np.sum(idx),),datetime.now(),dtype=datetime))) if np.any(idx): log_message(f'- {np.sum(idx == True):d} files are added to the queue.') idx = np.argsort(tarfiles) tarfiles = tarfiles[idx] progress = progress[idx] timetag = timetag[idx] save_progress() try: idx = np.where(progress == -1)[0] leos = np.unique([sub(r'.*_(....)\.tar', r'\1', tarfile.name) for tarfile in tarfiles[idx]]) dts = np.unique([datetime.strptime(tarfile.parent.name, '%Y-%m-%d') for tarfile in tarfiles[idx]]) for dt in dts: for leo in leos: idx_now = idx[[leo in tarfile.name and tarfile.parent.name==f'{dt:%Y-%m-%d}' for tarfile in tarfiles[idx]]] update_progress(idx_now, 0) save_progress() log_message(f"-- Processing {leo}-{dt:%Y/%m/%d} ({len(idx_now):d}/{len(idx):d})") root_out.mkdir(exist_ok=True) system(f'rm -r {root_out}/*') for ii in range(len(idx_now)): prepare_file(tarfiles[idx_now[ii]], working_leo) gather_data(working_leo) calc_tec(leo, dt, nProcess) update_progress(idx_now, 1) if len(idx)>0: save_progress() new_proc = True log_message(f'- Progress completed.') except Exception as e: log_message('- Unexpected error occurs.') log_message(f"-- {type(e).__name__} >> {str(e)}") for tb in extract_tb(e.__traceback__): log_message(f"-- line {tb.lineno} of {tb.filename}") update_progress(progress == 0, -2) save_progress() new_proc = True if not new_proc: if nloop > 1: log_message(f'- No new item, exit program.') break # extract_orb() # extract_att() # gather_crx() # leos_ref = ['GN04', 'GN05'] # ants_ref = [0, 1] # gnsses_ref = ['C06', 'C07', 'C08', 'C09', 'C10', 'C11', 'C12', 'C13', 'C14', 'C16', 'C19', 'C20', 'C21', 'C22', # 'C23', 'C24', 'C25', 'C26', 'C27', 'C28', 'C29', 'C30', 'C32', 'C33', 'C34', 'C35', 'C36', 'C37', # 'C38', 'C39', 'C40', 'C41', 'C42', 'C43', 'C44', 'C45', 'C47', 'C48', 'C49', 'E02', 'E03', 'E04', # 'E05', 'E06', 'E07', 'E08', 'E09', 'E10', 'E11', 'E12', 'E13', 'E14', 'E15', 'E16', 'E18', 'E19', # 'E21', 'E23', 'E24', 'E25', 'E26', 'E27', 'E29', 'E30', 'E31', 'E33', 'E34', 'E36', 'G01', 'G02', # 'G03', 'G04', 'G05', 'G06', 'G07', 'G08', 'G09', 'G10', 'G11', 'G12', 'G13', 'G14', 'G15', 'G16', # 'G17', 'G18', 'G19', 'G20', 'G21', 'G22', 'G23', 'G24', 'G25', 'G26', 'G27', 'G28', 'G29', 'G30', # 'G31', 'G32', 'J02', 'J03', 'J04', 'R02', 'R03', 'R04', 'R05', 'R06', 'R07', 'R08', 'R09', 'R11', # 'R12', 'R14', 'R15', 'R16', 'R17', 'R18', 'R19', 'R20', 'R21', 'R22', 'R24', 'R25', 'R26', 'R27', # 'R28'] # leos_nd, ants_nd, gnss_nd = np.meshgrid(leos_ref,ants_ref,gnsses_ref,indexing='ij') # leos_nd = leos_nd.flatten() # ants_nd = ants_nd.flatten() # gnss_nd = gnss_nd.flatten() # allargs = zip(leos_nd,ants_nd,gnss_nd) # import multiprocessing as mp # with mp.Pool(processes=24) as pool: # pool.map(calc_bore_sight, allargs) # leos_ref = ['GN04', 'GN05'] # ants_ref = [0, 1] # dts_ref = [datetime(2026,1,1) + timedelta(days=x) for x in range(31)] # leos_nd, ants_nd, dts_nd = np.meshgrid(leos_ref,ants_ref,dts_ref,indexing='ij') # leos_nd = leos_nd.flatten() # ants_nd = ants_nd.flatten() # dts_nd = dts_nd.flatten() # idx = np.full(ants_nd.shape, True) # for nn,leo,dt,ant in zip(range(len(dts_nd)),leos_nd,dts_nd,ants_nd): # outfile = f'./p04_teczdcb/teczdcb_{leo}.{dt:%Y-%m-%d}.A{ant:02d}.npz' # if os.path.exists(outfile) and datetime(1970,1,1)+timedelta(seconds=os.path.getmtime(outfile))>datetime(2026,4,10): # idx[nn] = False # leos_nd = leos_nd[idx] # ants_nd = ants_nd[idx] # dts_nd = dts_nd[idx] # allargs = zip(leos_nd, dts_nd, ants_nd) # import multiprocessing as mp # with mp.Pool(processes=10) as pool: # pool.map(calculate_dcb, allargs) # for dd in range(31): # reffiles = glob(f'./p01_podPair/{datetime(2026,1,1) + timedelta(days=dd):%Y-%m-%d}/*.npz') # for reffile in reffiles: # try: # sec_ref_0, tec, coszn, meps = calculate_tec(reffile) # except Exception as e: # print(f"{datetime.now():%H:%M:%S} - Error detected for {os.path.basename(reffile)}", flush=True) # print(f"{datetime.now():%H:%M:%S} -- {type(e).__name__} >> {str(e)}", flush=True) # for tb in extract_tb(e.__traceback__): # print(f"{datetime.now():%H:%M:%S} -- line {tb.lineno} of {tb.filename}", flush=True)