diff --git a/.gitignore b/.gitignore index c5c58fd..b133edb 100644 --- a/.gitignore +++ b/.gitignore @@ -149,3 +149,35 @@ docs/api docs/whatsnew/latest_changelog.txt hermes_spani/version.py htmlcov/ + +#mac + +# General +.DS_Store +.AppleDouble +.LSOverride + +# Icon must end with two \r +Icon + + +# Thumbnails +._* + +# Files that might appear in the root of a volume +.DocumentRevisions-V100 +.fseventsd +.Spotlight-V100 +.TemporaryItems +.Trashes +.VolumeIcon.icns +.com.apple.timemachine.donotpresent + +# Directories potentially created on remote AFP share +.AppleDB +.AppleDesktop +Network Trash Folder +Temporary Items +.apdisk + +_version.py \ No newline at end of file diff --git a/docs/conf.py b/docs/conf.py index a969810..eeae3ce 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -39,7 +39,6 @@ "sphinx.ext.mathjax", "sphinx_automodapi.automodapi", "sphinx_automodapi.smart_resolver", - "sphinx_changelog", ] # Set automodapi to generate files inside the generated directory diff --git a/hermes_spani/__init__.py b/hermes_spani/__init__.py index 8fcc90b..481af78 100644 --- a/hermes_spani/__init__.py +++ b/hermes_spani/__init__.py @@ -1,5 +1,8 @@ # Licensed under Apache License v2 - see LICENSE.rst +import os.path + from hermes_core import log +from hermes_spani.io.file_tools import read_file try: from ._version import version as __version__ @@ -7,17 +10,16 @@ except ImportError: __version__ = "unknown version" version_tuple = (0, 0, "unknown version") -from hermes_spani.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 = "spani" +INST_SHORTNAME = "spn" +INST_TARGETNAME = "SPANI" +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_spani version: {__version__}") +log.info(f"hermes_spani version: {__version__}") diff --git a/hermes_spani/calibration/calibration.py b/hermes_spani/calibration/calibration.py index 2d67169..af8682f 100644 --- a/hermes_spani/calibration/calibration.py +++ b/hermes_spani/calibration/calibration.py @@ -2,53 +2,205 @@ A module for all things calibration. """ import random -from hermes_spani 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_spani +from hermes_spani.io import read_file + +__all__ = [ + "process_file", + "parse_l0_sci_packets", + "l0_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_filename: str - Fully specificied filename of the non-calibrated file (data level < 2) + 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: Path + Fully specificied filename of the input data file. + + Returns + ------- + output_filename: Path + Fully specificied filename of the output file. Examples -------- + >>> from hermes_spani.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_spani.INST_NAME + and file_metadata["level"] == "l0" + ): + # data = parse_l0_sci_packets(data_filename) + data = {} + # test opening the file + with open(data_filename, "r") as fp: + pass + level1_filename = l0_sci_data_to_cdf(data, data_filename) + output_filename = level1_filename + elif ( + file_metadata["instrument"] == hermes_spani.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 parse_l0_sci_packets(data_filename: Path) -> dict: + """ + Parse a level 0 spani 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_spani.calibration as calib + >>> data_filename = "hermes_MAG_l0_2022339-000000_v0.bin" + >>> data = calib.parse_spani_sci_packets(data_filename) # doctest: +SKIP + """ + log.info(f"Parsing packets from file:{data_filename}.") + data = {} + # pkt = ccsdspy.FixedLength.from_file( + # os.path.join(hermes_spani._data_directory, "SPANI_sci_packet_def.csv") + # ) + # data = pkt.load(data_filename) + return data + + +def l0_sci_data_to_cdf(data: dict, original_filename: Path) -> Path: + """ + Write level 0 spani 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_spani.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_l0_sci_packets(data_filename) # doctest: +SKIP + >>> cdf_filename = calib.l0_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, time=None): + +def get_calibration_file(data_filename: Path, time=None) -> Path: """ Given a time, return the appropriate calibration file. @@ -69,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. @@ -86,7 +238,4 @@ def read_calibration_file(calib_filename): Examples -------- """ - - # if can't read the file - return None diff --git a/hermes_spani/io/__init__.py b/hermes_spani/io/__init__.py index e69de29..ff7f1fd 100644 --- a/hermes_spani/io/__init__.py +++ b/hermes_spani/io/__init__.py @@ -0,0 +1 @@ +from .file_tools import * diff --git a/hermes_spani/tests/test_calibration.py b/hermes_spani/tests/test_calibration.py index ead6805..95759a8 100644 --- a/hermes_spani/tests/test_calibration.py +++ b/hermes_spani/tests/test_calibration.py @@ -1,14 +1,82 @@ import pytest +import os.path +from pathlib import Path + import hermes_spani.calibration as calib +from hermes_core.util.util import create_science_filename, parse_science_filename + +level0_filename = "hermes_SPANI_l0_2022339-000000_v0.bin" +level1_filename = "hermes_spn_l1_20221205_000000_v1.0.0.cdf" +ql_filename = "hermes_spn_ql_20221205_000000_v1.0.0.cdf" + + +@pytest.fixture(scope="session") +def level0_file(tmp_path_factory): + fn = tmp_path_factory.mktemp("data") / level0_filename + with open(fn, "w"): + pass + return fn + + +@pytest.fixture(scope="session") +def level1_file(tmp_path_factory): + fn = tmp_path_factory.mktemp("data") / level1_filename + with open(fn, "w"): + pass + return fn + + +def test_l0_sci_data_to_cdf(level0_file): + """Test that the output filenames are correct and that a file was actually created.""" + data = {} + output_file = calib.l0_sci_data_to_cdf(data, level0_file) + assert output_file.name == level1_filename + assert output_file.is_file() + + +def test_calibrate_file_nofile_error(): + """Test that if file does not exist it produces the correct error. The file needs to be in the correct format.""" + with pytest.raises(FileNotFoundError): + calib.calibrate_file(Path("hermes_SPANI_l0_2032339-000000_v0.bin")) + + +def test_process_file_nofile_error(): + """Test that if file does not exist it produces the correct error. The file needs to be in the correct format.""" + with pytest.raises(FileNotFoundError): + calib.process_file(Path("hermes_SPANI_l0_2032339-000000_v0.bin")) + + +def test_calibrate_file(level0_file, level1_file): + """Test that the output filenames are correct and that a file was actually created.""" + output_file = calib.calibrate_file(level0_file) + assert output_file.name == level1_filename + assert output_file.is_file() + output_file = calib.calibrate_file(level1_file) + assert output_file.name == ql_filename + assert output_file.is_file() + + # with pytest.raises(ValueError) as excinfo: + # calib.calibrate_file("datafile_with_no_calib.cdf") + # assert ( + # str(excinfo.value) + # == "Calibration file for datafile_with_no_calib.cdf not found." + # ) + + +def test_process_file_level0(level0_file): + """Test that the output filenames are correct and that a file was actually created.""" + file_output = calib.process_file(level0_file) + assert len(file_output) == 1 + assert file_output[0].name == level1_filename + assert file_output[0].is_file() -def test_calibrate_file(): - with pytest.raises(ValueError) as excinfo: - calib.calibrate_file("datafile_with_no_calib.cdf") - assert ( - str(excinfo.value) - == "Calibration file for datafile_with_no_calib.cdf not found." - ) +def test_process_file_level1(level1_file): + """Test that the output filenames are correct and that a file was actually created.""" + file_output = calib.process_file(level1_file) + assert len(file_output) == 1 + assert file_output[0].name == ql_filename + assert file_output[0].is_file() def test_get_calibration_file(): diff --git a/pyproject.toml b/pyproject.toml index 5132e0f..0df2b79 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ dependencies = [ 'astropy>=4.1.0', 'numpy>=1.16.0', 'hermes_core @ git+https://github.com/HERMES-SOC/hermes_core/', + 'ccsdspy @ git+https://github.com/ddasilva/ccsdspy.git' ] [project.optional-dependencies] @@ -97,3 +98,5 @@ omit = [ '*/hermes_spani/version*', ] +[tool.black] +extend-exclude = '''hermes_spani/_version.py''' \ No newline at end of file