"""BUBE product
"""
import logging
from datetime import datetime as dt
import os.path as osp
import numpy as np
import astropy.time as at
from json_product_generator.fits_product_builder import product_builder as fpb
from ecpi.common.mission.time import create_time_tag_from_tstart
import ecpi.common.mission.time as mt
from ecpi.common.instru.array_convention import yzegp_to_ijdet_array
logger = logging.getLogger(__name__)
def _create_ecl_det_ubc_sdp(bube_sdp_data, filename_with_path, gti_type):
"""save detector corrected images in a fits file : ECL-DET-UBC
common keywords are used
..warning: it is assumed the global efficiency matrix is the same for each channel !
..warning: MJD to ISOT time conversion follows DC2 conventions. As a consequence, tstart
is correct in dynamic mode only.
filename is ('ECL-DET-UBC-' + proc_id + '.fits')
:param bube_sdp_data: Data dictionary created by Data_Flow() containing the data
organized by gti
:type bube_sdp_data: Dictionary
:param filename_with_path: Absolute path of the fits product to be created
:type filename_with_path: String
:param gti_type: Name of the gti for recovering the right dictionary
:type gti_type: String
:return: Final state of the creation
:rtype: Bool
"""
det_image = bube_sdp_data['det_images'][gti_type]['det_ima']
model = 'ECL-DET-UBC'
fpe = fpb.FitsProductEclairs(model, filename_with_path)
print(f'###################### FITS_output_file_path: {fpe.fits_output_file_path} ############')
# TODO: All this values must be recovered from Data_Flow in common_kws
vals = {'TSTART': bube_sdp_data['tstart'], 'TSTOP': bube_sdp_data['tstop'],
'ONTIME': bube_sdp_data['tstop'] - bube_sdp_data['tstart'],
'CREATOR': 'ECLAIRS-PIPELINE-v01',
'DATE': at.Time(dt.now(), precision=0).tt.isot, 'SIMUID': '01',
'GTISTART': bube_sdp_data['det_images'][gti_type]['gtistart'],
'GTISTOP': bube_sdp_data['det_images'][gti_type]['gtistop'],
'RA_PNT': bube_sdp_data['radec'][0], 'DEC_PNT': bube_sdp_data['radec'][1],
#'CARD': 'CORRECTED DETECTOR IMAGES',
'PIPLEVEL': 'BUBE',
'SJDSTART': bube_sdp_data['tstart'] / 86400,
'SJDSTOP': bube_sdp_data['tstop'] / 86400, 'SIMFLAG': 1,
'EXPOSURE': bube_sdp_data['det_images'][gti_type]['exposure'], 'FSCLEVEL': 'L2',
'DATE-OBS': mt.from_mjd_to_isot(bube_sdp_data['tstart']),
'DATE-END': mt.from_mjd_to_isot(bube_sdp_data['tstart'])}
fpe.update_common_kw(vals)
# fpe.update_common_kw(bube_sdp_data['common_kws'])
fpe.update_dict_data({'CARD': 'GROUP', 'EXTNAME': f'{model}-GRP', 'GRPNAME': f"{model}-GRP"}, 1)
idx_ref = fpe.image_hdu_params[0]
expected_ext = fpe._get_keyword_from_json('IMATYPE', idx_ref)['expected_values']['values']
cur_idx = idx_ref
logger.info(f"Entering in genfits_bube")
bkg_meth = det_image.bkg_cor_mod.upper()
for shad, shad_var, e_band in zip(det_image.get_shadowgrams(),
det_image.get_shadowgrams_var(),
det_image.energy_channels):
chanmin = e_band[0]
chanmax = e_band[1]
e_min = bube_sdp_data['dpix'].get_energymin_val(chanmin)
e_max = bube_sdp_data['dpix'].get_energymax_val(chanmax)
shd_bg_corr = yzegp_to_ijdet_array(shad)
shd_var = yzegp_to_ijdet_array(shad_var)
globeff = det_image.global_efficiency
list_data = (shd_bg_corr, shd_var, globeff)
for ind, imatyp in enumerate(expected_ext):
data = np.array(list_data[ind], dtype='float32')
fpe.add_image_hdu(idx_ref, cur_idx, chanmin, chanmax, imatyp, data,
E_MIN=e_min, E_MAX=e_max)
# TODO: Recover the list of bkg_meth from real data.
fpe.update_dict_data({"BKG_METH": bkg_meth, 'CARD': 'IMATYPE'}, cur_idx, 'h')
cur_idx += 1
try:
fpe.update_dict_data(fpe._add_rows_table_hdu1d(), 1, 'd')
except Exception as e:
logger.error(f"................Error : {e}")
fpe.fill_fits_product(fits_validator=True, use_check_dict=False)
return True
def _create_ecl_det_ima_sdp(bube_sdp_data, filename_with_path, gti_type):
"""Creates the SDP product ECL_DET_IMA
"""
pass
def _create_ecl_gti_sdp(bube_sdp_data, filename_with_path, gti_type):
"""Creates the SDP product ECL_GTI_3
"""
pass
BUBE_SDP = {
"ECL-DET-UBC": _create_ecl_det_ubc_sdp,
"ECL-DET-IMA": _create_ecl_det_ima_sdp,
# "ECL-GTI": _create_ecl_gti_sdp
}
[docs]def genfits_ecl_sdp_bube(bube_sdp_data, model, dir_path):
"""
Interface for handling the scientific data products (SDPs) for
the IMAG component
:param bube_sdp_data: Data dictionary created by Data_Flow() containing the data
organized by gti_type
:type bube_sdp_data: Dictionary
:param model: Model of the scientific data product to be constructed
:type model: String
:param dir_path: Absolut path of the pipeline's output folder
:type dir_path: String
:return: Final state of the function
:rtype: Bool
"""
if model not in BUBE_SDP:
logger.exception(f'[{model}] is not available as scientific product in component BUBE')
return False
time_isot_tag = create_time_tag_from_tstart(bube_sdp_data['tstart'])
for gti_type in bube_sdp_data.get('det_images').keys():
if gti_type == 'basic':
# basic gtis does not contain earth fraction information
# TODO: check the official convention for the SDPs
file_name = f"{model}_UTC{time_isot_tag}.fits"
else:
# other gtis contain earth fraction information
# TODO: check the official convention for the SDPs
file_name = f"{model}-{gti_type.upper()}_UTC{time_isot_tag}.fits"
filename_with_path = osp.join(dir_path, file_name)
BUBE_SDP.get(model)(bube_sdp_data, filename_with_path, gti_type)
return True
# TODO: Check coherence has been checked several times before
# Maybe it is not necessary
[docs]def check_time_coherence_evt_att(t_start, d_attitude):
"""Check the time consistence between event and
attitude files.
:param t_start: starting time of the observation in second.
:type t_start: float
:param d_attitude: dictionary collecting infos from attitude files.
:type d_attitude: dict
:return: whether origin of time is coherent
:rtype: boolean
"""
date_att = d_attitude['TIME_AAV']
logger.debug(f'check time coherence: t_start={t_start} | att file={date_att}')
return np.isclose(t_start, date_att, atol=1e-1)