"""Generic component implementation for SPEX
"""
import logging
import os.path as osp
from ecpi.common import add_path_current_module
from ecpi.process.generic.component import GenericComponentProcessing
from ecpi.process.spex.io.outputs import genfits_ecl_sps_soe, \
genfits_ecl_evs_soe, genfits_ecl_lcs_soe
from ecpi.process.spex.core.main_spex import spex
import ecpi.pipeline.io.inputs as ecio
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.common.io.events import EclairsCalEvtData
from ecpi.common import add_path_tests_eclairs
s_logger = logging.getLogger(__name__)
SPEX_FILE_SCHEME = add_path_current_module(__file__, osp.join('io', 'spex_schema.json'))
[docs]class ProcessSpexWithFiles(GenericComponentProcessing):
"""
High level handling of SPEX module.
"""
def __init__(self):
"""**Constructor**
"""
s_logger.info(f"Starting SPEX module with file {SPEX_FILE_SCHEME}")
super().__init__(SPEX_FILE_SCHEME)
self.name = "spex"
# TODO: Faire la liste d"erreurs pour spex et définir le range.
self.error_range = 200
dict_comp = {
20: "error in ECL-EVS-SOE product generation",
21: "error in ECL-LCS-SOE product generation",
22: "something wrong happened during SPEX module running",
23: "python exception",
24: "output file error",
}
self.d_status.update(dict_comp)
def _load_check_file_in(self): # pragma: no cover
"""Checks for existence of ECL-EVT-CAL & ECL-SOP-IMA files
Updates self.data_flow
"""
s_logger.debug(f'Running _load_check_file_in [{self.name}]')
if not super()._load_check_file_in():
return False
# TODO: check that the list of files to be loaded are sorted and
# verified (same OBS_Id, etc.)
# check if event & attitude files exist in input working dir.
evt_files = get_fits_files_with_extname('ECL-EVT-CAL', self.d_pars['working_in'])
det_files = get_fits_files_with_extname('ECL-DET-UBC', self.d_pars['working_in'])
sop_files = get_fits_files_with_extname('ECL-SOP-IMA', self.d_pars['working_in'])
gti_files = get_fits_files_with_extname('ECL-GTI-3', self.d_pars['working_in'])
if not evt_files:
self.status = 21
return False
if not det_files:
self.status = 23
return False
events = EclairsCalEvtData()
try:
for file in evt_files:
s_logger.info(f"Reading event file {file} in [{self.name}]")
events.read(file, add_data=True)
except IOError:
s_logger.exception(f"Reading event file {file} in [{self.name}]")
self.status = 30
return False
# load detector_images and sop_catalog files into memory
try:
t_det = ecio.read_files(det_files)
t_sop = ecio.read_files(sop_files)
except IOError:
s_logger.exception("ERROR reading IMA and SOP files")
self.status = 29
return False
# set events, images and catalog into data__flow
if not self.data_flow.set_images(t_det):
s_logger.error(f'Error setting images data from {sky_sop_files} and {sop_ima_files}')
return False
if not self.data_flow.set_events(events):
s_logger.error(f'Error setting data in Events object from {evt_files}')
return False
return True
def _create_output_files(self):
"""Create ECL-EVS-SOE, ECL-SPS-SOE and ECL-LCS-SOE.
"""
s_logger.debug(f'Running _create_output_files [{self.name}]')
if self.status != 0:
return False
# # TODO: Create spex interface in data_flow and
# # genfits_... function in io/outputs.py
# creator = 'ECPI'
# out_dir = self.d_pars['working_out']
# if 'ECL-EVS-SOE' in self.d_pars['out_files']:
# s_logger.info('Creating ECL-EVS-SOE file')
# spex_info = self.data_flow.get_spex_info()
# if not genfits_ecl_evs_soe(spex_info, out_dir, creator):
# self.status = 28
# return False
# s_logger.info(f'[{self.name}] files created successfully!')
# if 'ECL-LCS-SOE' in self.d_pars['out_files']:
# s_logger.info('Creating ECL-LCS-SOE file')
# spex_info = self.data_flow.get_spex_info()
# if not genfits_ecl_lcs_soe(spex_info, out_dir, creator):
# self.status = 28
# return False
# s_logger.info(f'[{self.name}] files created successfully!')
# if 'ECL-SPS-SOE' in self.d_pars['out_files']:
# s_logger.info('Creating ECL-SPS-SOE file')
# spex_info = self.data_flow.get_spex_info()
# if not genfits_ecl_sps_soe(spex_info, out_dir, creator):
# self.status = 28
# return False
# s_logger.info(f'[{self.name}] files created successfully!')
return True
def _process_component(self):
"""Run SPEX module.
"""
if not self._print_current_info():
s_logger.error("Printing process information")
return False
try:
spex(self.data_flow, self.d_pars)
except Exception:
s_logger.exception(f"Computing [{self.name}] component")
return False
# TODO: Set spex output into dataflow
# if not self.data_flow.set_...(spex_info):
# s_logger.error("Setting spex output in data_flow")
# return False
return True
def _check_extra(self):
"""Check parameters specific to the SPEX module in parameter INI file.
"""
s_logger.debug('Running _check_extra [spex]')
if self.status != 0:
return False
return True
def _print_current_info(self):
# check time between evt, att, orb in d_data_in
# TODO: refine this function with a better way to print info
return True