MNT: Move IMAP Lo star sensor l1b back into l1a

Author: greglucas

imap_processing/lo/l0/lo_star_sensor.py

@@ -14,9 +14,9 @@
 logger.setLevel(logging.INFO)
 
 
-def unpack_star_sensor(ds: xr.Dataset) -> xr.Dataset:
+def process_star_sensor(ds: xr.Dataset) -> xr.Dataset:
     """
-    Unpack Lo star sensor data.
+    Process Lo star sensor data.
 
     Parameters
     ----------
@@ -34,37 +34,15 @@ def unpack_star_sensor(ds: xr.Dataset) -> xr.Dataset:
     buffer = b"".join(ds["data_compressed"].values)
     data = np.frombuffer(buffer, dtype=np.uint8).reshape(-1, 720)
 
+    # Decompress from 8 -> 12 bits using the decompression tables
+    decompression = DECOMPRESSION_TABLES[Decompress.DECOMPRESS8TO12].astype(np.uint16)
+    # Use the mean value column (2)
+    data = decompression[data, 2]
+
     # There is already a variable called "count" in the dataset that
     # came with the packet
     ds["data_index"] = xr.DataArray(np.arange(720), dims="data_index")
     ds["data"] = xr.DataArray(data, dims=("epoch", "data_index"))
     # Remove the original compressed data field
     ds = ds.drop_vars("data_compressed")
     return ds
-
-
-def process_star_sensor(ds: xr.Dataset) -> xr.Dataset:
-    """
-    Process Lo star sensor data.
-
-    Parameters
-    ----------
-    ds : xr.Dataset
-        The packet dataset containing Lo star sensor data.
-
-    Returns
-    -------
-    xr.Dataset
-        Processed dataset with unpacked and decompressed data.
-    """
-    # Decompress from 8 -> 12 bits using the decompression tables
-    decompression = DECOMPRESSION_TABLES[Decompress.DECOMPRESS8TO12].astype(np.uint16)
-
-    # Use the mean value column (2)
-    ds["data"].values = decompression[ds["data"].values, 2]
-
-    # We don't want the original l1a ccsds variables, only data
-    vars_to_drop = [x for x in ds.data_vars if x != "data"]
-    ds = ds.drop_vars(vars_to_drop)
-
-    return ds

imap_processing/lo/l1a/lo_l1a.py

@@ -15,7 +15,7 @@
     parse_events,
     parse_histogram,
 )
-from imap_processing.lo.l0.lo_star_sensor import process_star_sensor, unpack_star_sensor
+from imap_processing.lo.l0.lo_star_sensor import process_star_sensor
 from imap_processing.utils import convert_to_binary_string, packet_file_to_datasets
 
 logger = logging.getLogger(__name__)
@@ -98,17 +98,10 @@ def lo_l1a(dependency: Path) -> list[xr.Dataset]:
         )
         logical_source = "imap_lo_l1a_star"
         ds = datasets_by_apid[LoAPID.ILO_STAR]
-        ds = unpack_star_sensor(ds)
+        ds = process_star_sensor(ds)
         ds = add_dataset_attrs(ds, attr_mgr, logical_source)
         datasets_to_return.append(ds)
 
-        # Also create l1b at the same time
-        logical_source_l1b = "imap_lo_l1b_prostar"
-        ds_l1b = ds.copy(deep=True)
-        ds_l1b = process_star_sensor(ds_l1b)
-        ds_l1b.attrs.update(attr_mgr.get_global_attributes(logical_source_l1b))
-        datasets_to_return.append(ds_l1b)
-
     logger.info(f"Returning [{len(datasets_to_return)}] datasets")
     return datasets_to_return
 

imap_processing/tests/lo/test_lo_l1a.py

@@ -15,7 +15,6 @@ def test_lo_l1a():
         "imap_lo_l1a_histogram",
         "imap_lo_l1a_de",
         "imap_lo_l1a_star",
-        "imap_lo_l1b_prostar",
     ]
     output_dataset = lo_l1a(dependency)
 

imap_processing/tests/lo/test_star_sensor.py

@@ -3,7 +3,7 @@
 
 from imap_processing import imap_module_directory
 from imap_processing.lo.l0.lo_apid import LoAPID
-from imap_processing.lo.l0.lo_star_sensor import process_star_sensor, unpack_star_sensor
+from imap_processing.lo.l0.lo_star_sensor import process_star_sensor
 from imap_processing.utils import packet_file_to_datasets
 
 
@@ -30,29 +30,24 @@ def test_star_sensor(star_sensor_ds):
     validation_arr = np.loadtxt(validation_file, delimiter=",", skiprows=1, dtype=int)
     validation_shcoarse = validation_arr[:, 0]
     validation_count = validation_arr[:, 1]
-    validation_data_l1a = validation_arr[:, 2:722]
-    validation_data_l1b = validation_arr[:, 722:-1]
+    # The first 720
+    # validation_data_compressed = validation_arr[:, 2:722]
+    validation_data_decompressed = validation_arr[:, 722:-1]
     validation_checksum = validation_arr[:, -1]
 
     ## Act
-    ds = unpack_star_sensor(star_sensor_ds)
+    ds = process_star_sensor(star_sensor_ds)
 
     ## Assert
     # 45 times and 720 count values
     assert ds["data"].shape == (45, 720)
-    assert ds["data"].dtype == np.uint8
+    assert ds["data"].dtype == np.uint16
 
     # We are only spot checking a few values from the validation file
     # the first 3 and the final value.
     small_ds = ds.isel(epoch=[0, 1, 2, -1])
     assert len(small_ds["epoch"]) == 4
     np.testing.assert_array_equal(small_ds["shcoarse"], validation_shcoarse)
     np.testing.assert_array_equal(small_ds["count"], validation_count)
-    np.testing.assert_array_equal(small_ds["data"], validation_data_l1a)
+    np.testing.assert_array_equal(small_ds["data"], validation_data_decompressed)
     np.testing.assert_array_equal(small_ds["chksum"], validation_checksum)
-
-    # Now test the l1b processed data
-    ds_l1b = process_star_sensor(small_ds)
-    # The 12 bit unpacking should be expanded to uint16
-    assert ds_l1b["data"].dtype == np.uint16
-    np.testing.assert_array_equal(ds_l1b["data"], validation_data_l1b)