diff --git a/imap_processing/spice/kernels.py b/imap_processing/spice/kernels.py index 1ff30df21..2d4e21154 100644 --- a/imap_processing/spice/kernels.py +++ b/imap_processing/spice/kernels.py @@ -1 +1,158 @@ """Functions for furnishing and tracking SPICE kernels.""" + +import functools +import logging +import os +from typing import Any, Callable, Optional + +import spiceypy as spice +from spiceypy.utils.exceptions import SpiceyError + +logger = logging.getLogger(__name__) + + +def ensure_spice( + f_py: Optional[Callable] = None, time_kernels_only: bool = False +) -> Callable: + """ + Decorator/wrapper that automatically furnishes SPICE kernels. + + Parameters + ---------- + f_py : Callable + The function requiring SPICE that we are going to wrap if being used + explicitly, otherwise None, in which case ensure_spice is being used, + not as a function wrapper (see l2a_processing.py) but as a true + decorator without an explicit function argument. + time_kernels_only : bool + Specify that we only need to furnish time kernels (if SPICE_METAKERNEL + is set, we still just furnish that metakernel and assume the time + kernels are included. + + Returns + ------- + Callable + Decorated function, with spice error handling. + + Notes + ----- + Before trying to understand this piece of code, read this: + https://stackoverflow.com/questions/5929107/decorators-with-parameters/60832711#60832711 + + **Control flow overview:** + 1. Try simply calling the wrapped function naively. + * SUCCESS? Great! We're done. + * SpiceyError? Go to step 2. + + 2. Furnish metakernel at SPICE_METAKERNEL + * SUCCESS? Great, return the original function again (so it can be + re-run). + * KeyError? Seems like SPICE_METAKERNEL isn't set, no problem. Go to + step 3. + + 3. Did we get the parameter time_kernels_only=True? + --> YES? We only need LSK and SCLK kernels to run this function. Go fetch + those and furnish and return the original function (so it can be re-run). + --> NO? Dang. This is sort of the end of the line. Re-raise the error + generated from the failed spiceypy function call but add a better + message to it. + + Examples + -------- + There are three ways to use this object + + 1. A decorator with no arguments + >>> @ensure_spice + ... def my_spicey_func(a, b): + ... pass + + 2. A decorator with parameters. This is useful + if we only need the latest SCLK and LSK kernels for the function involved. + >>> @ensure_spice(time_kernels_only=True) + ... def my_spicey_time_func(a, b): + ... pass + + 3. An explicit wrapper function, providing a dynamically set value for + parameters, e.g. time_kernels_only + >>> wrapped = ensure_spice(spicey_func, time_kernels_only=True) + ... result = wrapped(*args, **kwargs) + """ + if f_py and not callable(f_py): + raise ValueError( + f"Received a non-callable object {f_py} as the f_py argument to" + f"ensure_spice. f_py must be a callable object." + ) + + def _decorator(func: Callable[..., Callable]) -> Callable: + """ + Decorate or wrap input function depending on how ensure_spice is used. + + Parameters + ---------- + func : Callable + The function to be decorated/wrapped. + + Returns + ------- + Callable + If used as a function wrapper, the decorated function is returned. + """ + + @functools.wraps(func) + def wrapper_ensure_spice(*args: Any, **kwargs: Any) -> Any: + """ + Wrap the function that ensure_spice is used on. + + Parameters + ---------- + *args : list + The positional arguments passed to the decorated function. + **kwargs + The keyword arguments passed to the decorated function. + + Returns + ------- + Object + Output from wrapped function. + """ + try: + # Step 1. + return func( + *args, **kwargs + ) # Naive first try. Maybe SPICE is already furnished. + except SpiceyError as spicey_err: + try: + # Step 2. + metakernel_path = os.environ["SPICE_METAKERNEL"] + spice.furnsh(metakernel_path) + except KeyError: + # TODO: An additional step that was used on EMUS was to get + # a custom metakernel from the SDC API based on an input + # time range. + if time_kernels_only: + # Step 3. + # TODO: Decide if this is useful for IMAP. Possible + # implementation could include downloading + # the most recent leapsecond kernel from NAIF (see: + # https://lasp.colorado.edu/nucleus/projects/LIBSDC/repos/libera_utils/browse/libera_utils/spice_utils.py + # for LIBERA implementation of downloading and caching + # kernels) and finding the most recent IMAP clock + # kernel in EFS. + raise NotImplementedError from spicey_err + else: + raise SpiceyError( + "When calling a function requiring SPICE, we failed" + "to load a metakernel. SPICE_METAKERNEL is not set," + "and time_kernels_only is not set to True" + ) from spicey_err + return func(*args, **kwargs) + + return wrapper_ensure_spice + + # Note: This return was originally implemented as a ternary operator, but + # this caused mypy to fail due to this bug: + # https://github.com/python/mypy/issues/4134 + if callable(f_py): + return _decorator(f_py) + else: + return _decorator diff --git a/imap_processing/tests/conftest.py b/imap_processing/tests/conftest.py index 4a7c9bd76..b9adc63e3 100644 --- a/imap_processing/tests/conftest.py +++ b/imap_processing/tests/conftest.py @@ -1,7 +1,11 @@ """Global pytest configuration for the package.""" +import re + import imap_data_access +import numpy as np import pytest +import spiceypy as spice from imap_processing import imap_module_directory @@ -18,3 +22,106 @@ def _set_global_config(monkeypatch, tmp_path): @pytest.fixture(scope="session") def imap_tests_path(): return imap_module_directory / "tests" + + +# Furnishing fixtures for testing kernels +# --------------------------------------- +@pytest.fixture(autouse=True) +def _autoclear_spice(): + """Automatically clears out all SPICE remnants after every single test to + prevent the kernel pool from interfering with future tests. Option autouse + ensures this is run after every test.""" + yield + spice.kclear() + + +@pytest.fixture(scope="session") +def spice_test_data_path(imap_tests_path): + return imap_tests_path / "spice/test_data" + + +@pytest.fixture() +def furnish_test_lsk(spice_test_data_path): + """Furnishes (temporarily) the testing LSK""" + test_lsk = spice_test_data_path / "naif0012.tls" + spice.furnsh(test_lsk) + yield test_lsk + spice.kclear() + + +@pytest.fixture() +def furnish_sclk(spice_test_data_path): + """Furnishes (temporarily) the SCLK for JPSS stored in the package data directory""" + test_sclk = spice_test_data_path / "imap_sclk_0000.tsc" + spice.furnsh(test_sclk) + yield test_sclk + spice.kclear() + + +@pytest.fixture() +def use_test_metakernel(monkeypatch, spice_test_data_path): + """For the whole test session, set the SPICE_METAKERNEL environment variable + Prime the test metakernel by creating it from the template metakernel + (allows using absolute paths on any dev system)""" + + def make_metakernel_from_kernels(metakernel, kernels): + """Helper function that writes a test metakernel from a list of filenames""" + with open(metakernel, "w") as mk: + mk.writelines( + [ + "\n", + "\\begintext\n", + "\n", + "This is a temporary metakernel for imap_processing" + " unit and integration testing.\n", + "\n", + "\\begindata\n", + "\n", + "KERNELS_TO_LOAD = (\n", + ] + ) + # Put single quotes around every kernel name + kernels_with_quotes = [" '" + kern + "'" for kern in kernels] + # Add a comma and EOL to the end of each kernel path except the last. + formated_kernels = [kern + ",\n" for kern in kernels_with_quotes[0:-1]] + # Add ')' to the last kernel + formated_kernels.append(kernels_with_quotes[-1] + "\n)\n\n") + mk.writelines(formated_kernels) + + def get_test_kernels_to_load(): + """ + Helper function for grabbing a list of kernel filenames from the test + metakernel template. This is necessary in order to get absolute paths on + any system. Formats the absolute paths using the test data path fixture + value. + """ + test_metakernel = spice_test_data_path / "imap_test_metakernel.template" + kernels_to_load = [] + max_line_length = 80 + with open(test_metakernel) as mk: + for k in mk: + kernel = k.rstrip("\n").format( + **{"SPICE_TEST_DATA_PATH": str(spice_test_data_path.absolute())} + ) + while len(kernel) > 0: + if len(kernel) <= max_line_length: + kernels_to_load.append(kernel) + break + else: + slash_positions = np.array( + [m.start() for m in re.finditer("/", kernel)] + ) + stop_idx = ( + slash_positions[slash_positions < max_line_length - 1].max() + + 1 + ) + kernels_to_load.append(kernel[0:stop_idx] + "+") + kernel = kernel[stop_idx:] + return kernels_to_load + + metakernel_path = imap_data_access.config["DATA_DIR"] / "imap_2024_v001.tm" + kernels_to_load = get_test_kernels_to_load() + make_metakernel_from_kernels(metakernel_path, kernels_to_load) + monkeypatch.setenv("SPICE_METAKERNEL", str(metakernel_path)) + yield str(metakernel_path) + spice.kclear() diff --git a/tools/tests/test_data/spice/imap_sclk_0000.tsc b/imap_processing/tests/spice/test_data/imap_sclk_0000.tsc similarity index 100% rename from tools/tests/test_data/spice/imap_sclk_0000.tsc rename to imap_processing/tests/spice/test_data/imap_sclk_0000.tsc diff --git a/imap_processing/tests/spice/test_data/imap_test_metakernel.template b/imap_processing/tests/spice/test_data/imap_test_metakernel.template new file mode 100644 index 000000000..b4a5d9f82 --- /dev/null +++ b/imap_processing/tests/spice/test_data/imap_test_metakernel.template @@ -0,0 +1,2 @@ +{SPICE_TEST_DATA_PATH}/imap_sclk_0000.tsc +{SPICE_TEST_DATA_PATH}/naif0012.tls \ No newline at end of file diff --git a/tools/tests/test_data/spice/naif0012.tls b/imap_processing/tests/spice/test_data/naif0012.tls similarity index 100% rename from tools/tests/test_data/spice/naif0012.tls rename to imap_processing/tests/spice/test_data/naif0012.tls diff --git a/imap_processing/tests/spice/test_kernels.py b/imap_processing/tests/spice/test_kernels.py index 059d0f231..c8ce926ad 100644 --- a/imap_processing/tests/spice/test_kernels.py +++ b/imap_processing/tests/spice/test_kernels.py @@ -1 +1,69 @@ """Tests coverage for imap_processing/spice/kernels.py""" + +import pytest +import spiceypy as spice +from spiceypy.utils.exceptions import SpiceyError + +from imap_processing.spice import kernels + + +@kernels.ensure_spice +def single_wrap_et2utc(et, fmt, prec): + """Directly decorate a spice function with ensure_spice for use in tests""" + return spice.et2utc(et, fmt, prec) + + +@kernels.ensure_spice +def double_wrap_et2utc(et, fmt, prec): + """Decorate a spice function twice with ensure_spice for use in tests. This + simulates some decorated outer functions that call lower level functions + that are already decorated.""" + return single_wrap_et2utc(et, fmt, prec) + + +@kernels.ensure_spice(time_kernels_only=True) +def single_wrap_et2utc_tk_only(et, fmt, prec): + """Directly wrap a spice function with optional time_kernels_only set True""" + return spice.et2utc(et, fmt, prec) + + +@kernels.ensure_spice(time_kernels_only=True) +def double_wrap_et2utc_tk_only(et, fmt, prec): + """Decorate a spice function twice with ensure_spice for use in tests. This + simulates some decorated outer functions that call lower level functions + that are already decorated.""" + return single_wrap_et2utc(et, fmt, prec) + + +@pytest.mark.parametrize( + "func", + [ + single_wrap_et2utc, + single_wrap_et2utc_tk_only, + double_wrap_et2utc, + double_wrap_et2utc_tk_only, + ], +) +def test_ensure_spice_emus_mk_path(func, use_test_metakernel): + """Test functionality of ensure spice with SPICE_METAKERNEL set""" + assert func(577365941.184, "ISOC", 3) == "2018-04-18T23:24:31.998" + + +def test_ensure_spice_time_kernels(): + """Test functionality of ensure spice with timekernels set""" + wrapped = kernels.ensure_spice(spice.et2utc, time_kernels_only=True) + # TODO: Update/remove this test when a decision has been made about + # whether IMAP will use the time_kernels_only functionality and the + # ensure_spice decorator has been update. + with pytest.raises(NotImplementedError): + _ = wrapped(577365941.184, "ISOC", 3) == "2018-04-18T23:24:31.998" + + +def test_ensure_spice_key_error(): + """Test functionality of ensure spice when all branches fail""" + wrapped = kernels.ensure_spice(spice.et2utc) + # The ensure_spice decorator should raise a SpiceyError when all attempts to + # furnish a set of kernels with sufficient coverage for the spiceypy + # functions that it decorates. + with pytest.raises(SpiceyError): + _ = wrapped(577365941.184, "ISOC", 3) == "2018-04-18T23:24:31.998" diff --git a/tools/tests/unit/test_spice_examples.py b/tools/tests/unit/test_spice_examples.py index d998a3008..d47a607cd 100644 --- a/tools/tests/unit/test_spice_examples.py +++ b/tools/tests/unit/test_spice_examples.py @@ -4,6 +4,7 @@ import pytest import spiceypy as spice +from imap_processing import imap_module_directory from tools.spice.spice_examples import ( _get_particle_velocity, build_annotated_events, @@ -29,6 +30,12 @@ def kernels(kernel_directory): kernels = list_files_with_extensions( kernel_directory, [".tsc", ".tls", ".tf", ".bsp", ".ck"] ) + # Some kernels were moved into imap_processing package + kernels.extend( + list_files_with_extensions( + imap_module_directory / "tests/spice/test_data", [".tsc", ".tls"] + ) + ) return kernels