Source code for ecpi.process.imag.imag_with_files
"""
@author: BACON Philippe, APC/IN2P3/CNRS
"""
import logging
import numpy as np
from astropy.io import fits
from ecpi.common.mission.attitude import AttitudeECLAIRs
from ecpi.common import add_path_current_module
from ecpi.process.imag.core.main_imag import imag
from ecpi.process.imag.io.outputs import save_catalogs, save_sky_images
from ecpi.process.generic.composant import GenericComposantProcessing
from ecpi.process.bube.core.main_bube import DetectorImages
from ecpi.process.bube.bube_with_files import read_attitude_files, check_time_coherence_evt_att
from ecpi.common.mission.observation import EclairsObservation
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.process.imag.io.outputs import read_ecl_det_ubc_file
s_logger = logging.getLogger(__name__)
IMAG_FILE_SCHEME = add_path_current_module(__file__, "io/imag_schema.json")
[docs]class ProcessImagWithFiles(GenericComposantProcessing):
"""
High level handling of IMAG module.
"""
def __init__(self):
"""**Constructor**
"""
super().__init__(IMAG_FILE_SCHEME)
self.name = "imag"
self.error_range = 400
dict_comp = {
21: "something wrong happened during IMAG module running",
22: "error in ECL-SKY-IMA product generation",
23: "error in ECL-SOP-IMA product generation",
24: "starting times does not match between event and attitude files",
25: "error no ECL-DET-UBC files found",
26: "error no SVO-ATT-CNV files found"
}
self.d_status.update(dict_comp)
def _load_check_file_in(self):
"""Collect infos so that EclairsObservation and
DetectorImages object can be built from
- files ECL-DET-UBC (from BUBE module)
- files attitude
initialize self.d_data_in like key
"detector_images": DetectorImages (shadowgrams, obs times, nrj ranges)
"pvtatt" : EclairsObservation avec SVO_ATT-CNV...
"""
s_logger.debug('Running _load_check_file_in [imag]')
if not GenericComposantProcessing._load_check_file_in(self):
return False
# check for ECL-DET-UBC and attitude files in working_in dir.
shadowgram_files = get_fits_files_with_extname('ECL-DET-UBC', self.d_pars['working_in'])
attitude_files = get_fits_files_with_extname('SVO-ATT-CNV', self.d_pars['working_in'])
if shadowgram_files == []:
self.status = 25
return False
if attitude_files == []:
self.status = 26
return False
# load SVO-ATT-CNV & ECL-DET-UBC files in memory.
d_attitude = read_attitude_files(attitude_files)
d_shadowgram = read_ecl_det_ubc_file(shadowgram_files)
# check_params time coherence between event & attitude files
if check_time_coherence_evt_att(d_shadowgram['tstart'], d_attitude):
detim = DetectorImages(d_shadowgram['shadowgrams'], d_shadowgram['shadowgrams_var'],
d_shadowgram['obstime'], d_shadowgram['energy_ranges'])
att = AttitudeECLAIRs()
att.set_attitude_instru_quater(d_attitude['QPARAM'])
ptg = att.ptg_instru(deg=True)[0]
ori = np.array([att.ori_instru(deg=True)])
ptgori = np.concatenate((ptg, ori))
eclo = EclairsObservation(d_shadowgram['tstart'],
d_shadowgram['tstop'],
ptgori,
d_attitude['POSITION'],
d_attitude['ANGLE_VEL'],
self.mod_effect)
eclo.gti_compute()
self.d_data_in = {
'eclairs_observation': eclo,
'detector_images': detim
}
return True
else:
self.status = 24
return False
def _create_output_files(self):
"""Create IMAG scientific files.
Call _create_output_file_ECL_SKY_IMA and/or
_create_output_file_ECL_SOP_IMA or None functions depending
on .ini parameter file.
"""
s_logger.debug('Running _create_output_files [imag]')
if self.status != 0:
return False
product_files = self.d_pars['out_files']
if 'ECL-SKY-IMA' in product_files:
self._create_output_file_ECL_SKY_IMA()
if 'ECL-SOP-IMA' in product_files:
self._create_output_file_ECL_SOP_IMA()
return True
def _create_output_file_ECL_SKY_IMA(self):
"""Create ECL-SKY-IMA files.
"""
if self.status != 0:
return False
sky_images = self.d_data_out['sky_imgs']
dest_dir = self.d_pars['working_out']
eded = self.d_data_in['eclairs_observation'].dpix
obs_info_dict = {
'tstart': self.d_data_in['eclairs_observation'].start_time,
'tstop': self.d_data_in['eclairs_observation'].end_time,
'energy_ranges': self.d_data_in['detector_images'].energy_ranges,
'attitude': self.d_data_in['eclairs_observation'].attitude
}
creator = 'APC'
proc_id = ''
try: # save sky images
save_sky_images(sky_images, dest_dir, eded,
obs_info_dict, creator, proc_id)
except Exception:
self.status = 22
return False
s_logger.info('Successfully created ECL-SKY-IMA product')
return True
def _create_output_file_ECL_SOP_IMA(self):
"""Create ECL-SOP-IMA files.
"""
if self.status != 0:
return False
catalogs_output = self.d_data_out['catalog_sources']
dest_dir = self.d_pars['working_out']
eded = self.d_data_in['eclairs_observation'].dpix
obs_info_dict = {
'tstart': self.d_data_in['eclairs_observation'].start_time,
'tstop': self.d_data_in['eclairs_observation'].end_time,
'energy_ranges': self.d_data_in['detector_images'].energy_ranges,
'attitude': self.d_data_in['eclairs_observation'].attitude
}
creator = 'APC'
proc_id = ''
try: # save catalogs
save_catalogs(catalogs_output, dest_dir, eded,
obs_info_dict, creator, proc_id)
except Exception:
self.status = 23
return False
s_logger.info('Successfully created ECL-SOP-IMA product')
return True
def _process_composant(self):
"""Run IMAG module.
"""
s_logger.debug('Running _process_composant [imag]')
if self.status != 0:
return False
try: # run IMAG
cat_srcs, sky_imgs, new_src_flag = imag(self.d_data_in, self.d_pars)
except Exception:
s_logger.exception('IMAG Exception')
self.status = 21
return False
self.d_data_out = {
'catalog_sources': cat_srcs,
'sky_imgs': sky_imgs,
'new_src_flag': new_src_flag
}
return True
def _check_extra(self):
"""Check parameters specific to the IMAG module in parameter INI file.
"""
s_logger.debug('Running _check_extra [imag]')
return True