"""Generic component implementation for IMAG
"""
import logging
import os.path as osp
from ecpi.common import add_path_current_module
from ecpi.process.imag.core.main_imag import imag
from ecpi.process.imag.io.outputs import genfits_ecl_sdp_ima
from ecpi.process.generic.component import GenericComponentProcessing
from ecpi.pipeline.io.inputs import read_files
from ecpi.process.bube.io.outputs import check_time_coherence_evt_att
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.process.imag.io.inputs import load_det_images_from_files
s_logger = logging.getLogger(__name__)
IMAG_FILE_SCHEME = add_path_current_module(__file__, osp.join("io", "imag_schema.json"))
[docs]class ProcessImagWithFiles(GenericComponentProcessing):
"""
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 ke_load_check_file_iny
"detector_images": DetectorImages (shadowgrams, obs times, nrj ranges)
"pvtatt" : EclairsObservation avec SVO_ATT-CNV...
"""
s_logger.info('Running _load_check_file_in [imag]')
if not super()._load_check_file_in():
s_logger.error('Error checking parent working_in directory')
s_logger.error(f'Status {self.status} : {self.d_status.get(self.status)}')
return False
if self.data_flow.is_ready(self.name):
s_logger.info('Not necessary to load new data')
return True
# check for ECL-DET-UBC and attitude files in working_in dir.
shd_files = get_fits_files_with_extname('ECL-DET-UBC', self.d_pars['working_in'])
aav_files = get_fits_files_with_extname('SVO-ATT-CNV', self.d_pars['working_in'])
pvt_files = get_fits_files_with_extname('SVO-ORB-CNV', self.d_pars['working_in'])
# check files in
if not shd_files:
self.status = 25
return False
if not aav_files:
self.status = 26
return False
# load SVO-ATT-CNV & ECL-DET-UBC files in memory.
t_att = read_files(aav_files)
t_pvt = read_files(pvt_files)
d_det_images = load_det_images_from_files(shd_files)
# check_params time coherence between event & attitude files
if not check_time_coherence_evt_att(d_det_images['tstart'], t_att):
self.status = 24
return False
self.data_flow.set_att_pvt(t_att, t_pvt)
self.data_flow.set_det_images(d_det_images)
return True
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
response = True
for model in self.d_pars['out_files']:
cat_ima_data = self.data_flow.get_sdp_data(model)
if not genfits_ecl_sdp_ima(cat_ima_data, model, self.d_pars['working_out']):
s_logger.exception(f'Creating {model} product!')
self.status = 23
response = False
s_logger.info(f'Successfully created {model} product')
return response
def _process_component(self):
"""Run IMAG module.
"""
s_logger.debug(f'Running _process_component [{self.name}] with status {self.status}')
if self.status != 0:
return False
try:
# running IMAG
d_sky_images_by_gti = imag(self.data_flow, self.d_pars)
except Exception:
s_logger.exception(f'{Exception} processing main_imag component')
self.status = 21
return False
# Storing imag output into data_flow
s_logger.info(f'Stocking catalogs in data_flow [{self.component_name}]')
return self.data_flow.set_imag_output(d_sky_images_by_gti)
def _check_extra(self):
"""Check parameters specific to the IMAG module in parameter INI file.
"""
s_logger.debug('Running _check_extra [imag]')
return True