#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os as os_system
import argparse
import json
import logging
from copy import deepcopy
from os import makedirs
from os.path import exists as path_exist
import glob
import shutil
import time as time
from datetime import datetime
import re
#The following relative imports are necessary to create an executable command
from ..ANTARESS_process.ANTARESS_main import ANTARESS_main,ANTARESS_settings_overwrite
from ..ANTARESS_launch.ANTARESS_gridrun import ANTARESS_gridrun
from ..ANTARESS_general.utils import import_module,stop,dataload_npz,datasave_npz,path,time_count,path_nonpz
[docs]
def ANTARESS_launcher(sequence = '' , custom_systems = '',custom_settings = '',custom_plot_settings = '',working_path='',nbook_dic = {} , exec_comm = True):
r"""**ANTARESS launch routine.**
Runs ANTARESS with default or manual settings.
This routine can be called directly from a python environment, or from the terminal.
Args:
sequence (str): name of custom sequence (default "": default settings are used)
custom_systems (str): name of custom systems file (default "": ANTARESS_systems.py file is used)
custom_settings (str): name of custom settings file (default "": ANTARESS_settings.py file is used)
custom_plot_settings (str): name of custom plot settings file (default "": ANTARESS_plot_settings.py file is used)
working_path (str): path to the working directory, in which the workflow outputs will be stored (default "": current directory is used)
if custom files are used, they should be placed in the working directory
Returns:
None
"""
#Suppress log messages from the fontTools package
fontTools_logger = logging.getLogger('fontTools')
fontTools_logger.addHandler(logging.NullHandler())
#Read executable arguments
# - will be used when ANTARESS is called as an executable through terminal
if exec_comm:
parser=argparse.ArgumentParser(prog = "antaress",description='Launch ANTARESS workflow')
parser.add_argument("--sequence", type=str, default='',help = 'Name of custom sequence (default "": default settings are used)')
parser.add_argument("--custom_systems", type=str, default='',help = 'Name of custom systems file (default "": default file ANTARESS_systems.py is used)')
parser.add_argument("--custom_settings", type=str, default='',help = 'Name of custom settings file (default "": default file ANTARESS_settings.py is used)')
parser.add_argument("--custom_plot_settings",type=str, default='',help = 'Name of custom plot settings file (default "": default file ANTARESS_plot_settings.py is used)')
parser.add_argument("--working_path", type=str, default='' ,help = 'Path to user settings files (default "./": user files are retrieved from current directory)')
parser.add_argument("-d", "--nbook_dic", type=json.loads, default={})
input_args=parser.parse_args()
sequence = input_args.sequence
custom_systems = input_args.custom_systems
custom_settings = input_args.custom_settings
custom_plot_settings = input_args.custom_plot_settings
working_path = input_args.working_path
nbook_dic = input_args.nbook_dic
#Initializes main dictionaries
gen_dic={'sequence':sequence}
plot_dic={}
data_dic={
'DI':{'fit_prof':{},'mask':{}},
'Diff':{},
'PCA':{},
'Intr':{'fit_prof':{},'mask':{}},
'Atm':{'fit_prof':{},'mask':{}}}
mock_dic={}
theo_dic={}
detrend_prof_dic={}
glob_fit_dic={
'DIProp':{},
'DIProf':{},
'IntrProp':{},
'DiffProf':{},
'DiffProp':{},
'IntrProf':{},
'AtmProp':{},
'AtmProf':{},
}
input_dics = (data_dic,mock_dic,gen_dic,theo_dic,plot_dic,glob_fit_dic,detrend_prof_dic)
#Working directory
if working_path=='':gen_dic['save_dir'] = os_system.getcwd()+'/'
else:gen_dic['save_dir']= working_path
gen_dic['settings_path_dic'] = {'settings':{},'systems':{},'plots':{}}
#Code directory
code_dir = os_system.path.dirname(__file__).split('ANTARESS_launch')[0]
#----------------------------------------------------------------------------------------------------
#Default settings files
gen_dic['settings_path_dic']['settings']['default'] = code_dir+'ANTARESS_launch/ANTARESS_settings.py'
gen_dic['settings_path_dic']['systems']['default'] = code_dir+'ANTARESS_launch/ANTARESS_systems.py'
gen_dic['settings_path_dic']['plots']['default'] = code_dir+'ANTARESS_plots/ANTARESS_plot_settings.py'
#Retrieve default settings
from ..ANTARESS_launch.ANTARESS_settings import ANTARESS_settings
ANTARESS_settings(*input_dics)
#Overwrite with user settings
if custom_settings!='':
gen_dic['settings_path_dic']['settings']['custom'] = gen_dic['save_dir']+custom_settings
import_module(working_path+custom_settings).ANTARESS_settings(*input_dics)
#Overwrite with notebook settings
if ('settings' in nbook_dic) and (len(nbook_dic['settings'])>0):
ANTARESS_settings_overwrite(*input_dics,nbook_dic)
#Retrieve default or user-defined system properties
if custom_systems!='':
gen_dic['settings_path_dic']['systems']['custom'] = gen_dic['save_dir']+custom_systems
systems_file = import_module(working_path+custom_systems)
else:systems_file = import_module(gen_dic['settings_path_dic']['systems']['default'])
all_system_params = systems_file.get_system_params()
#Overwrite with notebook settings
if ('system' in nbook_dic) and (len(nbook_dic['system'])>0):
all_system_params.update(nbook_dic['system'])
#Retrieve chosen system
system_params = all_system_params[gen_dic['star_name']]
#Default and custom plot settings
if custom_plot_settings!='':gen_dic['settings_path_dic']['plots']['custom'] = gen_dic['save_dir']+custom_plot_settings
#Run directory
gen_dic['save_system'] = gen_dic['save_dir']+gen_dic['star_name']+gen_dic['dir_suff']+'/'
#----------------------------------------------------------------------------------------------------
print('****************************************')
print('Launching ANTARESS')
print('****************************************')
print('')
st0 = time.time()
#Sequence
if len(sequence)>0:
if sequence =='night_proc':txt_seq = 'processing of NIGHT data'
elif sequence=='drifts_NIGHT':txt_seq = 'computation of NIGHT drifts'
elif sequence=='st_master_tseries':txt_seq = 'computation of stellar master from time-series'
elif sequence=='st_rv_tseries':txt_seq = 'computation of stellar RVs from time-series'
elif sequence=='st_CCF_mask':txt_seq = 'computation of stellar CCF mask'
elif sequence=='system_view':txt_seq = 'system view'
elif sequence in ['tell_corr','tell_corr_RV']:txt_seq = 'optimization of telluric correction'
else:
print('WARNING: requested sequence "'+sequence+'" is unknown')
txt_seq=None
if txt_seq is not None:print('Sequence : '+txt_seq)
print('')
#Moving to code directory
os_system.chdir(code_dir)
#Run over single set of settings
# - notebook settings have already been used to overwrite configuration settings, and are only passed on to overwrite the plot settings if relevant
if not gen_dic['grid_run']:
ANTARESS_main(*input_dics,system_params,nbook_dic,custom_plot_settings)
#Run over a grid of properties
# - will overwrite default and notebook configuration settings
else:
ANTARESS_gridrun(*input_dics,system_params,nbook_dic,custom_plot_settings)
print('')
time_count(st0,units='sec')
print('End of workflow')
#Copy settings files to the save directory
# - after run is done, to avoid iterating on the settings archive when testing the workflow
copy_settings_files_to_save_dir(gen_dic['settings_path_dic'],gen_dic['save_system'])
return None
[docs]
def copy_settings_files_to_save_dir(settings_path_dic,run_path):
r"""**Generic launch routines: copy settings files to save directory.**
Copies the three settings files (ANTARESS_settings, ANTARESS_systems, ANTARESS_plot_settings)
to a run-specific subdirectory in Settings_archive. Each run is tracked with a timestamped
folder named RunXXX_YYYY-MM-DD_HH-MM-SS.
Args:
settings_path_dic (dict): dictionary containing paths to settings files
run_path (str) : generic path to current simulation
Returns:
None
"""
# Create Settings_archive subdirectory in the save_system directory
settings_archive_dir = run_path+'/Settings_archive/'
if not path_exist(settings_archive_dir):makedirs(settings_archive_dir)
# Determine run number by checking existing Run* directories
run_dirs = glob.glob(settings_archive_dir + 'Run*')
# Extract run numbers from existing directories
run_numbers = []
run_pattern = re.compile(r'Run(\d+)_')
for run_dir in run_dirs:
match = run_pattern.search(run_dir)
if match:
run_numbers.append(int(match.group(1)))
# Determine next run number
if run_numbers:next_run_num = max(run_numbers) + 1
else:next_run_num = 1
# Create run directory with formatted run number and timestamp
timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
run_dir_name = f'Run{next_run_num:03d}_{timestamp}'
run_dir_path = settings_archive_dir + run_dir_name + '/'
makedirs(run_dir_path)
# Copy settings files to the run-specific directory
for settings_key in settings_path_dic:
settings_types = list(settings_path_dic[settings_key].keys())
for settings_type in settings_types:
src_file = settings_path_dic[settings_key][settings_type]
scr_file_name = src_file.split('/')[-1]
if settings_type=='default':dest_file = run_dir_path + scr_file_name
elif settings_type=='custom':dest_file = run_dir_path +'/CUSTOM_' + scr_file_name
try:shutil.copy2(src_file, dest_file)
except Exception as e:print(f"WARNING: Could not copy {src_file} to {dest_file}: {e}")
return None
[docs]
def ANTARESS_DACE_launcher(star_name,inst,sub_inst,data_path,working_path,master1D = True,def_nthreads = 16 , debug_mode = False,del_dir = True):
r"""**ANTARESS launch routine: master computation**
Runs ANTARESS on a S2D dataset to generate 2D and 1D master spectra with minimal inputs.
The masters are aligned in the stellar rest frame.
This routine can be called directly from a python environment, or from terminal as::
antaress --sequence st_master_tseries --custom_settings ANTARESS_settings_sequence.py
Using a copy of the default configuration file where the only field that need to be modified is:
+ gen_dic['data_dir_list'] : path to your data directory, for the associated instrument
Args:
star_name (str): name of the star (to be saved in the outputs, not critical to the computation)
inst (str): name of the instrument used to acquire the dataset
sub_inst (str): name of the sub-instrument used to acquire the dataset
data_path (str): path to the dataset directory
working_path (str): path to the directory where ANTARESS outputs are stored
master1D (bool): set to True to calculate 1D master
def_nthreads (int): maximum number of CPUs to be used
debug_mode (bool): set to True to enter debugging mode
del_dir (bool): set to False to prevent automatic deletion of intermediate data products
Returns:
None
"""
#Initialization
settings_lines_default,star_inst,store_dir = ANTARESS_launcher_init(debug_mode,working_path,star_name,sub_inst)
#-------------
#Modifying settings file
# - defining settings files:
# + for dataset reduction, 2D processing, and computation of 2D master spectrum
# + for computation of 1D master spectrum using reduced products
settings_lines_reduc2D = deepcopy(settings_lines_default)
if master1D:settings_lines_master1D = deepcopy(settings_lines_default)
for idx_line, line in enumerate(settings_lines_default):
arr_line = line.split()
#Number of CPUs
if ("gen_dic['def_nthreads']=max([1,int(0.8*cpu_count())])" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['def_nthreads'] = "+str(max([1,def_nthreads])) + '\n'
if master1D:settings_lines_master1D[idx_line] = settings_lines_reduc2D[idx_line]
#Paths to data directory
if ("gen_dic['data_dir_list']={}" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['data_dir_list'] = {'"+inst+"':{'all':'"+path(data_path)+"'}}" + '\n'
if master1D:settings_lines_master1D[idx_line] = settings_lines_reduc2D[idx_line]
#Path to temporary output directory
if ("gen_dic['dir_suff']=''" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['dir_suff'] = '_"+star_name+'_'+sub_inst+"'"+'\n'
if master1D:settings_lines_master1D[idx_line] = settings_lines_reduc2D[idx_line]
#Deactive 2D/1D conversion for 2D master computation
if ("gen_dic['spec_1D_DI']=True" in arr_line):settings_lines_reduc2D[idx_line] = ' '+"gen_dic['spec_1D_DI'] = False" + '\n'
#Deactivation of modules for 1D master computation
# - to avoid reprocessing the 2D data
if master1D:
if ("gen_dic['calc_proc_data']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_proc_data'] = False" + '\n'
if ("gen_dic['calc_gcal']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_gcal'] = False" + '\n'
if ("gen_dic['calc_corr_tell']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_corr_tell'] = False" + '\n'
if ("gen_dic['calc_glob_mast']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_glob_mast'] = False" + '\n'
if ("gen_dic['calc_corr_Fbal']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_corr_Fbal'] = False" + '\n'
if ("gen_dic['calc_cosm']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_cosm'] = False" + '\n'
if ("gen_dic['calc_align_DI']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_align_DI'] = False" + '\n'
if ("gen_dic['calc_flux_sc']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_flux_sc'] = False" + '\n'
if ("gen_dic['calc_DImast']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_DImast'] = False" + '\n'
#Test settings
if debug_mode:
#Steps
#General processing
if ("gen_dic['calc_proc_data']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_proc_data'] = False" + '\n'
# ## if ("gen_dic['del_orders']={}" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['del_orders']={'ESPRESSO':np.append(np.arange(0,130),np.arange(160,170))}" + '\n'
#Calibration
if ("gen_dic['calc_gcal']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_gcal'] = False" + '\n'
#Tellurics
if ("gen_dic['calc_corr_tell']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_corr_tell'] = False" + '\n'
#Flux balance
if ("gen_dic['calc_glob_mast']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_glob_mast'] = False" + '\n'
if ("gen_dic['calc_corr_Fbal']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_corr_Fbal'] = False" + '\n'
#Cosmics
if ("gen_dic['calc_cosm']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_cosm'] = False" + '\n'
#Alignment
if ("gen_dic['calc_align_DI']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_align_DI'] = False" + '\n'
#Broadband scaling
if ("gen_dic['calc_flux_sc']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_flux_sc'] = False" + '\n'
#Master
if ("gen_dic['calc_DImast']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_DImast'] = False" + '\n'
#Binning into 2D master
if ("gen_dic['calc_DIbin']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_DIbin'] = False" + '\n'
#1D conversion
if ("gen_dic['calc_spec_1D_DI']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_spec_1D_DI'] = False" + '\n'
#Binning into 1D master
if ("gen_dic['calc_DIbin']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['calc_DIbin'] = False" + '\n'
# if ("gen_dic['corr_tell']=True" in arr_line): settings_lines_master1D[idx_line] = ' '+"gen_dic['corr_tell'] = False" + '\n'
#Plots
#Calibration
if 1==0:
if ("plot_dic['gcal_all']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['gcal_all'] = 'pdf'" + '\n'
if ("plot_dic['gcal_ord']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['gcal_ord'] = 'pdf'" + '\n'
if ("plot_dic['noises_ord']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['noises_ord'] = 'pdf'" + '\n'
#Tellurics
if 1==0:
if ("plot_dic['tell_CCF']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['tell_CCF'] = 'pdf'" + '\n'
if ("plot_dic['tell_prop']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['tell_prop'] = 'pdf'" + '\n'
#Flux balance
if 1==0:
if ("plot_dic['Fbal_corr']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['Fbal_corr'] = 'pdf'" + '\n'
#Cosmics
if 1==1:
if ("plot_dic['cosm_corr']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['cosm_corr'] = 'pdf'" + '\n'
#-------------
#Processing: reduction and 2D master spectrum
#Saving modified settings files
with open(working_path + "ANTARESS_settings_master2D_"+star_inst+".py", 'w') as file:file.writelines(settings_lines_reduc2D)
#Calling ANTARESS
ANTARESS_launcher(sequence = 'st_master_tseries' , working_path = working_path , custom_settings = 'ANTARESS_settings_master2D_'+star_inst+'.py' ,exec_comm=False)
#-------------
#Processing: 1D master spectrum
if master1D:
#Saving modified settings files
with open(working_path + "ANTARESS_settings_master1D_"+star_inst+".py", 'w') as file:file.writelines(settings_lines_master1D)
#Calling ANTARESS
ANTARESS_launcher(sequence = 'st_master_tseries' , working_path = working_path , custom_settings = 'ANTARESS_settings_master1D_'+star_inst+'.py' ,exec_comm=False)
#-------------
#Retrieve and store relevant files
dir_temp = working_path+'/Star_tseries_'+star_name+'_'+sub_inst
#Open master file
master2D = dataload_npz(glob.glob(dir_temp+'/Saved_data/DIbin_data/'+inst+'_all_spec2D_time0.npz')[0].split('.npz')[0])
#Store complementary information
master2D['star_name'] = star_name
#Save masters in storage directory
datasave_npz(store_dir+'Master2D',master2D)
#Process 1D master
if master1D:
master1D = dataload_npz(glob.glob(dir_temp+'/Saved_data/DIbin_data/'+inst+'_all_spec1D_time0.npz')[0].split('.npz')[0])
master1D['star_name'] = star_name
datasave_npz(store_dir+'Master1D',master1D)
#Move 1D plot
os_system.rename(glob.glob(dir_temp+'/Plots/Binned_DI_data/'+inst+'_all_Indiv/Data/Spec_time/idx0_out0_time*')[0],store_dir+'Master1D_plot.pdf')
#Delete ANTARESS outputs
if del_dir:
shutil.rmtree(dir_temp, ignore_errors=True)
os_system.remove(working_path + "ANTARESS_settings_master2D_"+star_inst+".py")
os_system.remove(working_path + "ANTARESS_settings_master1D_"+star_inst+".py")
return None
[docs]
def ANTARESS_DACE_launcher_RVs(star_name,inst,sub_inst,data_path,working_path,def_nthreads = 16 , debug_mode = False,del_dir = True):
r"""**ANTARESS launch routine: RV computation**
Runs ANTARESS on a S2D dataset to generate RV time series with minimal inputs.
RVs are derived from disk-integrated CCFs, in the solar barycentric rest frame.
This routine can be called directly from a python environment, or from terminal as::
antaress --sequence st_rv_tseries --custom_settings ANTARESS_settings_sequence.py
Using a copy of the default configuration file where the only field that need to be modified is:
+ gen_dic['data_dir_list'] : path to your data directory, for the associated instrument
Args:
star_name (str): name of the star (to be saved in the outputs, not critical to the computation)
inst (str): name of the instrument used to acquire the dataset
sub_inst (str): name of the sub-instrument used to acquire the dataset
data_path (str): path to the dataset directory
working_path (str): path to the directory where ANTARESS outputs are stored
def_nthreads (int): maximum number of CPUs to be used
debug_mode (bool): set to True to enter debugging mode
del_dir (bool): set to False to prevent automatic deletion of intermediate data products
Returns:
None
"""
#Initialization
settings_lines_default,star_inst,store_dir = ANTARESS_launcher_init(debug_mode,working_path,star_name,sub_inst)
#-------------
#Modifying settings file
# - defining settings files:
# + for dataset reduction, 2D processing, CCF computation, RV derivation
settings_lines_reduc2D = deepcopy(settings_lines_default)
for idx_line, line in enumerate(settings_lines_default):
arr_line = line.split()
#Number of CPUs
if ("gen_dic['def_nthreads']=max([1,int(0.8*cpu_count())])" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['def_nthreads'] = "+str(max([1,def_nthreads])) + '\n'
#Paths to data directory
if ("gen_dic['data_dir_list']={}" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['data_dir_list'] = {'"+inst+"':{'all':'"+path(data_path)+"'}}" + '\n'
#Path to temporary output directory
if ("gen_dic['dir_suff']=''" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['dir_suff'] = '_"+star_name+'_'+sub_inst+"'"+'\n'
#Test settings
if debug_mode:
#Steps
#General processing
if ("gen_dic['calc_proc_data']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_proc_data'] = False" + '\n'
#Calibration
if ("gen_dic['calc_gcal']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_gcal'] = False" + '\n'
#Tellurics
if ("gen_dic['calc_corr_tell']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_corr_tell'] = False" + '\n'
#Flux balance
if ("gen_dic['calc_glob_mast']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_glob_mast'] = False" + '\n'
if ("gen_dic['calc_corr_Fbal']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_corr_Fbal'] = False" + '\n'
#Cosmics
if ("gen_dic['calc_cosm']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_cosm'] = False" + '\n'
# #CCF conversion
# if ("gen_dic['calc_'+data_type+'_CCF']=True" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"gen_dic['calc_'+data_type+'_CCF'] = False" + '\n'
#CCFs
if 1==1:
if ("plot_dic['DI_prof']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['DI_prof'] = 'pdf'" + '\n'
if ("plot_dic['prop_DI']=''" in arr_line): settings_lines_reduc2D[idx_line] = ' '+"plot_dic['prop_DI'] = 'pdf'" + '\n'
#-------------
#Processing
#Saving modified settings files
with open(working_path + "ANTARESS_settings_DIrv_"+star_inst+".py", 'w') as file:file.writelines(settings_lines_reduc2D)
#Calling ANTARESS
ANTARESS_launcher(sequence = 'st_rv_tseries' , working_path = working_path , custom_settings = 'ANTARESS_settings_DIrv_'+star_inst+'.py' ,exec_comm=False)
#-------------
#Retrieve and store relevant files
dir_temp = working_path+'/Star_rv_tseries_'+star_name+'_'+sub_inst
# #Open custom file
# master2D = dataload_npz(glob.glob(dir_temp+'/Saved_data/DIbin_data/'+inst+'_all_spec2D_time0.npz')[0].split('.npz')[0])
# #Store complementary information
# master2D['star_name'] = star_name
# #Save masters in storage directory
# datasave_npz(store_dir+'Master2D',master2D)
# #Move 1D plot
# os_system.rename(glob.glob(dir_temp+'/Plots/Binned_DI_data/'+inst+'_all_Indiv/Data/Spec_time/idx0_out0_time*')[0],store_dir+'Master1D_plot.pdf')
#Delete ANTARESS outputs
if del_dir:
shutil.rmtree(dir_temp, ignore_errors=True)
os_system.remove(working_path + "ANTARESS_settings_DIrv_"+star_inst+".py")
return None
[docs]
def ANTARESS_launcher_CCFmask(star_name,inst,sub_inst,working_path,master_path,def_nthreads = 16 , debug_mode = False,del_dir = True):
r"""**ANTARESS launch routine: CCF mask computation**
Runs ANTARESS on a S1D master to generate a star-specific CCF mask with minimal inputs.
This routine can be called directly from a python environment, or from terminal as::
antaress --sequence st_CCF_mask --custom_settings ANTARESS_settings_sequence.py
Using a copy of the default configuration file where the only field that need to be modified is:
+ gen_dic['data_dir_list'] : path to your data directory, for the associated instrument
Args:
star_name (str): name of the star (to be saved in the outputs, not critical to the computation)
inst (str): name of the instrument used to acquire the dataset
sub_inst (str): name of the sub-instrument used to acquire the dataset
working_path (str): path to the directory where ANTARESS outputs are stored
master_path (str): path to an ANTARESS .npz file storing a stellar 1D master spectrum
def_nthreads (int): maximum number of CPUs to be used
debug_mode (bool): set to True to enter debugging mode
del_dir (bool): set to False to prevent automatic deletion of intermediate data products
Returns:
None
"""
#Initialization
settings_lines_default,star_inst,store_dir = ANTARESS_launcher_init(debug_mode,working_path,star_name,sub_inst)
#-------------
#Modifying settings file
settings_lines_reduc2D = deepcopy(settings_lines_default)
for idx_line, line in enumerate(settings_lines_default):
arr_line = line.split()
#Number of CPUs
if ("gen_dic['def_nthreads']=max([1,int(0.8*cpu_count())])" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['def_nthreads'] = "+str(max([1,def_nthreads])) + '\n'
#Path to temporary output directory
if ("gen_dic['dir_suff']=''" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['dir_suff'] = '_"+star_name+'_'+sub_inst+"'"+'\n'
#Path to input master
if ("gen_dic['imp_stcont_mast']={}" in arr_line):
settings_lines_reduc2D[idx_line] = ' '+"gen_dic['imp_stcont_mast'] = {'"+inst+"':{'DI':'"+path_nonpz(master_path)+"'}}"+'\n'
#-------------
#Processing
#Saving modified settings files
with open(working_path + "ANTARESS_settings_CCFmask_"+star_inst+".py", 'w') as file:file.writelines(settings_lines_reduc2D)
#Calling ANTARESS
ANTARESS_launcher(sequence = 'st_CCF_mask' , working_path = working_path , custom_settings = 'ANTARESS_settings_CCFmask_'+star_inst+'.py' ,exec_comm=False)
#-------------
#Retrieve and store relevant files
dir_temp = working_path+'/Star_CCFmask_'+star_name+'_'+sub_inst
#Delete ANTARESS outputs
if del_dir:
shutil.rmtree(dir_temp, ignore_errors=True)
os_system.remove(working_path + "ANTARESS_settings_CCFmask_"+star_inst+".py")
return None
return None
[docs]
def ANTARESS_launcher_init(debug_mode,working_path,star_name,sub_inst):
r"""**ANTARESS launch routine: general initialization**
"""
if debug_mode:print('** DEBUG MODE **')
#Creating storage directory
working_path = path(working_path)
star_inst = star_name+'_'+sub_inst
store_dir = working_path+'/'+star_name+'/'+star_inst+'/'
if (not path_exist(store_dir)):makedirs(store_dir)
#Default settings file
code_dir = os_system.path.dirname(__file__).split('ANTARESS_launch')[0]
default_settings_file = code_dir+'ANTARESS_launch/ANTARESS_settings.py'
with open(default_settings_file,'r') as file:settings_lines_default = file.readlines()
return settings_lines_default,star_inst,store_dir
[docs]
def ANTARESS_NIGHT_proc(star_name,pl_name,date,data_path,working_path,debug_mode = False,del_dir = True):
r"""**ANTARESS launch routine: NIGHT fast processing**
Runs ANTARESS on a NIGHT dataset to derive ... with minimal inputs.
This routine can be called directly from a python environment.
To derive RVs with minimal input, you can also run the workflow from terminal as::
antaress --sequence night_proc --custom_settings ANTARESS_settings_sequence.py
Using a copy of the default configuration file where the only fields that need to be modified are:
+ gen_dic['data_dir_list'] : path to your data directory, for the associated instrument
Args:
TBD
Returns:
None
"""
if debug_mode:print('** DEBUG MODE **')
#Creating storage directory
working_path = path(working_path)
store_dir = working_path+'/'+star_name+'/'+pl_name+'_'+date+'/'
if (not path_exist(store_dir)):makedirs(store_dir)
#Default settings file
code_dir = os_system.path.dirname(__file__).split('ANTARESS_launch')[0]
default_settings_file = code_dir+'ANTARESS_launch/ANTARESS_settings.py'
with open(default_settings_file,'r') as file:settings_lines_default = file.readlines()
return None
