"""
A module for all things calibration.
"""
from datetime import datetime, timezone, timedelta
import random
import os.path
from pathlib import Path
import sys
import ccsdspy
import numpy as np
from hermes_core import log
from hermes_core.util.util import create_science_filename, parse_science_filename
import hermes_eea
from hermes_eea.io import read_file
import hermes_eea.calibration as calib
from hermes_eea.io.EEA import EEA
from hermes_eea.SkymapFactory import SkymapFactory
# cdflib -> spacepy
from spacepy.pycdf import lib
from hermes_eea.calibration.build_spectra import Hermes_EEA_Data_Processor
from astropy.time import Time
__all__ = [
"process_file",
"parse_l0_sci_packets",
"l0_sci_data_to_cdf",
"calibrate_file",
"get_calibration_file",
"read_calibration_file",
]
[docs]
def process_file(data_filename: Path) -> list:
"""
This is the entry point for the pipeline processing.
It runs all of the various processing steps required
to create a L1A Hermes CDF File
calls:
calibrate_file()
parse_science
CCSDSPY (parse_L0_sci_packets())
l0_sci_data_to_cdf()
SkymapFactory()
Use HermesData to populate CDF output file
Write the File
A Custom EEA SkymapFactory
HermesData
Parameters
----------
data_filename: str
Fully specificied filename of an input file.
The file contents:
Traditional binary packets in CCSDS format
Returns
-------
output_filenames: list
Fully specificied filenames for the output files.
The file contents:
CDF formatted file with n packets iincluding time and [41,4,32] skymap.
"""
log.info(f"Processing file {data_filename}.")
output_files = []
# Get the Directory of the File
destination_dir = data_filename.parent
# Calibrate the Input File
calibrated_file = calibrate_file(data_filename, destination_dir)
output_files.append(calibrated_file)
# Add Plots to the Output Files if we want
# data_plot_files = plot_file(data_filename)
# calib_plot_files = plot_file(calibrated_file)
# add other tasks below
return output_files
[docs]
def calibrate_file(data_filename: Path, destination_dir) -> Path:
"""
Given an input data file, raise it to the next level
(e.g. level 0 to level 1, level 1 to quicklook) it and return a new file.
Parameters
----------
data_filename: Path
Fully specificied filename of the input data file.
Returns
-------
output_filename: Path
Fully specificied filename of the output file.
"""
log.info(f"Calibrating file:{data_filename}.")
file_metadata = parse_science_filename(data_filename.name)
# check if level 0 binary file, if so call appropriate functions
if (
file_metadata["instrument"] == hermes_eea.INST_NAME
and file_metadata["level"] == "l0"
):
# call CCSDSPY to parse our packets.
data = parse_l0_sci_packets(data_filename)
level1_filename = l0_sci_data_to_cdf(data, data_filename, destination_dir)
output_filename = level1_filename
elif (
file_metadata["instrument"] == hermes_eea.INST_NAME
and file_metadata["level"] == "l1"
):
# generate the quicklook data
#
# the following shows an example flow for calibrating a file
# data = read_file(data_filename)
# calib_file = get_calibration_file(data_filename)
# if calib_file is None:
# raise ValueError(f"Calibration file for {data_filename} not found.")
# else:
# calib_data = read_calibration_file(calib_file)
# test opening the file
with open(data_filename, "r") as fp:
pass
# now that you have your calibration data, you can calibrate the science data
ql_filename = data_filename.parent / create_science_filename(
file_metadata["instrument"],
file_metadata["time"],
"ql",
file_metadata["version"],
)
# write your cdf file below
# create an empty file for testing purposes
with open(data_filename.parent / ql_filename, "w"):
pass
# here
data = parse_l0_sci_packets(data_filename)
output_filename = ql_filename
else:
raise ValueError(f"The file {data_filename} is not recognized.")
return output_filename
[docs]
def parse_l0_sci_packets(data_filename: Path) -> dict:
"""
Parse a level 0 eea binary file containing CCSDS packets.
Parameters
----------
data_filename: str
Fully specificied filename
Returns
-------
result: dict
A dictionary of arrays which includes the ccsds header fields
Examples
--------
>>> import hermes_eea.calibration as calib
>>> data_filename = "hermes_EEA_l0_2022339-000000_v0.bin"
>>> data = calib.parse_eea_sci_packets(data_filename) # doctest: +SKIP
"""
log.info(f"Parsing packets from file:{data_filename}.")
pkt = ccsdspy.FixedLength.from_file(
os.path.join(hermes_eea._data_directory, "hermes_EEA_sci_packet_def.csv")
)
data = pkt.load(data_filename)
return data
[docs]
def l0_sci_data_to_cdf(
data: dict, original_filename: Path, destination_dir: Path
) -> Path:
"""
Write level 0 eea science data to a level 1 cdf file.
Parameters
----------
data: dict
A dictionary of arrays which includes the ccsds header fields
original_filename: Path
The Path to the originating file.
Returns
-------
output_filename: Path
Fully specificied filename of cdf file
Examples
--------
>>> from pathlib import Path
>>> from hermes_core.util.util import parse_science_filename
>>> import hermes_eea.calibration as calib
>>> data_filename = Path("hermes_EEA_l0_2022339-000000_v0.bin")
>>> metadata = parse_science_filename(data_filename) # doctest: +SKIP
>>> data_packets = calib.parse_l0_sci_packets(data_filename) # doctest: +SKIP
>>> cdf_filename = calib.l0_sci_data_to_cdf(data_packets, data_filename) # doctest: +SKIP
"""
# this is transferring name.bin to name.cdf
file_metadata = parse_science_filename(original_filename.name)
cdf_filename = original_filename.parent / create_science_filename(
file_metadata["instrument"],
file_metadata["time"],
"l1",
f'1.0.{file_metadata["version"]}',
)
if data:
calibration_file = get_calibration_file(hermes_eea.stepper_table)
read_calibration_file(calibration_file)
myEEA = EEA(file_metadata)
# SkymapFactory, now as does FPI, also populates my EEA data model
SkymapFactory(data, calib.energies, calib.deflections, myEEA)
# In the beginning, testing phase of this project, while we adjust things.
# This will show us which packets have a workable amount of data
most_active = np.where(np.array(myEEA.stats) > 150)
# these eample start times are also something I would like to keep around for a while
# example_start_times = Time(
# [lib.tt2000_to_datetime(e) for e in myEEA.Epoch[0:10]]
# )
n_packets = len(myEEA.Epoch)
# https://hermes-core.readthedocs.io/en/latest/user-guide/reading_writing_data.html
# https://hermes-core.readthedocs.io/en/latest/generated/api/hermes_core.timedata.HermesData.html#hermes_core.timedata.HermesData
hermes_eea_factory = Hermes_EEA_Data_Processor(myEEA)
hermes_eea_factory.build_HermesData()
try:
# this writes out the data to CDF file format
cdf_path = hermes_eea_factory.hermes_eea_data.save(
str(destination_dir), True
)
except Exception as e:
log.error(e)
sys.exit(2)
return cdf_path
[docs]
def get_calibration_file(data_filename: Path, time=None) -> Path:
"""
Given a time, return the appropriate calibration file.
Parameters
----------
data_filename: str
Fully specificied filename of the non-calibrated file (data level < 2)
time: ~astropy.time.Time
Returns
-------
calib_filename: str
Fully specificied filename for the appropriate calibration file.
Examples
--------
"""
return os.path.join(hermes_eea._calibration_directory, data_filename)
[docs]
def read_calibration_file(calib_filename: Path):
"""
Given a calibration, return the calibration structure.
Parameters
----------
calib_filename: str
Fully specificied filename of the non-calibrated file (data level < 2)
Returns
-------
output_filename: str
Fully specificied filename of the appropriate calibration file.
Examples
--------
"""
lines = read_file(os.path.join(calib_filename))
calib.energies = []
calib.deflections = []
for line in lines:
calib.energies.append(int(line[8:10], 16))
calib.deflections.append(int(line[10:12], 16))
