"""Generic component implementation for MOSA
"""
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.mosa.io.outputs import genfits_ecl_sky_mos, genfits_ecl_sop_mos
from ecpi.process.mosa.core.main_mosa import mosa
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__)
MOSA_FILE_SCHEME = add_path_current_module(__file__, osp.join('io', 'mosa_schema.json'))
[docs]class ProcessMosaWithFiles(GenericComponentProcessing):
"""
High level handling of MOSA module.
"""
def __init__(self):
"""**Constructor**
"""
s_logger.info(f"Starting MOSA module with file {MOSA_FILE_SCHEME}")
super().__init__(MOSA_FILE_SCHEME)
self.name = "mosa"
# TODO: Faire la liste d"erreurs pour mosa et définir le range.
self.error_range = 200
dict_comp = {
20: "error in ECL-SKY-MOS product generation",
21: "error in ECL-SOP-MOS product generation",
22: "something wrong happened during MOSA 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-sEC & SVO-ATT-CNV 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.
sky_img_files = get_fits_files_with_extname('ECL-SKY-IMA', self.d_pars['working_in'])
sop_img_files = get_fits_files_with_extname('ECL-SOP-IMA', self.d_pars['working_in'])
if not sky_img_files:
self.status = 21
return False
if not sop_img_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 attitude and position files into memory
try:
t_sop = ecio.read_files(sky_sop_files)
t_ima = ecio.read_files(sky_ima_files)
except IOError:
s_logger.exception("ERROR reading IMA and SOP files")
self.status = 29
return False
# set events, attitude and position into data__flow
if not self.data_flow.set_images(t_sop, t_sky):
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-SKY-MOS and ECL-SOP-MOS.
"""
s_logger.debug(f'Running _create_output_files [{self.name}]')
if self.status != 0:
return False
# # TODO: Create mosa interface in data_flow and
# # genfits_... function in io/outputs.py
# creator = 'ECPI'
# out_dir = self.d_pars['working_out']
# if 'ECL-SKY-MOS' in self.d_pars['out_files']:
# s_logger.info('Creating ECL-SKY-MOS file')
# mosa_info = self.data_flow.get_mosa_info()
# if not genfits_ecl_sky_mos(mosa_info, out_dir, creator):
# self.status = 28
# return False
# s_logger.info(f'[{self.name}] files created successfully!')
# if 'ECL-SOP-MOS' in self.d_pars['out_files']:
# s_logger.info('Creating ECL-SOP-MOS file')
# mosa_info = self.data_flow.get_mosa_info()
# if not genfits_ecl_sop_mos(mosa_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 MOSA module.
"""
if not self._print_current_info():
s_logger.error("Printing process information")
return False
try:
mosa(self.data_flow, self.d_pars)
except Exception:
s_logger.exception(f"Computing [{self.name}] component")
return False
# TODO: Set mosa output into dataflow
# if not self.data_flow.set_...(mosa_info):
# s_logger.error("Setting mosa output in data_flow")
# return False
return True
def _check_extra(self):
"""Check parameters specific to the MOSA module in parameter INI file.
"""
s_logger.debug('Running _check_extra [mosa]')
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