728 spice kernel furnishing decorator

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

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()

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

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"

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