"""Generic component implementation for CALI
"""
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.cali.io.outputs import genfits_ecl_sdp_cali
from ecpi.process.cali.core.main_cali import cali
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
s_logger = logging.getLogger(__name__)
CALI_FILE_SCHEME = add_path_current_module(__file__, osp.join('io', 'cali_schema.json'))
[docs]class ProcessCaliWithFiles(GenericComponentProcessing):
"""
High level handling of CALI module.
"""
def __init__(self):
"""**Constructor**
"""
s_logger.info(f"Starting CALI module with file {CALI_FILE_SCHEME}")
super().__init__(CALI_FILE_SCHEME)
self.name = "cali"
self.error_range = 200
dict_comp = {
20: "error in ECL-CALI-OUT product generation",
21: "error no SVO-ORB-CNV files found",
22: "error no ECL-EVT-SEC files found",
23: "error no SVO-ATT-CNV files found",
24: "something wrong happened during BUBE module running",
26: "starting/stopping times do not match between event/attitude/orbit files",
27: "python exception",
28: "output file error",
29: "ATT or ORB read error",
30: "ECL-EVT-SEC read 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
if self.data_flow.is_ready(self.name):
s_logger.info('Not necessary to load new data')
return True
# 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-SEC', self.d_pars['working_in'])
pvt_files = get_fits_files_with_extname('SVO-ORB-CNV', self.d_pars['working_in'])
att_files = get_fits_files_with_extname('SVO-ATT-CNV', self.d_pars['working_in'])
if not evt_files:
self.status = 22
return False
if not att_files:
self.status = 23
return False
if not pvt_files:
self.status = 21
return False
# load event files into memory.
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_att = ecio.read_files(att_files)
t_pvt = ecio.read_files(pvt_files)
except IOError:
s_logger.exception("ERROR read ATT or ORB")
self.status = 29
return False
# set events, attitude and position into data__flow
if not self.data_flow.set_aav_pvt(t_att, t_pvt, compute_eof=True):
s_logger.error(f'Error setting data in attPvt object from {att_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-EVT-CAL , GTI files.
"""
s_logger.debug(f'Running _create_output_files [{self.name}]')
if self.status != 0:
return False
response = True
for model in self.d_pars["out_files"]:
cali_sdp_data = self.data_flow.get_sdp_data(model)
if not cali_sdp_data:
s_logger.exception(f"Scientific data for {model} are missing!")
self.status = 31
response = False
continue
if not genfits_ecl_sdp_cali(cali_sdp_data, model, self.d_pars["working_out"]):
s_logger.exception(f"Creating {model} product!")
self.status = 28
response = False
s_logger.exception(f"Successfully created {model} product")
return response
def _process_component(self):
"""
Run CALI module.
"""
if not self._print_current_info():
s_logger.error("Printing process information")
return False
try:
evts_cal = cali(self.data_flow, self.d_pars)
except Exception:
s_logger.exception(f"Computing [{self.name}] component")
return False
# Checking for consistency before adding GTI and EOF into data_flow
if not self.data_flow.set_pi_vector(evts_cal):
s_logger.error("Setting cali output in data_flow")
return False
return True
def _check_extra(self):
"""
Check parameters specific to the CALI module in parameter INI file.
"""
s_logger.debug('Running _check_extra [cali]')
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