#!/usr/bin/env python3
'''This is a executable python script to run the ECLAIRs simulator.
The script needs a configuration file in argument.
.. seealso:: simu/pipeline/readme.md
Le script has 2 inputs:
* configuration file: example here ecpi/common/data/config_files/simu_config.cfg
* output directory
usage: simu_script.py [-h] [-v] [-s] config_file output_dir
local script to run the ECLAIRs simulator
positional arguments:
config_file configuration file absolute path
output_dir directory to store the output files from the simulator. For
security reasons, files are store in a sub-directory named
ECLAIRs_simulation_dir_*, cleaned each time.
optional arguments:
-h, --help show this help message and exit
-v, --verbose increase output verbosity for additional information in
stdout (e.g., no -v : quiet, -v : info, -vv : debug
-s, --show show flag for display some graphs
This is a executable python script to run the ECLAIRs simulator. The script
needs a configuration file in argument. An example of a working configuration
file could be found in ecpi/common/data/config_files/simu_config.cfg
@author: Catalano Camille, APC/IN2P3/CNRS
'''
import argparse
import os
import configparser
import astropy.time as astrotime
import numpy as np
import logging
import json
from astropy.io import fits
import common.mission.attitude as ptgsvom
import common.sky.catalog as cat
import simu.lib.InstruX as simX
from common.instru.model_geom import InstruECLAIRs
from simu.lib.eclairs_channel import SimuEclairsEnergyChannel
from common.instru.model_effect import ECLAIRsDetectorEffectDefault
from common.mission.observation import EclairsObservation
from common.mission.time import convert_svomref_seconds_in_mjd
from common.instru.array_convention import yzegp_to_ijdet_array
[docs]def simulation_ray_tracing(cfg, simu_id, observation, working_dir, verbose=0, show=None):
"""run the simulation using the ray_tracing method
evts file is saved in working_dir/in/.
:param cfg: configuration container from the configuration file
:type cfg: ConfigParser
:param simu_id: observation section
:type simu_id: string
:param observation: observation parameters
:type observation: EclairsObservation
:param working_dir: PATH/directory where to save files
:type working_dir: string
:param verbose: print outputs from simulation (default=0)
:type verbose: int
:return: sources catalog used in the simulation
:rtype: CatalogFovEclairs
"""
eclairs_pointing = ptgsvom.AttitudeECLAIRs()
eclairs_pointing.set_attitude_instru_ptgori(observation.attitude)
ref_catalog_fov = cat.CatalogFovEclairs(InstruECLAIRs())
ref_catalog_fov.set_ptg_instru(eclairs_pointing)
if cfg[simu_id]['sources_file_type'] == 'swift_cat':
swift_cat = cat.CatalogAstroSourcesBasic()
swift_cat.read_cat_swift_light()
ref_catalog_fov.from_astro_catalog(swift_cat)
else:
ref_catalog_fov = construction_catalog(cfg, simu_id, ref_catalog_fov)
if ref_catalog_fov == None:
return
simu_obs_x = simX.SimuECLAIRsRayTracing(ref_catalog_fov, show)
if not int(cfg['general']['random_flag']):
simu_obs_x.b_photon_noise = False
logging.info("no statistical effects in simulations")
simu_obs_x.simu_catalog(observation.obs_time, (verbose >= 2), show)
try:
simu_obs_x.add_cxb(float(cfg[simu_id]['bkg_lvl']))
except:
pass
simu_obs_x.save_cal_evts(working_dir + '/in/ECL-EVT-CAL_{}.fits'.format(simu_id), t_start=observation.start_time)
simu_obs_x.save_stack_evt_image(working_dir + '/in_shadow/shadowgram_{}.fits'.format(simu_id))
return ref_catalog_fov
[docs]def simulation_projection(cfg, simu_id, observation, working_dir, verbose=0, show=None):
"""run the simulation using the projection method
evts file is saved in working_dir/in/.
:param cfg: configuration container from the configuration file
:type cfg: ConfigParser
:param simu_id: simulation section
:type simu_id: string
:param observation: observation parameters
:type observation: EclairsObservation
:param working_dir: PATH/directory where to save files
:type working_dir: string
:param verbose: print outputs from simulation (default=0)
:type verbose: int
:return: sources catalog used in the simulation
:rtype: CatalogFovEclairs
"""
if cfg['general']['off_axis_effects'] == 'irf':
use_irf = True
else:
use_irf = False
dpix = observation.dpix
dpix.reduce_channel(int(cfg['general']['simu_channels_nb']))
# create object simulator
simu = SimuEclairsEnergyChannel(dpix)
simu.set_context_simu(observation.obs_time, observation.start_time)
eclairs_pointing = ptgsvom.AttitudeECLAIRs()
eclairs_pointing.set_attitude_instru_ptgori(observation.attitude)
if cfg[simu_id]['sources_file_type'] == 'swift_cat':
simu.add_src_point_with_catalog_swift(eclairs_pointing, use_irf, (verbose >= 2))
ref_catalog_fov = simu.cat
else:
ref_catalog_fov = cat.CatalogFovEclairs(InstruECLAIRs())
ref_catalog_fov.set_ptg_instru(eclairs_pointing)
ref_catalog_fov = construction_catalog(cfg, simu_id, ref_catalog_fov)
if ref_catalog_fov == None:
return
simu.add_src_point_with_catalog_fov(ref_catalog_fov, use_irf, (verbose >= 2))
try:
simu.add_src_cxb('Flat',float(cfg[simu_id]['bkg_lvl']))
except:
pass
simu.simu_src(verbose=(verbose >= 2))
simu._simu_arf_mean()
simu._create_shadows_int()
if int(cfg['general']['random_flag']):
simu.simu_global_noise_poisson((verbose >= 1))
else:
logging.info("no statistical effects in simulations")
simu.create_evts((verbose >= 1))
t_start = astrotime.Time(convert_svomref_seconds_in_mjd(observation.start_time), format='mjd', scale='tt')
t_start = t_start.isot[:-4].replace('-','').replace(':','')
simu._evts.write_L1(working_dir + '/in/ECL-EVT-SEC-{}.fits'.format(t_start), 'ECPI simulator')
chan_nb = 0
for shadowgram in simu.chan_shadows:
hdu = fits.PrimaryHDU(yzegp_to_ijdet_array(shadowgram))
hdu.writeto(working_dir + '/in_shadow/shadowgram_{}_{}.fits'.format(simu_id, chan_nb))
chan_nb += 1
return ref_catalog_fov
[docs]def simulation_projection_old(cfg, simu_id, observation, working_dir, verbose=0):
"""run the simulation using the projection method without energy
evts file is saved in working_dir/in/.
:param cfg: configuration container from the configuration file
:type cfg: ConfigParser
:param simu_id: simulation section
:type simu_id: string
:param observation: observation parameters
:type observation: EclairsObservation
:param working_dir: PATH/directory where to save files
:type working_dir: string
:param verbose: print outputs from simulation (default=0)
:type verbose: int
:return: sources catalog used in the simulation
:rtype: CatalogFovEclairs
"""
eclairs_pointing = ptgsvom.AttitudeECLAIRs()
eclairs_pointing.set_attitude_instru_ptgori(observation.attitude)
ref_catalog_fov = cat.CatalogFovEclairs(InstruECLAIRs())
ref_catalog_fov.set_ptg_instru(eclairs_pointing)
if cfg[simu_id]['sources_file_type'] == 'swift_cat':
swift_cat = cat.CatalogAstroSourcesBasic()
swift_cat.read_cat_swift_light()
ref_catalog_fov.from_astro_catalog(swift_cat)
else:
ref_catalog_fov = construction_catalog(cfg, simu_id, ref_catalog_fov)
if ref_catalog_fov == None:
return
simu_obs_x = simX.SimuECLAIRsMaskProjection(ref_catalog_fov)
if not int(cfg['general']['random_flag']):
simu_obs_x.b_photon_noise = False
logging.info("no statistical effects in simulations")
simu_obs_x.simu_catalog(observation.obs_time, (verbose >= 2))
try:
simu_obs_x.add_cxb(float(cfg[simu_id]['bkg_lvl']))
except:
pass
simu_obs_x.save_cal_evts(working_dir + '/in/ECL-EVT-SEC_{}.fits'.format(simu_id), t_start=observation.start_time, l1=True)
simu_obs_x.save_stack_evt_image(working_dir + '/in_shadow/shadowgram_{}.fits'.format(simu_id))
return ref_catalog_fov
[docs]def simulation_switcher(cfg, simu_id, observation, working_dir, verbose=0, show=None):
"""simulation method switcher
select the right simulation method and run the associated simulation
evts file is saved in working_dir/in/.
:param cfg: configuration container from the configuration file
:type cfg: ConfigParser
:param simu_id: observation section
:type simu_id: string
:param observation: observation parameters
:type observation: EclairsObservation
:param working_dir: PATH/directory where to save files
:type working_dir: string
:param verbose: print outputs from simulation (default=0)
:type verbose: int
:param show: plot mask from instru (default=None)
:type show: bool
:return: sources catalog used in the simulation
:rtype: CatalogFovEclairs
"""
if cfg['general']['simulation_method'] == 'ray_tracing':
return simulation_ray_tracing(cfg, simu_id, observation, working_dir, verbose, show)
elif cfg['general']['simulation_method'] == 'projection':
return simulation_projection(cfg, simu_id, observation, working_dir, verbose, show)
else:
logging.warning("simulation method not recognized")
return
[docs]def construction_catalog(cfg, simu_id, catalog_fov):
"""build the sources catalog for simulation
build the sources catalog from the sources_file_type
sources_file_type :
- sources_file : configuration file with a list of sources
- swift_cat : swift-BAT catalog
- ECLAIRs_cat : a ECLAIRs sources catalog
:param cfg: configuration container from the configuration file
:type cfg: ConfigParser
:param simu_id: observation section
:type simu_id: string
:param catalog_fov: sources catalog used in the simulation
:type catalog_fov: CatalogFovEclairs
:return: updated sources catalog used in the simulation
:rtype: CatalogFovEclairs
"""
if cfg[simu_id]['sources_file_type'] == 'sources_config':
src_cfg = configparser.ConfigParser()
src_cfg.read(cfg[simu_id]['sources_file'])
nb_chan = int(cfg['general']['simu_channels_nb'])
catalog_fov.intensity_with_channel(nb_chan)
for src in src_cfg.sections():
if src_cfg[src]['coord_type'] == '0':
catalog_fov.add_src_pix([float(src_cfg[src]['Xs']),float(src_cfg[src]['Ys']),json.loads(src_cfg[src]['flux']),src])
elif src_cfg[src]['coord_type'] == '1':
astro_cat = cat.CatalogAstroWithEnergySpecSampling(np.ones(nb_chan))
astro_cat.add_src([float(src_cfg[src]['Xs']),float(src_cfg[src]['Ys']),json.loads(src_cfg[src]['flux']),src])
catalog_fov.from_astro_catalog(astro_cat)
else:
logging.warning("coord_type not recognized in sources config file")
return
return catalog_fov
elif cfg[simu_id]['sources_file_type'] == 'ECLAIRs_cat':
catalog_fov.read_catalog(cfg[simu_id]['sources_file'])
return catalog_fov
else:
logging.warning("source_file_type not recognized")
return
[docs]def read_observation_config(cfg, simu_id):
"""read the observation config parameters
:param cfg: simulation input configuration
:type cfg: ConfigParser
:param simu_id: observation section
:type simu_id: string
:return: t_0, t_1, eclairs_attitude
:rtype: Time, Time, array([ra, dec, ori])
"""
t_obs = float(cfg[simu_id]['t_obs'])
try:
t_0 = astrotime.Time(cfg[simu_id]['t_start'], format='isot', scale='tt')
except:
t_0 = astrotime.Time('2018-06-16T14:37:07', format='isot', scale='tt')
t_1 = t_0 + astrotime.TimeDelta(t_obs, format='sec')
logging.info('observation of {:.2f}s from {}'.format((t_1-t_0).sec, t_0.isot))
try:
eclairs_attitude = np.array([float(cfg[simu_id]['pointing_ra']), float(cfg[simu_id]['pointing_dec']), float(cfg[simu_id]['pointing_ori'])])
except:
eclairs_attitude = np.array([0, 0, 0])
logging.warning("WARNING: no attitude in input. Default is 0,0,0.")
return t_0, t_1, eclairs_attitude
[docs]def main(args):
"""This is the main function of the executable python script to run the ECLAIRs simulator.
:param args: argument parser
:type args: parse_args
"""
logging.info('########################## ECLAIRs simulator ##########################')
logging.info('reading config file')
show=args.show
verbose=args.verbose
if not verbose:
verbose = 0
cfg = configparser.ConfigParser()
cfg.read(args.config_file.name)
if cfg['general']['random_seed'] != '0':
np.random.seed(int(cfg['general']['random_seed']))
for simu_id in cfg.sections()[1:]:
# env config
logging.info('########################## NEW SIMULATION ##########################')
logging.info('##########################')
logging.info('configuration of the simulation environment')
working_dir = os.path.join(args.output_dir,'./ECLAIRs_simulation_dir_{}'.format(simu_id))
working_dir = os.path.abspath(working_dir)
try:
os.makedirs(working_dir)
except FileExistsError:
pass
os.system('rm -rf ' + working_dir + '/*')
#evts file is saved in working_dir/in/.
os.system("mkdir {}/in".format(working_dir))
#shadowgrams file is saved in working_dir/in_shadow/.
os.system("mkdir {}/in_shadow".format(working_dir))
# simulation config
logging.info('working directory : {}'.format(working_dir))
logging.info('##########################')
logging.info('configuration of the simulation {}'.format(simu_id))
t_0, t_1, eclairs_attitude = read_observation_config(cfg, simu_id)
logging.info("ECLAIRs' pointing is ra={}°, dec={}°, ori={}°".format(eclairs_attitude[0], eclairs_attitude[1], eclairs_attitude[2]))
obs = EclairsObservation(t_0.isot, t_1.isot, eclairs_attitude, [0,0,0], [0,0,0], ECLAIRsDetectorEffectDefault())
obs.gti_compute()
#start the process
logging.info('##########################')
logging.info('start the simulation')
catalog_fov = simulation_switcher(cfg, simu_id, obs, working_dir, verbose, show)
if catalog_fov == None:
logging.info('simulation failed !!!')
logging.info('##########################')
continue
catalog_fov.save_catalog_pix("{}/catalog_fov_pix.fits".format(working_dir))
catalog_fov.save_catalog("{}/catalog_fov.fits".format(working_dir))
catalog_fov.plot_on_astro_sphere("x-ray sources in FOV", show=show, save="{}/sources_inFOV_sky_plot.png".format(working_dir))
logging.info('##########################')
logging.info('simulation {} finished !'.format(simu_id))
logging.info('########################## all simulation have been processed ##########################')
[docs]def check_directory(value_array):
if not os.path.exists(value_array):
raise argparse.ArgumentTypeError("%s isn't a valid directory" % value_array)
return value_array
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description='local script to run the ECLAIRs simulator',
epilog="""This is a executable python script to run the ECLAIRs simulator.
The script needs a configuration file in argument. An example of a working configuration file could be found in ecpi/common/data/config_files/simu_config.cfg
""")
parser.add_argument(
'config_file',
help='configuration file absolute path',
type=argparse.FileType('r'))
parser.add_argument(
'output_dir',
help="""directory to store the output files from the simulator.
For security reasons, files are store in a sub-directory named ECLAIRs_simulation_dir_*, cleaned each time.""",
type=check_directory)
group = parser.add_mutually_exclusive_group()
group.add_argument('-v', '--verbose',
action='count',
help='increase output verbosity for additional information in stdout (e.g., no -v : quiet, -v : info, -vv : debug' )
parser.add_argument('-s', '--show',
action='store_true',
help='show flag for display some graphs' )
args = parser.parse_args()
if args.verbose == 2:
logging.basicConfig(format='%(message)s', level=logging.DEBUG)
elif args.verbose == 1:
logging.basicConfig(format='%(message)s', level=logging.INFO)
main(args)