import os from pathlib import Path import numpy as np from datetime import datetime, timedelta, timezone from glob import glob from re import sub from traceback import extract_tb from sys import argv from cosmic2.step1 import gather_data from cosmic2.step3 import pair_ro_ref from cosmic2.step4 import cal_exph from configuration_cosmic2 import hostname current_path = Path.cwd() #working_dir = (current_path / __package__).resolve() working_dir = Path(__file__).absolute().parent if '__file__' in dir() else Path.cwd() # data_dirs = [Path('/data2/mchen36/POD_DATA/cosmic2/LEOPOD/leoOrb/'), # #Path('/data3/xinjiaz/data_GPSRO/SCDR_unpack_STAR/cosmic2/v1.26/level1a/leoOrb/'), # #Path('/data3/xinjiaz/data_GPSRO/UCAR/cosmic2/level1b/leoOrb/'), # #Path('/data2/mchen36/POD_DATA/cosmic2/LEOPOD/leoOrb/'), # # Path('/data2/xinjiaz/data_GPSRO/SCDR_unpack_STAR/cosmic2/v1.26/level1a/leoAtt/'), # Path('/data2/xinjiaz/data_GPSRO/SCDR_unpack_STAR/cosmic2/v1.26/level1a/podCrx/'), # Path('/data2/xinjiaz/data_GPSRO/SCDR_unpack_STAR/cosmic2/v1.26/level1a/opnGns/')] from configuration_cosmic2 import data_dirs 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 = ['leoOrb',' leoAtt', 'podCrx', 'opnGns'] os.chdir(working_dir) #working_dir = str(working_dir) root_out = working_dir / f'L1_extract_{working_leo}' os.makedirs(root_out, exist_ok=True) os.makedirs('s01_podRx3',exist_ok=True) os.makedirs('s01_leoAtt',exist_ok=True) os.makedirs('s01_leoOrb',exist_ok=True) os.makedirs('s04_atmPhs',exist_ok=True) os.makedirs('log',exist_ok=True) logfile = working_dir / 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 = working_dir / f'progress_rt_{working_leo}.npz' if progfile.is_file(): with np.load(progfile, allow_pickle=True) as fid: tarfiles = fid['tarfiles'] filetypes = fid['filetypes'] fileleos = fid['fileleos'] progress = fid['progress'] timetag = fid['timetag'] lastaccess = fid['lastaccess'] else: tarfiles = np.array([],dtype=object) filetypes = np.array([],dtype='int8') fileleos = np.array([],dtype='int8') progress = np.array([],dtype='int8') timetag = np.array([],dtype=datetime) lastaccess = np.array([datetime(2000, 1, 1)]) def update_progress(pnum,state): progress[pnum] = state timetag[pnum] = datetime.now() lastaccess[0] = datetime.now() def save_progress(): np.savez(f'progress_rt_{working_leo}.npz', tarfiles = tarfiles, filetypes = filetypes, fileleos = fileleos, progress = progress, timetag = timetag, lastaccess = lastaccess, ) if datetime.now() - lastaccess[0] < timedelta(minutes=60): log_message("-- Still progressing.") else: 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.") save_progress() #manual_date = len(working_leo)>0 # set True to manually set dates to process manual_date = True # set True to manually set dates to process manual_date = False # set False for cronjob nloop = 0 while True: new_proc = False nloop += 1 if manual_date: # manually assign dates to process idx = int(sub(r'[^\d]',r'',working_leo)) process_dates = [datetime(2026, 5, 1) + timedelta(days=x) for x in range(61) if x!=39][3*idx:3*idx+3] else: #dt_today = datetime.today() #process_dates = [dt_today + timedelta(days=x) for x in range(-2,2)] # 3 recent days dt_today = datetime.now(timezone.utc) process_dates = [dt_today + timedelta(days=x) for x in range(-1,1)] # 2 recent days, yesterday and today root_files = [np.hstack([list((data_dirs[0] / f'{x:%Y/%j}').glob('*_sp3') if 'UCAR' in data_dirs[0].as_posix() else (data_dirs[0] / f'{x:%Y.%j}').glob('*1.SP3')) for x in process_dates]), np.hstack([list((data_dirs[1] / f'{x:%Y/%j}').glob('*_txt')) for x in process_dates]), np.hstack([list((data_dirs[2] / f'{x:%Y/%j}').glob('*_rnx')) for x in process_dates]), np.hstack([list((data_dirs[3] / f'{x:%Y/%j}').glob('*_bin')) for x in process_dates])] root_types = [np.full((len(root_files[x]),),x) for x in range(len(root_files))] root_leos = [[int(sub(r'.*_\d{4}\.\d{3}\.\d\d(\d)\.\d\d_.*' if 'UCAR' in data_dirs[0].as_posix() else r'.*_C2(\d)\d\.SP3',r'\1',x.name)) for x in root_files[0]], [int(sub(r'.*_\d{4}\.\d{3}\.\d\d(\d)\.\d\d_txt',r'\1',x.name)) for x in root_files[1]], [int(sub(r'.*\.\d\d\.\d\d\.\d\d(\d)\.\d\d\..*',r'\1',x.name)) for x in root_files[2]], [int(sub(r'.*\.\d\d\.\d\d\.\d\d(\d)\.\d\d\..*',r'\1',x.name)) for x in root_files[3]]] root_files = np.hstack(root_files) root_types = np.hstack(root_types).astype('int8') root_leos = np.hstack(root_leos).astype('int8') idx = np.array([tarfile in root_files for tarfile in tarfiles], dtype=bool) tarfiles = tarfiles[idx] filetypes = filetypes[idx] fileleos = fileleos[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])) filetypes = np.concatenate((filetypes, root_types[idx])) fileleos = np.concatenate((fileleos, root_leos[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 = fileleos != 1 # tarfiles = tarfiles[idx] # filetypes = filetypes[idx] # fileleos = fileleos[idx] # progress = progress[idx] # timetag = timetag[idx] # idx = filetypes!=3 # progress[idx] = 1 idx = np.argsort(tarfiles) tarfiles = tarfiles[idx] filetypes = filetypes[idx] fileleos = fileleos[idx] progress = progress[idx] timetag = timetag[idx] save_progress() try: idx = np.where(progress==-1)[0] uniq_leos = np.unique(fileleos[idx]) for leo in uniq_leos: idx_now = idx[fileleos[idx] == leo] log_message(f"-- Processing C2E{leo} files ({len(idx_now)}/{len(idx)}).") update_progress(idx_now, 0) save_progress() 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.makedirs(f't03_ROhold_{working_leo}', exist_ok=True) 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') gather_data(working_leo, tarfiles[idx_now], filetypes[idx_now], f'C2E{leo}') pair_ro_ref(f'C2E{leo}', working_leo) cal_exph(f'C2E{leo}', working_leo, nProcess, '') update_progress(idx_now, 1) # 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 if not new_proc: if nloop > 1: log_message(f'- No new item, exit program.') lastaccess[0] = datetime(2000, 1, 1) save_progress() break