"""DPCO products
"""
import logging
import os.path as osp
from inspect import currentframe as cf
from datetime import datetime as dt
import numpy as np
from astropy.time import Time
from json_product_generator.fits_product_builder import product_builder as fpb
import ecpi.common.mission.time as mt
from ecpi.common.mission.time import create_time_tag_from_tstart
logger = logging.getLogger(__name__)
def _create_ecl_gti_sdp(dpco_sdp_data, filename_with_path):
"""save level 1 gtis in fits file ECL-GTI
filename is 'ECL-GTI-' + proc_id + '.fits'
:param dpco_sdp_data: list of sources catalogs
:type dpco_sdp_data: dict
:param dir_path: PATH to the directory where to write the file
:type dir_path: string
:param creator: program that has generated the file.
:type creator: string
:param proc_id: id of the process. Default="01".
:type proc_id: string
"""
[logger.debug(f"Parsing key {key} in dpco_sdp_data with value {dpco_sdp_data[key]}.") \
for key in dpco_sdp_data.keys()]
model = 'ECL-GTI'
fpe = fpb.FitsProductEclairs(model, filename_with_path)
print(f'######################FITS_output_file_path: {fpe.fits_output_file_path} ############')
# TODO: Recover the following information :
vals = {'TSTART': dpco_sdp_data['tstart'], 'TSTOP': dpco_sdp_data['tstop'],
# 'ONTIME': dpco_sdp_data['tstop'] - dpco_sdp_data['tstart'],
'CREATOR': 'ECLAIRS-PIPELINE-v01', 'RUN_TYPE': 'QLA',
# 'SIMUID': '01',
'RA_PNT': dpco_sdp_data['radec'][0], 'DEC_PNT': dpco_sdp_data['radec'][1],
'CARD': 'LIST OF FIRST LEVEL GTIS', 'PIPLEVEL': 'DPCO',
# 'SJDSTART': dpco_sdp_data['tstart'] / 86400,
# 'SJDSTOP': dpco_sdp_data['tstop'] / 86400, 'SIMFLAG': 1,
# 'EXPOSURE': dpco_sdp_data['exposure'],
'FSCLEVEL': 'L1',
'DATE-OBS': mt.from_mjd_to_isot(dpco_sdp_data['tstart']),
'DATE-END': mt.from_mjd_to_isot(dpco_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': 'ECL-GTI-GRP',
'FSCLEVEL': 'L1'
}, 1
)
hdu_idx = 2
for gti, values in dpco_sdp_data['gtis'].items():
if values == []:
continue
name = gti.upper().replace('_', '-')
gti_type = 'GTI-' + name
# a_values = np.array(values)
logger.info(f"GTI [{gti}] found with content [{values}]")
vals = [[], [], [], []]
for i in range(len(values)):
gti_fill = {
'SJDSTART': values[i][0] / (24 * 3600),
'SJDSTOP': values[i][1] / (24 * 3600),
'START': values[i][0],
'STOP': values[i][1]
}
fpe.update_dict_data(gti_fill, hdu_idx, 'd')
print(f'Recuperamos [{vals[0]}] and [{vals[1]}]')
# We write the extension Header and Data
fpe.update_dict_data({'EXTNAME': gti_type}, hdu_idx)
# We insert this information as a row in the first extension
fpe.update_dict_data(
{
'GTI_NAME': gti_type,
'MEMBER_XTENSION': 'BINTABLE',
'MEMBER_NAME': gti_type,
'MEMBER_VERSION': 1,
'MEMBER_POSITION': hdu_idx
}, 1, 'd'
)
hdu_idx += 1
# logger.debug(f'Content of the fits dictionaries: {fpe.print_dict()}')
fpe.fill_fits_product(True, True)
return True
def _create_ecl_evt_cor_sdp(evt_info, dir_path, creator, proc_id="01"):
"""save DPCO output events in fits file ECL-EVT-COR
filename is 'ECL-EVT-COR-' + date_iso + proc_id + '.fits'
:param evt_info: list of sources catalogs
:type evt_info: dict
:param dir_path: PATH to the directory where to write the file
:type dir_path: string
:param creator: program that has generated the file.
:type creator: string
:param proc_id: id of the process. Default="01".
:type proc_id: string
"""
# unpack observation related variables
t_start = evt_info['tstart']
t_stop = evt_info['tstop']
t_elapse = t_stop - t_start
time_isot_tag = create_time_tag_from_tstart(t_start)
creation_date = Time(dt.now(), precision=0).tt.isot
# TODO: Model ECL-EVT-COR does not exist in http://saada.unistra.fr/svom/products/
model = 'ECL-EVT-COR'
# TODO: define the final filename format
filename = model + time_isot_tag + proc_id + '.fits'
fpe = fpb.FitsProductEclairs(model, osp.join(dir_path, filename))
print(f'######################FITS_output_file_path: {fpe.fits_output_file_path} ############')
# TODO: define the CARD and EXTNAME content in strings s_car end s_extname for this product
# for the moment, we set ECL-EVT-COR
s_card = 'EVT-COR'
s_extname = 'EVT-COR'
fpe.update_common_kw(
{
'TSTART': t_start,
'TSTOP': t_stop,
'TELAPSE': t_elapse,
'CREATOR': cf().f_code.co_name,
'DATE': creation_date,
'FSCLEVEL': 'L1',
'OBS_ID': '01',
'OBS_TYPE': 1,
'RUN_TYPE': 'QLA',
'RA_PNT': evt_info['radec'][0],
'DEC_PNT': evt_info['radec'][-1],
'CARD': s_card,
'PIPLEVEL': 'DPCO'
}
)
fpe.update_dict_data({'CARD': 'GROUP', 'EXTNAME': s_extname, 'FSCLEVEL': 'L1'}, 1)
# hdu_idx = 2
# TODO: Filling the data extensions
# pix_fill = {}
# fpe.update_dict_data(pix_fill, hdu_idx, 'd')
fpe.fill_fits_product(False, True)
def _create_ecl_pix_lif_sdp(pix_info, dir_path, creator, proc_id="01"):
"""save DPCO output pix-lif in fits file ECL-PIX-LIF
filename is 'ECL-PIX-LIF-' + date_iso + proc_id + '.fits'
:param pix_info: list of sources catalogs
:type pix_info: dict
:param dir_path: PATH to the directory where to write the file
:type dir_path: string
:param creator: program that has generated the file.
:type creator: string
:param proc_id: id of the process. Default="01".
:type proc_id: string
"""
# unpack observation related variables
t_start = pix_info['tstart']
t_stop = pix_info['tstop']
t_elapse = t_stop - t_start
time_isot_tag = create_time_tag_from_tstart(t_start)
creation_date = Time(dt.now(), precision=0).tt.isot
# TODO: Model ECL-PIX_LIF does not exist in http://saada.unistra.fr/svom/products/
model = 'ECL-PIX-LIF'
# TODO: define the final filename format
filename = model + time_isot_tag + proc_id + '.fits'
fpe = fpb.FitsProductEclairs(model, osp.join(dir_path, filename))
print(f'######################FITS_output_file_path: {fpe.fits_output_file_path} ############')
# TODO: define the CARD and EXTNAME content in strings s_car end s_extname for this product
# for the moment, we set PIX-LIF
s_card = 'PIX-LIF'
s_extname = 'PIX-LIF'
fpe.update_common_kw(
{
'TSTART': t_start,
'TSTOP': t_stop,
'TELAPSE': t_elapse,
'CREATOR': cf().f_code.co_name,
'DATE': creation_date,
'FSCLEVEL': 'L1',
'OBS_ID': '01',
'OBS_TYPE': 1,
'RUN_TYPE': 'QLA',
'RA_PNT': pix_info['radec'][0],
'DEC_PNT': pix_info['radec'][-1],
'CARD': s_card,
'PIPLEVEL': 'DPCO'
}
)
fpe.update_dict_data({'CARD': 'GROUP', 'EXTNAME': s_extname, 'FSCLEVEL': 'L1'}, 1)
# hdu_idx = 2
# TODO: Filling the data extensions
# pix_fill = {}
# fpe.update_dict_data(pix_fill, hdu_idx, 'd')
fpe.fill_fits_product(False, True)
DPCO_SDP = {
"ECL-EVT-COR": _create_ecl_evt_cor_sdp,
# "ECL-PIX-LIF": _create_ecl_pix_lif_sdp,
"ECL-GTI-1": _create_ecl_gti_sdp
}
[docs]def genfits_ecl_sdp_dpco(dpco_sdp_data, model, dir_path):
"""
Interface for handling the scientific data products (SDPs) for
the IMAG component
:param dpco_sdp_data: Data dictionary created by Data_Flow() containing the data
organized by gti_type
:type dpco_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 DPCO_SDP:
logger.exception(f'[{model}] is not available as scientific product in component DPCO')
return False
time_isot_tag = create_time_tag_from_tstart(dpco_sdp_data['tstart'])
file_name = f"{model}_UTC{time_isot_tag}.fits"
filename_with_path = osp.join(dir_path, file_name)
DPCO_SDP.get(model)(dpco_sdp_data, filename_with_path)
return True
# def genfits_ecl_sdp_dpco(gtis_info, dir_path, creator, proc_id="01"):
# """save level 1 gtis in fits file ECL-GTI
# filename is 'ECL-GTI-' + proc_id + '.fits'
#
# :param gtis_info: list of sources catalogs
# :type gtis_info: dict
# :param dir_path: PATH to the directory where to write the file
# :type dir_path: string
# :param creator: program that has generated the file.
# :type creator: string
# :param proc_id: id of the process. Default="01".
# :type proc_id: string
# """
# for key in gtis_info.keys():
# print(f"Parsing key {key} in gtis_info with value {gtis_info[key]}.")
#
# # unpack observation related variables
# t_start = gtis_info['tstart']
# t_stop = gtis_info['tstop']
# t_elapse = t_stop - t_start
# time_isot_tag = create_time_tag_from_t_start(t_start)
# creation_date = Time(dt.now(), precision=0).tt.isot
# model = 'ECL-GTI'
# filename = model + time_isot_tag + '_' + proc_id + '_test.fits'
# fpe = fpb.FitsProductEclairs(model, osp.join(dir_path, filename))
# print(f'######################FITS_output_file_path: {fpe.fits_output_file_path} ############')
# # TODO: Recover the following information :
# # OBS_ID, OBS_TYPE, COMMENT, HISTORY
# fpe.update_common_kw({'TSTART': t_start, 'TSTOP': t_stop, 'TELAPSE': t_elapse, 'CREATOR': cf().f_code.co_name,
# 'DATE': creation_date, 'FSCLEVEL': 'L1',
# 'OBS_ID': 1, 'OBS_TYPE': 1, 'RUN_TYPE': 'QLA',
# 'RA_PNT': gtis_info['radec'][0], 'DEC_PNT': gtis_info['radec'][-1],
# 'CARD': 'LISTS OF FIRST LEVEL GTIS', 'PIPLEVEL': 'DPCO'})
# fpe.update_dict_data({'CARD': 'GROUP', 'EXTNAME': 'ECL-GTI-GROUP', 'FSCLEVEL': 'L1'
# }, 1)
# hdu_idx = 2
# for gti, values in gtis_info['gtis'].items():
# if values == []:
# continue
# gti_fill = {}
# # TODO: Verify if it is necessary to have different names for GTI-NSA
# name = gti.upper().replace('_', '-')
# gti_name = 'GTI-' + name
# if 0 <= name.rfind('NSA'):
# gti_type = 'GTI-NSA'
# else:
# gti_type = gti_name
# a_values = np.array(values)
# logger.info(f"GTI [{gti}] found with content [{a_values}]")
# # TODO: convert in SJDTIME
# gti_fill['SJDSTART'] = a_values[:, 0] / (24 * 3600)
# gti_fill['SJDSTOP'] = a_values[:, 1] / (24 * 3600)
# gti_fill['START'] = a_values[:, 0]
# gti_fill['STOP'] = a_values[:, 1]
# # We write the extension Header and Data
# fpe.update_dict_data({'EXTNAME': gti_name}, hdu_idx)
# fpe.update_dict_data(gti_fill, hdu_idx, 'd')
# # We insert this information as a row in the first extension
# fpe.update_dict_data({'GTI_NAME': gti_type, 'MEMBER_XTENSION': 'BINTABLE',
# 'MEMBER_NAME': gti_name, 'MEMBER_VERSION': 1, 'MEMBER_POSITION': hdu_idx}, 1, 'd')
# hdu_idx += 1
#
# fpe.print_dict()
# fpe.fill_fits_product(False, False)
# return True
#