"""
"""
import logging
import os.path as osp
import numpy as np
from astropy.time import Time
import ecpi.logger_ecpi as log_ecpi
import ecpi.common as ec
from ecpi.common.instru.model_effect import ECLAIRsDetectorEffectDefault
from ecpi.common.mission.attitude import AttitudeECLAIRs
from ecpi.simu.lib.eclairs_channel import SimuEclairsEnergyChannel, AVAILABLE_CXB_MODES
from ecpi.simu.lib.sources import ListModelPointSrcFromCatalog, ModelPointSrcIRF
# from ecpi_garage.simu.lib_simu_pipeline_dc2 import SimuDC2
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.simu.lib.sources import ModelPointSrcSinElev
from ecpi.simu.lib.context import GlobalContextSimulator
slogger = logging.getLogger(log_ecpi.get_logger_path(__file__))
TREF = Time('2017-01-01T00:00:00.000')
# AVAILABLE_OPTS_CXB_MODE = ['flat', 'flat_moretti', 'shape_moretti', 'shape_moretti_earth', 'no_cxb']
[docs]def simu_cxb_only(dict_simu_pars):
"""Generates event file(s) for a CXB model only (no source !).
:param dict_simu_pars: simulation parameter dictionnary
:type dict_simu_pars: dict
"""
assert dict_simu_pars['cxbmode'] in AVAILABLE_CXB_MODES
assert dict_simu_pars['srcname'] is ''
ctxt = GlobalContextSimulator()
dpix = ECLAIRsDetectorEffectDefault()
swift = ListModelPointSrcFromCatalog()
swift.set_use_irf(dict_simu_pars['useirf'])
orb_file = get_fits_files_with_extname("SVO-ORB-CNV", dict_simu_pars["destdir"])[0]
att_file = get_fits_files_with_extname("SVO-ATT-CNV", dict_simu_pars["destdir"])[0]
ecl_orb = EclairsOrbit(orb_file, att_file)
idx_line = 0 # statics simulator
time_pvt = ecl_orb.data['time_pvt'][idx_line]
position = ecl_orb.data['position'][idx_line]
t_start_s = (dict_simu_pars['tstart'].mjd - TREF.mjd) * 24 * 3600
secl = SimuEclairsEnergyChannel(dpix)
secl.set_range_channel(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_t_start(t_start_s)
secl.context.set_pos_sat(position)
secl.context.set_quaternion_svom(dict_simu_pars['quater'])
secl.context._set_energy_band(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_duration(dict_simu_pars['texposure'])
# add CXB source if required
if dict_simu_pars['cxbmode'] is 'flat':
secl.add_src_cxb(cxb_mode="flat")
elif dict_simu_pars['cxbmode'] is 'shape_moretti':
secl.add_src_cxb(cxb_mode="shape_moretti")
elif dict_simu_pars['cxbmode'] is 'flat_moretti':
secl.add_src_cxb(cxb_mode="flat_moretti")
elif dict_simu_pars['cxbmode'] is 'shape_moretti_earth':
secl.add_src_cxb(cxb_mode="shape_moretti_earth")
# add CXB mode, Poisson noise
# and simulate arf.
secl.simu_src()
secl.simu_arf_mean()
print(f"After ARF mult= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.simu_noise_poisson()
print(f"After Pois. noise= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.create_evts()
# write event file
if dict_simu_pars['eventfile']:
srcname = 'CXBonly'
nb_chan = secl._range_chan.stop - secl._range_chan.start
name_file = f"{srcname}_{secl.context.duration}s_{nb_chan}" \
f"chan_cxbmode_{dict_simu_pars['cxbmode']}.fits"
secl._evts.write_L1( osp.join(dict_simu_pars['destdir'], name_file), "ECPI/APC")
slogger.info('Created event file')
[docs]def simu_onesrc_with_cxb_model_fake(dict_simu_pars, info_src):
"""Generates event file(s) for a single source and CXB.
Simulated source is in SWIFT/BAT catalog.
:param dict_simu_pars: simulation parameter dictionary
:type dict_simu_pars: dict
:param info_src: info for the source to be simulated
:type info_src: dict
"""
assert dict_simu_pars['cxbmode'] in AVAILABLE_CXB_MODES
att = AttitudeECLAIRs()
dpix = ECLAIRsDetectorEffectDefault()
att.set_attitude_svom_quater(dict_simu_pars['quater'])
orb_file = get_fits_files_with_extname("SVO-ORB-CNV", dict_simu_pars["destdir"])[0]
att_file = get_fits_files_with_extname("SVO-ATT-CNV", dict_simu_pars["destdir"])[0]
ecl_orb = EclairsOrbit(orb_file, att_file)
idx_line = 0 # statics simulator
time_pvt = ecl_orb.data['time_pvt'][idx_line]
position = ecl_orb.data['position'][idx_line]
t_start_s = (dict_simu_pars['tstart'].mjd - TREF.mjd) * 24 * 3600
secl = SimuEclairsEnergyChannel(dpix)
secl.set_range_channel(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_t_start(t_start_s)
secl.context.set_pos_sat(position)
secl.context.set_quaternion_svom(dict_simu_pars['quater'])
secl.context._set_energy_band(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_duration(dict_simu_pars['texposure'])
# load source in SWIFT catalog
src = ModelPointSrcSinElev(info_src)
# add CXB source if required
if dict_simu_pars['cxbmode'] is 'flat':
secl.add_src_cxb(cxb_mode="flat")
elif dict_simu_pars['cxbmode'] is 'shape_moretti':
secl.add_src_cxb(cxb_mode="shape_moretti")
elif dict_simu_pars['cxbmode'] is 'flat_moretti':
secl.add_src_cxb(cxb_mode="flat_moretti")
elif dict_simu_pars['cxbmode'] is 'shape_moretti_earth':
secl.add_src_cxb(cxb_mode="shape_moretti_earth")
# add lonely source, Poisson noise
# and simulate arf.
secl.add_src(src)
secl.simu_src()
secl.simu_arf_mean()
print(f"After ARF mult= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.simu_noise_poisson()
print(f"After Pois. noise= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.create_evts()
# write event file
if dict_simu_pars['eventfile']:
nb_chan = secl._range_chan.stop - secl._range_chan.start
name_file = f"{info_src['name']}_{dict_fits_pars['texposure']}s_{nb_chan}" \
f"chan_cxbmode_{dict_simu_pars['cxbmode']}.fits"
secl._evts.write_L1(osp.join(dict_simu_pars['destdir'], name_file), "ECPI/APC")
slogger.info('Created event file')
[docs]def simu_onesrc_with_cxb_model(dict_simu_pars):
"""Generates event file(s) for a single source and CXB.
Simulated source is in SWIFT/BAT catalog.
:param dict_simu_pars: simulation parameter dictionnary
:type dict_simu_pars: dict
"""
assert dict_simu_pars['cxbmode'] in AVAILABLE_CXB_MODES
assert dict_simu_pars['srcname'] is not ''
dpix = ECLAIRsDetectorEffectDefault()
orb_file = get_fits_files_with_extname("SVO-ORB-CNV", dict_simu_pars["destdir"])[0]
att_file = get_fits_files_with_extname("SVO-ATT-CNV", dict_simu_pars["destdir"])[0]
ecl_orb = EclairsOrbit(orb_file, att_file)
idx_line = 0 # statics simulator
time_pvt = ecl_orb.data['time_pvt'][idx_line]
position = ecl_orb.data['position'][idx_line]
t_start_s = (dict_simu_pars['tstart'].mjd - TREF.mjd) * 24 * 3600
secl = SimuEclairsEnergyChannel(dpix)
secl.set_range_channel(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_t_start(t_start_s)
secl.context.set_pos_sat(position)
secl.context.set_quaternion_svom(dict_simu_pars['quater'])
secl.context._set_energy_band(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
secl.context.set_duration(dict_simu_pars['texposure'])
# add astro source
swift = ListModelPointSrcFromCatalog()
swift.set_use_irf(dict_simu_pars['useirf'])
d_src = swift.init_with_cat_swift_2012()
# load source in SWIFT catalog
if dict_simu_pars['srcname'] in list(d_src.keys()):
src = d_src[dict_simu_pars['srcname']]
else:
print(f"No such a source: {dict_simu_pars['srcname']}")
assert isinstance(src, ModelPointSrcIRF)
del d_src
# add CXB source if required
if dict_simu_pars['cxbmode'] is 'flat':
secl.add_src_cxb(cxb_mode="flat")
elif dict_simu_pars['cxbmode'] is 'shape_moretti':
secl.add_src_cxb(cxb_mode="shape_moretti")
elif dict_simu_pars['cxbmode'] is 'flat_moretti':
secl.add_src_cxb(cxb_mode="flat_moretti")
elif dict_simu_pars['cxbmode'] is 'shape_moretti_earth':
secl.add_src_cxb(cxb_mode="shape_moretti_earth")
# add lonely source, Poisson noise
# and simulate arf.
secl.add_src(src)
secl.simu_src()
secl.simu_arf_mean()
print(f"After ARF mult= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.simu_noise_poisson()
print(f"After Pois. noise= {np.sum(secl.chan_shadows[dict_simu_pars['idx_emin']:dict_simu_pars['idx_emax']+1, :, :])}")
secl.create_evts()
# write event file
if dict_simu_pars['eventfile']:
nb_chan = secl._range_chan.stop - secl._range_chan.start
name_file = f"{dict_simu_pars['srcname']}_{secl.context.duration}s_{nb_chan}" \
f"chan_cxbmode_{dict_simu_pars['cxbmode']}.fits"
secl._evts.write_L1(osp.join(dict_simu_pars['destdir'], name_file), "ECPI/APC")
slogger.info('Created event file')
[docs]def simu_fov_with_cxb_model(dict_simu_pars):
"""Generates event file(s) for all source in FOV and CXB.
:param dict_simu_pars: simulation parameter dictionnary
:type dict_simu_pars: dict
"""
assert dict_simu_pars['cxbmode'] in AVAILABLE_CXB_MODES
assert dict_simu_pars['srcname'] is ''
# dt = t_start-TREF
# t_new = Time(dt.value, format='mjd', scale='tt')
t_new = dict_simu_pars['tstart']
osim = SimuDC2(dict_simu_pars['texposure'] / 60.,
dict_simu_pars['position'],
dict_simu_pars['nfiles'], t_0=t_new)
osim.set_quater(dict_simu_pars['quater'])
osim.create_obj_simu()
osim.simulate_cxb(
dict_simu_pars['destdir'],
use_irf=dict_simu_pars['useirf'],
cxb_mode=dict_simu_pars['cxbmode'])
osim.secl.set_range_channel(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
if dict_simu_pars['eventfile']:
osim.create_event_files(dict_simu_pars['destdir'])
slogger.info('Created event file')
[docs]def create_attitude_orbit_files(dict_fits_pars):
"""Generates attitude and/or orbit file(s).
:param dict_fits_pars: parameter dictionnary
:type dict_fits_pars: dict
"""
t_new = dict_fits_pars['tstart']
osim = SimuDC2(dict_fits_pars['texposure'] / 60.,
dict_fits_pars['position'],
dict_fits_pars['nfiles'], t_0=t_new)
osim.set_quater(dict_fits_pars['quater'])
osim.create_obj_simu()
# osim.secl.set_range_channel(dict_simu_pars['idx_emin'], dict_simu_pars['idx_emax'])
if dict_fits_pars['attfile']:
osim.create_attitude_file(dict_fits_pars['destdir'])
slogger.info('Created attitude file')
if dict_fits_pars['orbfile']:
osim.create_orbit_file(dict_fits_pars['destdir'])
slogger.info('Created orbit file')
[docs]def get_list_src_in_fov(dict_fits_pars, filter_substring=None):
"""Return the list of sources in FOV specified by quaternion.
Possibility to filter list.
"""
dpix = ECLAIRsDetectorEffectDefault()
swift = ListModelPointSrcFromCatalog()
swift.set_use_irf(False)
att = AttitudeECLAIRs()
att.set_attitude_svom_quater(dict_fits_pars['quater'])
d_src = swift.init_with_cat_swift_2012(att)
list_src = list(d_src.keys())
counter_src_fov = 0
counter_src_fov_filter = 0
for elt in list_src:
src = d_src[elt]
if filter_substring is None:
print(src._name)
else:
if filter_substring in src._name:
print(src._name)
counter_src_fov_filter += 1
counter_src_fov += 1
print(f"Total number of sources in FOV: {counter_src_fov}")
print(f"Total number of sources in FOV [filtered]: {counter_src_fov_filter}")
if __name__ == '__main__':
# Start of observation
tstart = Time('2021-01-02T00:14:00.000')
# Where to dump fits files
dest_dir = ec.add_path_output_eclairs("")
# Create attitude and orbit files.
dict_fits_pars = {
'nfiles': 1,
'quater': [-0.4274988, 0.6577391, -0.3015459, 0.5419356],
'position': [6934890, 2563.61, -928121],
'texposure': 1, # s
'attfile': True,
'orbfile': True,
'tstart': tstart,
'destdir': dest_dir
}
create_attitude_orbit_files(dict_fits_pars)
if dict_fits_pars['attfile'] or dict_fits_pars['orbfile']:
slogger.info('Created attitude and/or orbit fits file(s)')
# Simulation parameters.
# choose between 'onesrc', 'fakesrc', fov' and 'cxbonly'
mode = 'cxbonly'
info_src = {
'elev': 90.,
'dir': 0.,
'intensity': 0.01,
'name': 'fictive source'
}
dict_simu_pars = {
'srcname': '',
'idx_emin': 0, # 4 keV
'idx_emax': 583, # 150 keV
'nfiles': 1,
'texposure': 1000, # s
'quater': dict_fits_pars["quater"],
'position': dict_fits_pars["position"],
'cxbmode': 'shape_moretti_earth',
'useirf': False,
'eventfile': True,
'tstart': tstart,
'destdir': dest_dir
}
if mode is 'fov':
slogger.info('Running simulation with FOV...')
simu_fov_with_cxb_model(dict_simu_pars)
elif mode is 'onesrc':
slogger.info('Running simulation with single source...')
simu_onesrc_with_cxb_model(dict_simu_pars)
elif mode is 'fakesrc':
slogger.info('Running simulation with single artificial source...')
simu_onesrc_with_cxb_model_fake(dict_simu_pars, info_src)
elif mode is 'cxbonly':
slogger.info('Running simulation with CXB only...')
simu_cxb_only(dict_simu_pars)
else:
slogger.error(f'Unknown mode {mode}')
if dict_fits_pars['attfile'] or dict_fits_pars['orbfile']:
slogger.info('Created event file(s)')