"""Generic component implementation for DPCO
"""
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.dpco.io.outputs import genfits_ecl_sdp_dpco
from ecpi.process.dpco.core.main_dpco import dpco
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__)
DPCO_FILE_SCHEME = add_path_current_module(__file__, osp.join('io', 'dpco_schema.json'))
[docs]class ProcessDpcoWithFiles(GenericComponentProcessing):
"""
High level handling of DPCO module.
"""
def __init__(self):
"""**Constructor**
"""
s_logger.info(f"Starting DPCO module with file {DPCO_FILE_SCHEME}")
super().__init__(DPCO_FILE_SCHEME)
self.name = "dpco"
self.error_range = 200
dict_comp = {
20: "error in ECL-EVT-COR 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
# 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'])
aav_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 aav_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_aav = ecio.read_files(aav_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_aav, t_pvt):
s_logger.error(f'Error setting data in attPvt object from {aav_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-COR , 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']:
dpco_sdp_data = self.data_flow.get_sdp_data(model)
if not genfits_ecl_sdp_dpco(dpco_sdp_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 DPCO module.
"""
if not self._print_current_info():
s_logger.error("Printing process information")
return False
try:
d_gtis, d_eo_gtis, t_eof = dpco(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 len(t_eof) != self.data_flow.get_aav_size():
s_logger.error("Inconsistent list of occultation fractions :")
s_logger.error(f"\t{len(t_eof)}/{self.data_flow.get_aav_size()} elements.")
return False
if not self.data_flow.set_eof(t_eof):
s_logger.error("Setting Earth occultation fraction in data_flow")
return False
if not self.data_flow.compute_ptgori_mean():
s_logger.error("Computing the pointing mean attitude")
return False
if not self.data_flow.set_basic_gtis(d_gtis):
s_logger.error("Setting basic GTIs in data_flow")
return False
if not self.data_flow.set_earth_gtis(d_eo_gtis):
s_logger.error("Setting Earth GTIs in data_flow")
return False
return True
def _check_extra(self):
"""Check parameters specific to the DPCO module in parameter INI file.
"""
s_logger.debug('Running _check_extra [dpco]')
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
#if self.data_flow.evt:
# tstart_evt = self.data_flow.evt.tb_evts['TIME'][0]
# tstop_evt = self.data_flow.evt.tb_evts['TIME'][-1]
# s_logger.info(f"EclairsCalEvtData object tstart/tstop: {tstart_evt} / {tstop_evt}")
tstart_aav_pvt = self.data_flow.aav_pvt.data['TIME_AAV'][0]
tstop_aav_pvt = self.data_flow.aav_pvt.data['TIME_AAV'][-1]
s_logger.info(f"SVO-ATT-CNV table tstart/tstop: {tstart_aav_pvt} / {tstop_aav_pvt}")
return True