"""CALI products
"""
import logging
import os.path as osp
# import numpy as np
from datetime import datetime as dt
from astropy.time import Time
# from inspect import currentframe as cf
from json_product_generator.fits_product_builder import product_builder as fpb
from ecpi.common.mission.time import create_time_tag_from_tstart
logger = logging.getLogger(__name__)
def _create_ecl_evt_cal(cali_sdp_data, filename_with_path):
"""save level 1 cali in fits file ECL-EVT-CAL
filename is "ECL-EVT-CAL-" + time_isot_tag + proc_id + ".fits"
:param cali_sdp_data: list of sources catalogs
:type cali_sdp_data: dict
:param filename_with_path: PATH to the directory where to write the file
:type filename_with_path: string
"""
for key in cali_sdp_data.keys():
print(f"Parsing key {key} in cali_sdp_data with value {cali_sdp_data[key]}.")
creation_date = Time(dt.now(), precision=0).tt.isot
model = "ECL-EVT-CAL"
logger.info(f'Constructing {model} scientific product')
return True
fpe = fpb.FitsProductEclairs(model, filename_with_path)
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(cali_sdp_data["common_kw"])
# header of data
fpe.update_dict_data({"CARD": "GROUP", "EXTNAME": f"{model}-GRP", "FSCLEVEL": "L1"}, 1)
# We insert this information as a row in the first extension
fpe.update_dict_data(
{
"TIME": creation_date,
"JDET": cali_sdp_data["JDET"],
"IDET": cali_sdp_data["IDET"],
"PHA": cali_sdp_data["PHA"],
"PI": cali_sdp_data["PI"],
"FLAG": "pixel_status"
}, 1, "d"
)
fpe.print_dict()
fpe.fill_fits_product(False, False)
return True
def _create_ecl_gti_2(cali_sdp_data, filename_with_path):
"""
:param cali_sdp_data:
:type: cali_sdp_data:
:param filename_with_path:
:type: filename_with_path:
:return:
"""
pass
CALI_SDP = {
"ECL-GTI-2": _create_ecl_gti_2,
"ECL-EVT-CAL": _create_ecl_evt_cal
}
[docs]def genfits_ecl_sdp_cali(cali_sdp_data, model, dir_path):
"""
Interface for handling the scientific data products (SDPs) for
the CALI component
:param cali_sdp_data: Data dictionary created by Data_Flow()
:type cali_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 CALI_SDP:
logger.exception(f'[{model}] is not available as scientific product in component CALI')
return False
time_isot_tag = create_time_tag_from_tstart(cali_sdp_data['tstart'])
file_name = f"{model}_UTC{time_isot_tag}.fits"
filename_with_path = osp.join(dir_path, file_name)
return CALI_SDP.get(model)(cali_sdp_data, filename_with_path)