Source code for ecpi.process.dpco.dpco_with_files
"""
@author: BACON Philippe, APC/IN2P3/CNRS
"""
import logging
from astropy.time import Time, TimeDelta
import numpy as np
from ecpi.common import add_path_current_module
from ecpi.process.generic.composant import GenericComposantProcessing
from ecpi.process.bube.bube_with_files import ProcessBubeWithFiles
from ecpi.process.dpco.io.outputs import save_cal_events
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.common import add_path_current_module
from ecpi.process.generic.composant import GenericComposantProcessing
import ecpi.process.bube.io.outputs as buio
from ecpi.process.dpco.io.outputs import save_cal_events
from ecpi.common.mission.attitude import AttitudeECLAIRs
from ecpi.common.io.events import EclairsCalEvtData
from ecpi.common.mission.observation import EclairsObservation
from ecpi.common.io.fits_tools import get_fits_files_with_extname
from ecpi.common.mission.time import convert_mjd_in_svomref_seconds_ref
from ecpi.common.mission.orbit import EclairsOrbit
s_logger = logging.getLogger(__name__)
DPCO_FILE_SCHEME = add_path_current_module(__file__, "io/dpco_schema.json")
[docs]class ProcessDpcoWithFiles(GenericComposantProcessing):
"""
High level handling of DPCO module.
"""
def __init__(self):
"""**Constructor**
"""
s_logger.info("Starting DPCO module")
GenericComposantProcessing.__init__(self, DPCO_FILE_SCHEME)
self.name = "dpco"
self.error_range = 200
dict_comp = {
10: "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 times does not match between event and attitude 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
OUTPUT : self.d_data_in
"""
s_logger.debug('Running _load_check_file_in [bube]')
if not GenericComposantProcessing._load_check_file_in(self):
return False
# check if event & attitude files exist in input working dir.
event_files = get_fits_files_with_extname('ECL-EVT-SEC', self.d_pars['working_in'])
attitude_files = get_fits_files_with_extname('SVO-ATT-CNV', self.d_pars['working_in'])
orbit_files = get_fits_files_with_extname('SVO-ORB-CNV', self.d_pars['working_in'])
if event_files == []:
self.status = 22
if self.status == 0:
if attitude_files == []:
self.status = 23
if self.status == 0:
if orbit_files == []:
self.status = 21
if self.status > 0:
return False
# load event files content in memory.
events = EclairsCalEvtData()
try:
for file in event_files:
events.read(file, add_data=True)
except:
s_logger.exception(f"ERROR read file '{file}'")
self.status = 30
return False
try:
t_att = buio.read_table_svo_att_cnv(attitude_files[0])
t_orb = buio.read_table_svo_orb_cnv(orbit_files[0])
except:
s_logger.exception("ERROR read ATT or ORB")
self.status = 29
return False
self.d_data_in = {
'evt': events,
'att': t_att,
'orb': t_orb
}
return True
def _create_output_files(self):
"""Create ECL-EVT-COR , GTI files.
"""
s_logger.debug('Running _create_output_files [dpco]')
if self.status != 0:
return False
if 'ECL-EVT-COR' in self.d_pars['out_files']:
try:
s_logger.debug('No implementation to save ECL-EVT-COR')
except Exception:
self.status = 28
return False
if 'ECL-GTI-1' in self.d_pars['out_files']:
try:
s_logger.debug('No implementation to save ECL-GTI-1')
except Exception:
self.status = 28
return False
return True
def _process_composant(self):
"""Run DPCO module.
create GTI
"""
#TODO: check time between evt, att, orb in d_data_in
# GTI
#TODO: create GTI as boolean array in evt.d_gti of type pyinterval
# https://pyinterval.readthedocs.io/en/latest/guide.html
#TODO: GTI SAA with flag SAA in orb => evt.d_gti["saa"]
#TODO: GTI earth low => evt.d_gti["e_low"]
#TODO: GTI earth partial => evt.d_gti["e_part"]
#TODO: GTI earth high => evt.d_gti["e_high"]
#TODO: GTI stable pointing but definition is not clear ....
# Mean pointing for DC-2
att = AttitudeECLAIRs()
quater = self.d_data_in['att']['QPARAM'][0]
att.set_attitude_svom_quater(quater)
ptg = att.ptg_instru(deg=True)[0]
ori = np.array([att.ori_instru(deg=True)])
ptgori = np.concatenate((ptg, ori))
# output DPCO, GTI in evt key
self.d_data_out = self.d_data_in
self.d_data_out['ptgori_mean'] = ptgori
# # load attitude files content in memory.
# t_start = Time(self.d_pars['t_start']).mjd
# t_stop = t_start + TimeDelta(self.d_pars['t_exposure']).value / 24 / 3600.
#
# # set time reference
# t_ref = Time("2017-01-01T00:00:00.000").mjd
# t_start = convert_mjd_in_svomref_seconds_ref(t_start, t_ref)
# t_stop = convert_mjd_in_svomref_seconds_ref(t_stop, t_ref)
#
# # check_params time coherence between event & attitude files
# if buio.check_time_coherence_evt_att(t_start, d_attitude):
# att = AttitudeECLAIRs()
# att.set_attitude_instru_quater(d_attitude['QPARAM'])
#
# ptg = att.ptg_instru(deg=True)[0]
# ori = np.array([att.ori_instru(deg=True)])
# ptgori = np.concatenate((ptg, ori))
#
# ecl_orb = EclairsOrbit(
# orbit_filename=orbit_files[0],
# attitude_filename=attitude_files[0])
#
#
# s_logger.debug('Running _process_composant [dpco]')
# if self.status != 0:
# return False
# # Create GTI in list_events
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