Added process_file function

hermes_nemisis/__init__.py

@@ -1,21 +1,23 @@
 # Licensed under Apache License v2 - see LICENSE.rst
+import os.path
+
 from hermes_core import log
+from hermes_nemisis.io.file_tools import read_file
 
 try:
     from .version import __version__
 except ImportError:
     __version__ = "unknown"
-from hermes_nemisis.io.file_tools import read_file
 
-# from hermes_core.util.config import load_config, print_config
-# from hermes_core.util.logger import _init_log
+__all__ = ["log", "read_file"]
 
-# Load user configuration
-# config = load_config()
+INST_NAME = "nemisis"
+INST_SHORTNAME = "nms"
+INST_TARGETNAME = "MAG"
+INST_TO_SHORTNAME = {INST_NAME: INST_SHORTNAME}
+INST_TO_TARGETNAME = {INST_NAME: INST_TARGETNAME}
 
-# log = _init_log(config=config)
+_package_directory = os.path.dirname(os.path.abspath(__file__))
+_data_directory = os.path.abspath(os.path.join(_package_directory, "data"))
 
-# Then you can be explicit to control what ends up in the namespace,
-# __all__ = ["config", "print_config", "do_primes"]
-# __all__ = ["read_file"]
-log.debug(f"hermes_nemesis version: {__version__}")
+log.info(f"hermes_nemesis version: {__version__}")

hermes_nemisis/calibration/calibration.py

@@ -2,52 +2,205 @@
 A module for all things calibration.
 """
 import random
-from hermes_nemisis import log
+import os.path
+from pathlib import Path
 
-__all__ = ["calibrate_file", "get_calibration_file", "read_calibration_file"]
+import ccsdspy
 
+from hermes_core import log
+from hermes_core.util.util import create_science_filename, parse_science_filename
 
-def calibrate_file(data_filename, output_level=2):
+import hermes_nemisis
+from hermes_nemisis.io import read_file
+
+__all__ = [
+    "process_file",
+    "parse_nemisis_sci_packets",
+    "nemisis_sci_data_to_cdf",
+    "calibrate_file",
+    "get_calibration_file",
+    "read_calibration_file",
+]
+
+
+def process_file(data_filename: Path) -> list:
     """
-    Given an input file, calibrate it and return a new file.
+    This is the entry point for the pipeline processing.
+    It runs all of the various processing steps required.
 
     Parameters
     ----------
     data_filename: str
-        Fully specificied filename of the non-calibrated file (data level < 2)
-    output_level: int
-        The requested data level of the output file.
+        Fully specificied filename of an input file
+
+    Returns
+    -------
+    output_filenames: list
+        Fully specificied filenames for the output files.
+    """
+    log.info(f"Processing file {data_filename}.")
+    output_files = []
+
+    calibrated_file = calibrate_file(data_filename)
+    output_files.append(calibrated_file)
+    #  data_plot_files = plot_file(data_filename)
+    #  calib_plot_files = plot_file(calibrated_file)
+
+    # add other tasks below
+    return output_files
+
+
+def calibrate_file(data_filename: Path) -> 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: str
+        Fully specificied filename of the input data file.
 
     Returns
     -------
     output_filename: str
-        Fully specificied filename of the non-calibrated file (data level < 2)
+        Fully specificied filename of the output file.
 
     Examples
     --------
+    >>> from hermes_nemisis.calibration import calibrate_file
+    >>> level1_file = calibrate_file('hermes_MAG_l0_2022239-000000_v0.bin')  # doctest: +SKIP
     """
-    # example log messages
-    log.info(
-        "Despiking removing {num_spikes} spikes".format(
-            num_spikes=random.randint(0, 10)
-        )
+    log.info(f"Calibrating file:{data_filename}.")
+    output_filename = (
+        data_filename  # TODO: for testing, the output filename MUST NOT same as input
     )
-    log.warning(
-        "Despiking could not remove {num_spikes}".format(
-            num_spikes=random.randint(1, 5)
+    file_metadata = parse_science_filename(data_filename.name)
+
+    # check if level 0 binary file, if so call appropriate functions
+    if (
+        file_metadata["instrument"] == hermes_nemisis.INST_NAME
+        and file_metadata["level"] == "l0"
+    ):
+        #  data = parse_nemisis_sci_packets(data_filename)
+        data = {}
+        # test opening the file
+        with open(data_filename, "r") as fp:
+            pass
+        level1_filename = nemisis_sci_data_to_cdf(data, data_filename)
+        output_filename = level1_filename
+    elif (
+        file_metadata["instrument"] == hermes_nemisis.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"],
         )
-    )
 
-    calib_file = get_calibration_file(data_filename)
-    if calib_file is None:
-        raise ValueError("Calibration file for {} not found.".format(data_filename))
+        # write your cdf file below
+        # create an empty file for testing purposes
+        with open(data_filename.parent / ql_filename, "w"):
+            pass
+
+        # example log messages
+        log.info(f"Despiking removing {random.randint(0, 10)} spikes")
+        log.warning(f"Despiking could not remove {random.randint(1, 5)}")
+        output_filename = ql_filename
     else:
-        calib_data = read_calibration_file(calib_file)
+        raise ValueError(f"The file {data_filename} is not recognized.")
 
-    return None
+    return output_filename
 
 
-def get_calibration_file(data_filename, time=None):
+def parse_nemisis_sci_packets(data_filename: Path) -> dict:
+    """
+    Parse a level 0 nemisis 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_nemisis.calibration as calib
+    >>> data_filename = "hermes_MAG_l0_2022339-000000_v0.bin"
+    >>> data = calib.parse_nemisis_sci_packets(data_filename)  # doctest: +SKIP
+    """
+    log.info(f"Parsing packets from file:{data_filename}.")
+
+    pkt = ccsdspy.FixedLength.from_file(
+        os.path.join(hermes_nemisis._data_directory, "MAG_sci_packet_def.csv")
+    )
+    data = pkt.load(data_filename)
+    return data
+
+
+def nemisis_sci_data_to_cdf(data: dict, original_filename: Path) -> Path:
+    """
+    Write level 0 nemisis 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_nemisis.calibration as calib
+    >>> data_filename = Path("hermes_MAG_l0_2022339-000000_v0.bin")
+    >>> metadata = parse_science_filename(data_filename)  # doctest: +SKIP
+    >>> data_packets = calib.parse_nemisis_sci_packets(data_filename)  # doctest: +SKIP
+    >>> cdf_filename = calib.nemisis_sci_data_to_cdf(data_packets, data_filename)  # doctest: +SKIP
+    """
+    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"]}',
+    )
+
+    # create an empty file for testing purposes
+    with open(cdf_filename, "w"):
+        pass
+
+    return cdf_filename
+
+
+def get_calibration_file(data_filename: Path, time=None) -> Path:
     """
     Given a time, return the appropriate calibration file.
 
@@ -68,7 +221,7 @@ def get_calibration_file(data_filename, time=None):
     return None
 
 
-def read_calibration_file(calib_filename):
+def read_calibration_file(calib_filename: Path):
     """
     Given a calibration, return the calibration structure.
 
@@ -85,7 +238,4 @@ def read_calibration_file(calib_filename):
     Examples
     --------
     """
-
-    # if can't read the file
-
     return None

hermes_nemisis/data/MAG_sci_packet_def.csv

@@ -0,0 +1,28 @@
+name, bit_length, data_type, note
+SECONDS,32,UINT, Tlm Secondary Header - Seconds
+SUBSECS,16,UINT, Tlm Secondary Header - Subseconds
+SYNC,16,UINT,uint16_t sync
+PKT_ID,32,UINT,uint32_t pkt_id_num
+ACQ_POS_NUM,8,UINT,uint8_t acq_pos_num
+XSUM,1,UINT, 0 disagree, 1 agree
+START_FLAG,1,UINT, 0 disagree, 1 agree
+UNUSED,4,UINT, op status 
+OPS_MODE,1,UINT, 0 Normal, 1 - Safe
+RELAY_STATE,1,UINT, 0 disengaged, 1 engaged
+RTC_HUNDRETH,8,UINT,uint8_t rtc_seconds
+RTC_SECONDS,8,UINT,uint8_t rtc_seconds
+RTC_MINUTES,8,UINT,uint8_t rtc_minutes
+RTC_HOURS,8,UINT,uint8_t rtc_hours
+RTC_DAY,8,UINT,int8_t rtc_day
+RTC_MONTH,8,UINT,uint8_t rtc_month
+RTC_YEAR,8,UINT,uint8_t rtc_year
+MET_SECONDS,32,UINT,int32_t met_seconds
+SCTF_SECONDS,32,UINT,uint32_t stcf_seconds
+SCTF_SUBSECONDS,32,UINT,uint32_t stcf_subseconds
+MAG_DATA, 8, UINT(1080), uint8_t mag_data[1080]
+BOX_TEMP,16,UINT,uint16_t box_temp
+FLUXGATE_TEMP,16,UINT,uint16_t fluxgate_temp
+MIS1_TEMP,16,UINT,uint16_t mis1_temp
+MIS2_TEMP,16,UINT,uint16_t mis2_temp
+CHECKSUM,16,UINT,uint16_t chksum
+SPARE,8,UINT,int8 spare

hermes_nemisis/io/__init__.py

@@ -0,0 +1 @@
+from .file_tools import *