import os import numpy as np from datetime import datetime, timedelta, timezone # TC: updated 2026/04/28 (1 line, import zoneinfo) from zoneinfo import ZoneInfo from glob import glob from re import sub from traceback import extract_tb import tarfile from sys import argv from pathlib import Path # Jun modified------------- from spire.step0 import prepare_file from spire.step1 import gather_data from spire.step3 import pair_ro_ref from spire.step4 import cal_exph from configuration_spire import hostname, data_dirs current_path = Path.cwd() working_dir = (current_path / __package__).resolve() # data_dirs = ['/data3/xinjiaz/data_GPSRO/spire_scdr/LEMUR_L0_NAVOBS/', # '/data3/xinjiaz/data_GPSRO/spire_scdr/LEMUR_L0_NAVSOL/', # '/data3/xinjiaz/data_GPSRO/spire_scdr/LEMUR_L0_ATTOBS/', # '/data3/xinjiaz/data_GPSRO/spire_scdr/LEMUR_L0_RO/'] 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 ztypes = ['navObs', 'leoOrb', 'attObs', 'navSol', 'rocObs'] os.chdir(working_dir) working_dir = str(working_dir) root_out = working_dir + f'/L0_extract_{working_leo}' os.makedirs(root_out, exist_ok=True) # Jun modified --------------------------- os.makedirs('s01_podRx3', exist_ok=True) os.makedirs('s01_leoAtt', exist_ok=True) os.makedirs('s01_leoOrb', exist_ok=True) os.makedirs('s01_leoClk', exist_ok=True) os.makedirs('s04_atmPhs', exist_ok=True) os.makedirs('log', exist_ok=True) #Jun modified ---------------------------- logfile = f'log/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 = f'progress_rt_{working_leo}.npz' if os.path.exists(progfile): with np.load(progfile, allow_pickle=True) as fid: tarfiles = fid['tarfiles'] progress = fid['progress'] timetag = fid['timetag'] # TC: updated 2026/04/29 (add 4 lines, add a new variable to indicate processing status) if 'lastaccess' in fid: lastaccess = fid['lastaccess'] else: lastaccess = np.array([datetime(2000,1,1)]) else: tarfiles = np.array([],dtype=' 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.") save_progress() # manual_date = True # set True to manually set dates to process # manual_date = False # set False for cronjob manual_date = len(working_leo)>0 if working_leo.startswith('ucar'): data_dirs[1] = '/STARRO_data2/xinjiaz/data_GPSRO/UCAR/spire/DO1/level1b/leoOrb/' nloop = 0 while True: new_proc = False nloop += 1 if manual_date: # manually assign dates to process process_dates = [datetime(2026, 8, 22) + timedelta(days=x) for x in range(1)] else: #dt_today = datetime.today() #process_dates = [dt_today + timedelta(days=x) for x in range(0,2)] # 2 recent days, today and tomorrow dt_today = datetime.now(timezone.utc) ## safer than datetime.utcnow process_dates = [dt_today + timedelta(days=x) for x in range(-1,1)] # 2 recent days, yesterday and today root_files = [] for data_dir in data_dirs: if 'leoOrb' in data_dir: if working_leo.startswith('ucar'): root_files.append(np.hstack([glob(data_dir + f'{x:%Y/%j}/*_sp3') for x in process_dates])) else: root_files.append(np.hstack([glob(data_dir + f'{x:%Y.%j}/*.SP3') for x in process_dates])) else: root_files.append(np.hstack([glob(data_dir + f'{x:%Y-%m-%d}/*.tar.gz') for x in process_dates])) root_files = np.hstack(root_files) # root_files = np.hstack([np.hstack([glob(data_dir + f'{x:%Y-%m-%d}/*.tar.gz') for x in process_dates]) if 'leoOrb' not in data_dir else np.hstack([glob(data_dir + f'{x:%Y.%j}/*.SP3') for x in process_dates]) for data_dir in data_dirs]) 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) ## sort by the basename, by Xinjia get_base = np.vectorize(os.path.basename) base_tarfiles = get_base(tarfiles) idx = np.argsort(base_tarfiles) tarfiles = tarfiles[idx] print(f"tarfiles: {tarfiles}") progress = progress[idx] timetag = timetag[idx] save_progress() print("after save_progress (1)") try: #ztypes = ['navObs', 'navSol', 'attObs', 'rocObs'] for ztype in ztypes[:-1]: print(f"ztype: {ztype}",flush=True) print(len(progress), len(tarfiles),flush=True) print("progress == -1 count:", np.sum(progress == -1),flush=True) print("example tarfiles:", tarfiles[:5],flush=True) idx = np.where(np.logical_and(progress==-1, [f'{ztype}_' in tt for tt in tarfiles]))[0] # TC: updated 2026/04/28 (add 2 lines, suppress the non-necessary output) if len(idx)==0: continue log_message(f"-- Processing {len(idx)} {ztype} files") update_progress(idx, 0) save_progress() os.makedirs(root_out, exist_ok=True) os.system('rm -r ' + root_out + '/*') os.makedirs(root_out+'/'+ztype, exist_ok=True) for nn, ii in enumerate(idx): if tarfiles[ii].endswith('tar.gz'): with tarfile.open(tarfiles[ii], 'r:gz') as tar: tar.extractall(path=root_out+'/'+ztype) else: os.system(f'ln -s {tarfiles[ii]} {root_out}/{ztype}/') prepare_file(working_leo, ztype) update_progress(idx, 1) save_progress() # TC: updated 2026/04/28 (inserted 7 lines, reprocess failed cases within 3 hours) # TC: updated 2026/04/29 (edited 2 lines and inserted 1 additional line, add 25-minute cool-down for revisit) # TC: updated 2026/05/07 (edited 2 lines [5,6], test run for 3-hour revisit limit and 20-minute cool-down) if hostname.endswith('.star1.nesdis.noaa.gov'): idx1 = np.where(np.logical_and(progress == 2, ['/spire_gnss-ro_' in tt for tt in tarfiles]))[0] else: idx1 = np.where(np.logical_and(progress == 2, ['_RO/' in tt for tt in tarfiles]))[0] if len(idx1) > 0: rotime = np.array([datetime.strptime(sub(r'.*_gnss-ro_(.*)_FM.*', r'\1', tt), '%Y-%m-%dT%H-%M-%S').replace(tzinfo=ZoneInfo('UTC')) for tt in tarfiles[idx1]]) idxt = datetime.now().astimezone(ZoneInfo('UTC')) - rotime < timedelta(hours=3) #idx = datetime.now() - timetag[idx1] > timedelta(minutes=20) idx = datetime.now() - timetag[idx1] > timedelta(minutes=10) ## TC, Xinjia, change to 10 minute update_progress(idx1[np.logical_and(idx,idxt)], -1) update_progress(idx1[np.logical_and(idx,~idxt)], 1) if hostname.endswith('.star1.nesdis.noaa.gov'): idx = np.where(np.logical_and(progress==-1, ['/spire_gnss-ro_' in tt for tt in tarfiles]))[0] else: idx = np.where(np.logical_and(progress==-1, ['_RO/' in tt for tt in tarfiles]))[0] all_leos = np.array([sub(r'.*_(FM...)_.*', r'\1', x) for x in tarfiles[idx]]) print(f"all_leos: {all_leos}") uniq_leos = np.unique(all_leos) print(f"uniq_leos: {uniq_leos}") for leo in uniq_leos: idx_now = idx[all_leos==leo] log_message(f"-- Processing {leo} ({len(idx_now):d}/{np.sum(progress==-1):d})") update_progress(idx_now,0) save_progress() # dt = datetime.strptime(sub(r'.*/(\d{4}-\d\d-\d\d)/.*',r'\1',tarfiles[idx[ii]]),'%Y-%m-%d') os.makedirs(root_out, exist_ok=True) os.makedirs(f't01_RO_{working_leo}', exist_ok=True) os.makedirs(f't01_REF_{working_leo}', exist_ok=True) os.makedirs(f't03_pair_{working_leo}', exist_ok=True) os.system('rm -r ' + root_out + '/*') os.system(f'rm t01_RO_{working_leo}/*.npz') os.system(f'rm t01_REF_{working_leo}/*.npz') os.system(f'rm t03_pair_{working_leo}/*.npz') os.makedirs(root_out + '/' + ztypes[-1], exist_ok=True) print("root_out/ztypes", root_out + '/' + ztypes[-1]) for ii in idx_now: # TC: updated 2026/05/07 (4 lines, preserve directories from original tarfile) tartoken = sub(r'.*/(.*)\.tar\.gz',r'\1', tarfiles[ii]) os.makedirs(root_out + '/' + ztypes[-1] + '/' + tartoken, exist_ok=True) with tarfile.open(tarfiles[ii], 'r:gz') as tar: #tar.extractall(path=root_out+'/'+ztypes[-1], filter="data") #tar.extractall(path=root_out+'/'+ztypes[-1]+'/'+tartoken) ; by Tom tar.extractall(path=root_out+'/'+ztypes[-1]+'/'+tartoken, filter="data") # TC: updated 2026/05/12 (1 line and add 2 lines, set bad profiles as completed and are removed from idx_now) results = gather_data(working_leo) update_progress(idx_now[~results], 1) idx_now = idx_now[results] print("after gather_data") pair_ro_ref(working_leo) print("after pair_ro_ref") # TC: updated 2026/04/28 (3 lines, take results and switch status accordingly) results = cal_exph(working_leo, nProcess, '') print("after cal_exph") update_progress(idx_now[results], 1) update_progress(idx_now[~results], 2) #cal_exph(working_leo, nProcess, sub(r'.*/', r'', tarfiles[ii])) #update_progress(idx_now, 1) print("after update_progress") save_progress() print("after save_progress") 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 # break if not new_proc: if nloop > 1: log_message(f'- No new item, exit program.') break # TC: updated 2026/04/29 (add 2 lines, release processing status) lastaccess[0] = datetime(2000,1,1) save_progress()