import numpy as np import os from datetime import datetime import argparse from astropy.time import Time from ObservationSim.Config import Pointing def parse_args(): ''' Parse command line arguments. Many of the following can be set in the .yaml config file as well. ''' parser = argparse.ArgumentParser() parser.add_argument('--config_file', type=str, required=True, help='.yaml config file for simulation settings.') parser.add_argument('--catalog', type=str, required=True, help='name of the catalog interface class to be loaded.') parser.add_argument('-c', '--config_dir', type=str, help='Directory that houses the .yaml config file.') parser.add_argument('-d', '--data_dir', type=str, help='Directory that houses the input data.') parser.add_argument('-w', '--work_dir', type=str, help='The path for output.') return parser.parse_args() def generate_pointing_list(config, pointing_filename=None, data_dir=None): pointing_list = [] # Only valid when the pointing list does not contain time stamp column t0 = datetime(2021, 5, 25, 12, 0, 0) delta_t = 10. # Time elapsed between exposures (minutes) # Calculate starting time(s) for CAL exposures # NOTE: temporary implementation t = datetime.timestamp(t0) ncal = config['obs_setting']['np_cal'] ipoint = 0 for i in range(ncal): pointing = Pointing( id = ipoint, ra=config["obs_setting"]["ra_center"], dec=config["obs_setting"]["dec_center"], img_pa=config["obs_setting"]["image_rot"], timestamp=t, pointing_type='CAL') t += 3 * delta_t * 60. # 3 calibration exposures for each pointing pointing_list.append(pointing) ipoint += 1 run_pointings = config['obs_setting']['run_pointings'] if config["obs_setting"]["exp_time"]: exp_time = config["obs_setting"]["exp_time"] else: exp_time = 150. if pointing_filename and data_dir: pointing_file = os.path.join(data_dir, pointing_filename) f = open(pointing_file, 'r') for _ in range(1): header = f.readline() iline = 0 for line in f: if run_pointings and iline not in run_pointings: iline += 1 ipoint += 1 continue line = line.strip() columns = line.split() pointing = Pointing(exp_time=exp_time) pointing.read_pointing_columns(columns=columns, id=ipoint, t=t) t += delta_t * 60. pointing_list.append(pointing) iline += 1 ipoint += 1 f.close() else: pointing = Pointing( id=ipoint, ra=config["obs_setting"]["ra_center"], dec=config["obs_setting"]["dec_center"], img_pa=config["obs_setting"]["image_rot"], timestamp=t, exp_time=exp_time, pointing_type='SCI' ) t += delta_t * 60. pointing_list.append(pointing) ipoint += 1 return pointing_list def make_run_dirs(work_dir, run_name, pointing_list): if not os.path.exists(work_dir): try: os.makedirs(work_dir, exist_ok=True) except OSError: pass imgDir = os.path.join(work_dir, run_name) if not os.path.exists(imgDir): try: os.makedirs(imgDir, exist_ok=True) except OSError: pass prefix = "MSC_" # for pointing in pointing_list: # fname=prefix + str(pointing.id).rjust(7, '0') # subImgDir = os.path.join(imgDir, fname) # if not os.path.exists(subImgDir): # try: # os.makedirs(subImgDir, exist_ok=True) # except OSError: # pass return imgDir def imgName(tt=0): ut = datetime.utcnow() eye, emo, eda, eho, emi, ese = str(ut.year), str(ut.month), str(ut.day), str(ut.hour), str(ut.minute), str(ut.second) emse = str(ut.microsecond) if int(emo)<10: emo = "0%s"%emo if int(eda)<10: eda = "0%s"%eda if int(eho)<10: eho = "0%s"%eho if int(emi)<10: emi = "0%s"%emi if int(ese)<10: ese = "0%s"%ese if tt==0: namekey = "CSST%s%s%sT%s%s%s"%(eye,emo,eda,eho,emi,ese) elif tt==1: namekey = "%s-%s-%sT%s:%s:%s.%s"%(eye,emo,eda,eho,emi,ese,emse) elif tt==2: namekey = "%s%s%s%s%s%s"%(eye,emo,eda,eho,emi,ese) else: raise ValueError("!!! Give a right 'tt' value.") return namekey def makeSubDir_PointingList(path_dict, config, pointing_ID=0): imgDir = os.path.join(path_dict["work_dir"], config["run_name"]) if not os.path.exists(imgDir): try: os.makedirs(imgDir, exist_ok=True) except OSError: pass prefix = "MSC_" + str(pointing_ID).rjust(8, '0') subImgdir = os.path.join(imgDir, prefix) if not os.path.exists(subImgdir): try: os.makedirs(subImgdir, exist_ok=True) except OSError: pass return subImgdir, prefix def get_shear_field(config): if not config["shear_setting"]["shear_type"] in ["constant", "catalog"]: raise ValueError("Please set a right 'shear_method' parameter.") if config["shear_setting"]["shear_type"] == "constant": g1 = config["shear_setting"]["reduced_g1"] g2 = config["shear_setting"]["reduced_g2"] nshear = 1 # TODO logging else: g1, g2 = 0., 0. nshear = 0 return g1, g2, nshear