Add Lo star tracker packet processing

imap_processing/lo/l0/lo_star_sensor.py

@@ -0,0 +1,49 @@
+"""Processing function for Lo star sensor data."""
+
+import logging
+
+import numpy as np
+import xarray as xr
+
+from imap_processing.lo.l0.utils.bit_decompression import (
+    DECOMPRESSION_TABLES,
+    Decompress,
+)
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+
+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 a decompressed data field.
+    """
+    # Make one long flat buffer
+    # This assumes that all data_compressed entries are of the same length
+    # but allows for only one frombuffer call
+    buffer = b"".join(ds["data_compressed"].values)
+    data = np.frombuffer(buffer, dtype=np.uint8).reshape(-1, 720)
+
+    # Now 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_count"] = xr.DataArray(np.arange(720), dims="data_count")
+    # Create a new DataArray with dimensions (epoch, count)
+    ds["data"] = xr.DataArray(data, dims=("epoch", "data_count"))
+    # Remove the original compressed data field
+    ds = ds.drop_vars("data_compressed")
+    return ds

imap_processing/lo/l1a/lo_l1a.py

@@ -15,6 +15,7 @@
     parse_events,
     parse_histogram,
 )
+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__)
@@ -100,10 +101,24 @@ def lo_l1a(dependency: Path) -> list[xr.Dataset]:
         datasets_by_apid[LoAPID.ILO_SCI_DE] = add_dataset_attrs(
             datasets_by_apid[LoAPID.ILO_SCI_DE], attr_mgr, logical_source
         )
+    if LoAPID.ILO_STAR in datasets_by_apid:
+        logger.info(
+            f"\nProcessing {LoAPID(LoAPID.ILO_STAR).name} "
+            f"packet (APID: {LoAPID.ILO_STAR.value})"
+        )
+        datasets_by_apid[LoAPID.ILO_STAR] = process_star_sensor(
+            datasets_by_apid[LoAPID.ILO_STAR]
+        )
 
-    good_apids = [LoAPID.ILO_SPIN, LoAPID.ILO_SCI_CNT, LoAPID.ILO_SCI_DE]
-    logger.info(f"\nReturning datasets: {[LoAPID(apid) for apid in good_apids]}")
-    return [datasets_by_apid[good_apid] for good_apid in good_apids]
+    good_apids = [
+        LoAPID.ILO_SPIN,
+        LoAPID.ILO_SCI_CNT,
+        LoAPID.ILO_SCI_DE,
+        LoAPID.ILO_STAR,
+    ]
+    apids_with_data = [apid for apid in good_apids if apid in datasets_by_apid]
+    logger.info(f"\nReturning datasets: {[LoAPID(apid) for apid in apids_with_data]}")
+    return [datasets_by_apid[apid] for apid in apids_with_data]
 
 
 def add_dataset_attrs(

imap_processing/tests/lo/test_star_sensor.py

@@ -1,35 +1,50 @@
-from collections import namedtuple
-
+import numpy as np
 import pytest
 
-from imap_processing.ccsds.ccsds_data import CcsdsData
-from imap_processing.lo.l0.data_classes.star_sensor import StarSensor
+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
+from imap_processing.utils import packet_file_to_datasets
 
 
 @pytest.fixture
-def star_sensor():
-    fake_data_field = namedtuple("fake_packet", ["raw_value", "derived_value"])
-    star_sensor = StarSensor.__new__(StarSensor)
-    star_sensor.ccsds_header = CcsdsData(
-        {
-            "VERSION": fake_data_field(0, 0),
-            "TYPE": fake_data_field(0, 0),
-            "SEC_HDR_FLG": fake_data_field(0, 0),
-            "PKT_APID": fake_data_field(706, 706),
-            "SEQ_FLGS": fake_data_field(0, 0),
-            "SRC_SEQ_CTR": fake_data_field(0, 0),
-            "PKT_LEN": fake_data_field(0, 0),
-        }
+def star_sensor_ds():
+    xtce_file = imap_module_directory / "lo/packet_definitions/lo_xtce.xml"
+    dependency = (
+        imap_module_directory / "tests/lo/test_pkts/imap_lo_l0_raw_20240803_v002.pkts"
+    )
+    datasets_by_apid = packet_file_to_datasets(
+        packet_file=dependency.resolve(),
+        xtce_packet_definition=xtce_file.resolve(),
+        use_derived_value=False,
     )
-    return star_sensor
+    return datasets_by_apid[LoAPID.ILO_STAR]
 
 
-def test_science_counts(star_sensor):
-    ## Arrange
-    star_sensor.DATA_COMPRESSED = "0" * 5760
+def test_science_counts(star_sensor_ds):
+    validation_file = (
+        imap_module_directory
+        / "tests/lo/validation_data"
+        / "Instrument_FM1_T104_R129_20240803_ILO_STAR_EU_trimmed.csv"
+    )
+    validation_arr = np.loadtxt(validation_file, delimiter=",", skiprows=1, dtype=int)
+    validation_shcoarse = validation_arr[:, 0]
+    validation_count = validation_arr[:, 1]
+    validation_data = validation_arr[:, 2:-1]
+    validation_checksum = validation_arr[:, -1]
 
     ## Act
-    star_sensor._decompress_data()
+    ds = process_star_sensor(star_sensor_ds)
 
     ## Assert
-    assert star_sensor.DATA.shape == (720,)
+    # 45 times and 720 count values
+    assert ds["data"].shape == (45, 720)
+
+    # We are only spot checking a few values from the validation file
+    small_ds = ds.where(ds["shcoarse"].isin(validation_shcoarse), drop=True)
+    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)
+    np.testing.assert_array_equal(small_ds["chksum"], validation_checksum)