ULTRA L1b: replace the DE product lookup with the extendedspin priority config

docs/source/conf.py

@@ -116,6 +116,7 @@
     (r"py:.*", r".*.hit.l0.utils.*"),
     (r"py:.*", r".*.hit.l0.data_classes.*"),
     (r"py:.*", r".*.hit.l1a.*"),
+    (r"py:.*", r".*ExtendedSpinConfig.*"),
     (r"py:.*", r".*InitVar*"),
     (r"py:.*", r".*.glows.utils.constants.TimeTuple.*"),
     (r"py:.*", r".*glows.utils.constants.DirectEvent.*"),

imap_processing/cli.py

@@ -2069,6 +2069,10 @@ def do_processing(
                 )
             datasets = ultra_l1a.ultra_l1a(science_files[0], create_derived_l1b=True)
         elif self.data_level == "l1b":
+            if self.repointing is None:
+                raise ValueError(
+                    "Repointing must be provided for ULTRA L1b processing."
+                )
             science_files = dependencies.get_file_paths(source="ultra", data_type="l1a")
             l1a_dict = {
                 dataset.attrs["Logical_source"]: dataset
@@ -2084,7 +2088,7 @@ def do_processing(
             ancillary_files = {}
             for path in anc_paths:
                 ancillary_files[path.stem.split("_")[2]] = path
-            datasets = ultra_l1b.ultra_l1b(combined, ancillary_files)
+            datasets = ultra_l1b.ultra_l1b(combined, ancillary_files, self.repointing)
         elif self.data_level == "l1c":
             science_files = dependencies.get_file_paths(source="ultra", data_type="l1a")
             l1a_dict = {

imap_processing/tests/ultra/data/l1/imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv

@@ -0,0 +1,6 @@
+pointing,date,pri_config,cal_config,deflector_Vthresh,cullThresh_0,cullThresh_1,cullThresh_2,cullThresh_3,cullThresh_4
+44,11/10/25,p0,c0,3400,200,7.5,4.5,3.5,3.5
+134,1/22/26,p0,c1,3400,96.5,80.5,8.5,10.5,3.5
+233,4/30/26,p1,c2,3400,200,200,35.5,14.5,8.5
+282,6/17/26,p1,c3,3400,200,200,11.5,26.5,2.5
+337,8/11/26,p1,c4,2900,117.5,73.5,2.5,26.5,8.5

imap_processing/tests/ultra/data/l1/imap_ultra_l1b-90sensor-extendedspin-config_20251001_v001.csv

@@ -0,0 +1,6 @@
+pointing,date,pri_config,cal_config,deflector_Vthresh,cullThresh_0,cullThresh_1,cullThresh_2,cullThresh_3,cullThresh_4
+44,11/10/25,p0,c0,3400,9.5,13.5,4.5,4.5,0.5
+128,1/16/26,p0,c1,3400,200,118.5,78.5,12.5,5.5
+238,5/4/26,p1,c2,3400,200,200,21.5,6.5,5.5
+282,6/17/26,p1,c3,3400,81.5,49.5,15.5,16.5,9.5
+337,8/11/26,p1,c4,3400,200,97.5,82.5,66.5,8.5

imap_processing/tests/ultra/unit/test_lookup_utils.py

@@ -1,3 +1,4 @@
+import datetime
 from unittest import mock
 
 import numpy as np
@@ -7,6 +8,7 @@
 from imap_processing import imap_module_directory
 from imap_processing.quality_flags import ImapDEOutliersUltraFlags
 from imap_processing.ultra.l1b.lookup_utils import (
+    ExtendedSpinConfig,
     get_angular_profiles,
     get_back_position,
     get_de_product_name,
@@ -222,8 +224,8 @@ def test_get_scattering_thresholds(ancillary_files):
 def test_get_de_product_name_no_repoint():
     """Tests function get_de_product_name when the lookup is missing the repoint."""
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1c-45sensor-de-product-lookup": TEST_PATH
+        / "imap_ultra_l1c-45sensor-culling-config_20251001_v001.csv"
     }
     with mock.patch(
         "imap_processing.ultra.l1b.lookup_utils.pd.read_csv"
@@ -239,14 +241,14 @@ def test_get_de_product_name_no_repoint():
             }
         )
         with pytest.raises(ValueError, match="No DE product found for repoint ID 0"):
-            get_de_product_name("repoint00000", 45, "l1b", ancillary_files)
+            get_de_product_name("repoint00000", 45, ancillary_files)
 
 
 def test_get_de_product_name_multiple_products():
     """Tests function get_de_product_name when the lookup is ambiguous."""
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1c-45sensor-de-product-lookup": TEST_PATH
+        / "imap_ultra_l1c-45sensor-culling-config_20251001_v001.csv"
     }
     with mock.patch(
         "imap_processing.ultra.l1b.lookup_utils.pd.read_csv"
@@ -262,14 +264,14 @@ def test_get_de_product_name_multiple_products():
             }
         )
         with pytest.raises(ValueError, match="Multiple DE products found"):
-            get_de_product_name("repoint00002", 45, "l1b", ancillary_files)
+            get_de_product_name("repoint00002", 45, ancillary_files)
 
 
 def test_get_de_product_name():
     """Tests function get_de_product_name."""
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1c-45sensor-de-product-lookup": TEST_PATH
+        / "imap_ultra_l1c-45sensor-culling-config_20251001_v001.csv"
     }
     with mock.patch(
         "imap_processing.ultra.l1b.lookup_utils.pd.read_csv"
@@ -288,9 +290,41 @@ def test_get_de_product_name():
         # Test with a repoint in the future. Should return the priority 2 de product
         # since the last repoint range does not have an end and should be assumed to
         # cover all future repoints.
-        de_product = get_de_product_name("repoint00100", 45, "l1b", ancillary_files)
+        de_product = get_de_product_name("repoint00100", 45, ancillary_files)
         assert de_product == "imap_ultra_l1b_45sensor-priority-2-de"
 
         # Test with valid repoint that falls in the second range.
-        de_product = get_de_product_name("repoint00003", 45, "l1b", ancillary_files)
+        de_product = get_de_product_name("repoint00003", 45, ancillary_files)
         assert de_product == "imap_ultra_l1b_45sensor-priority-1-de"
+
+
+def test_extended_config_class():
+    """Tests the ExtendedSpinConfig class."""
+    config_path = (
+        TEST_PATH / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
+    )
+    config = ExtendedSpinConfig.from_csv(config_path, "repoint00128")
+
+    assert config.priority == "p0"
+    assert config.calibration == "c0"
+    np.testing.assert_array_equal(
+        config.energy_thresholds, np.array([200, 7.5, 4.5, 3.5, 3.5, 3.5])
+    )
+    assert config.date == datetime.datetime(2025, 11, 10)
+    assert config.voltage_threshold == 3400
+
+
+def test_extended_config_class_last_repoint_range():
+    """Tests the ExtendedSpinConfig class."""
+    config_path = (
+        TEST_PATH / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
+    )
+    config = ExtendedSpinConfig.from_csv(config_path, "repoint00383")
+
+    assert config.priority == "p1"
+    assert config.calibration == "c4"
+    np.testing.assert_array_equal(
+        config.energy_thresholds, np.array([117.5, 73.5, 2.5, 26.5, 8.5, 8.5])
+    )
+    assert config.date == datetime.datetime(2026, 8, 11)
+    assert config.voltage_threshold == 2900

imap_processing/tests/ultra/unit/test_ultra_l1b.py

@@ -159,7 +159,7 @@ def test_cdf_de(
     use_fake_spin_data_for_time(511000000, 511000000 + 86400 * 5)
     use_fake_repoint_data_for_time(np.arange(511000000, 511000000 + 86400 * 5, 86400))
 
-    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
 
     assert (
         l1b_de_dataset[0].attrs["Logical_source_description"]
@@ -197,7 +197,7 @@ def test_cdf_de_flags(
     # Use repoint data that will NOT cover the event times to test flag setting
     use_fake_repoint_data_for_time(np.arange(0, +86400 * 5, 86400))
 
-    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
     # All valid events should be flagged as DURINGREPOINT since the repoint data does
     # not cover any of the event times
     valid_events = l1b_de_dataset[0]["event_times"] != FILLVAL_FLOAT32
@@ -220,6 +220,7 @@ def test_ultra_l1b_extendedspin(
         key: l1b_de_dataset
         for key in [
             "imap_ultra_l1b_45sensor-de",
+            "imap_ultra_l1b_45sensor-priority-1-de",
             "imap_ultra_l1a_45sensor-params",
         ]
     }
@@ -228,10 +229,10 @@ def test_ultra_l1b_extendedspin(
     data_dict["imap_ultra_l1b_45sensor-status"] = status_dataset
 
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1b-45sensor-extendedspin-config": TEST_PATH
+        / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
     }
-    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
 
     assert len(l1b_extendedspin_dataset) == 1
     assert (
@@ -255,6 +256,7 @@ def test_cdf_extendedspin(
         key: l1b_de_dataset
         for key in [
             "imap_ultra_l1b_45sensor-de",
+            "imap_ultra_l1b_45sensor-priority-1-de",
             "imap_ultra_l1a_45sensor-params",
         ]
     }
@@ -263,10 +265,10 @@ def test_cdf_extendedspin(
     data_dict["imap_ultra_l1b_45sensor-status"] = status_dataset
 
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1b-45sensor-extendedspin-config": TEST_PATH
+        / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
     }
-    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
     """Tests that CDF file is created and contains same attributes as xarray."""
     l1b_extendedspin_dataset[0].attrs["Data_version"] = "999"
     l1b_extendedspin_dataset[0].attrs["Repointing"] = "repoint99999"
@@ -295,6 +297,7 @@ def test_cdf_goodtimes(
         key: l1b_de_dataset
         for key in [
             "imap_ultra_l1b_45sensor-de",
+            "imap_ultra_l1b_45sensor-priority-1-de",
             "imap_ultra_l1a_45sensor-params",
         ]
     }
@@ -303,14 +306,15 @@ def test_cdf_goodtimes(
     data_dict["imap_ultra_l1b_45sensor-status"] = status_dataset
 
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1b-45sensor-extendedspin-config": TEST_PATH
+        / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
     }
-    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
 
     goodtimes_dataset = ultra_l1b(
         {"imap_ultra_l1b_45sensor-extendedspin": l1b_extendedspin_dataset[0]},
         ancillary_files,
+        "repoint99999",
     )
     goodtimes_dataset[0].attrs["Data_version"] = "999"
     goodtimes_dataset[0].attrs["Repointing"] = "repoint99999"
@@ -339,6 +343,7 @@ def test_cdf_badtimes(
         key: l1b_de_dataset
         for key in [
             "imap_ultra_l1b_45sensor-de",
+            "imap_ultra_l1b_45sensor-priority-1-de",
             "imap_ultra_l1a_45sensor-params",
         ]
     }
@@ -347,15 +352,16 @@ def test_cdf_badtimes(
     data_dict["imap_ultra_l1b_45sensor-status"] = status_dataset
 
     ancillary_files = {
-        "l1b-45sensor-de-product-lookup": TEST_PATH
-        / "imap_ultra_l1b-45sensor-de-product-lookup_20251001_v001.csv"
+        "l1b-45sensor-extendedspin-config": TEST_PATH
+        / "imap_ultra_l1b-45sensor-extendedspin-config_20251001_v001.csv"
     }
-    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_extendedspin_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
 
     ancillary_files = {}
     goodtimes_dataset = ultra_l1b(
         {"imap_ultra_l1b_45sensor-extendedspin": l1b_extendedspin_dataset[0]},
         ancillary_files,
+        "repoint99999",
     )
 
     l1b_badtimes_dataset = ultra_l1b(
@@ -364,6 +370,7 @@ def test_cdf_badtimes(
             "imap_ultra_l1b_45sensor-goodtimes": goodtimes_dataset[0],
         },
         ancillary_files,
+        "repoint99999",
     )
     l1b_badtimes_dataset[0].attrs["Data_version"] = "999"
     l1b_badtimes_dataset[0].attrs["Repointing"] = "repoint99999"
@@ -385,7 +392,7 @@ def test_ultra_l1b_error(mock_data_l1a_rates_dict):
     with pytest.raises(
         ValueError, match="Data dictionary does not contain the expected keys."
     ):
-        ultra_l1b(mock_data_l1a_rates_dict, ancillary_files)
+        ultra_l1b(mock_data_l1a_rates_dict, ancillary_files, "repoint99999")
 
 
 @pytest.mark.external_test_data
@@ -410,7 +417,7 @@ def test_ultra_l1b_priority_de(
     data_dict["imap_ultra_l1a_45sensor-priority-1-de"] = de_dataset
     data_dict[aux_dataset.attrs["Logical_source"]] = aux_dataset
 
-    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files)
+    l1b_de_dataset = ultra_l1b(data_dict, ancillary_files, "repoint99999")
 
     assert l1b_de_dataset[0]
     assert (