add sc status

Author: laspsandoval

imap_processing/ialirt/l0/process_status.py

@@ -0,0 +1,77 @@
+"""Functions to support status processing."""
+
+import logging
+
+import xarray as xr
+
+from imap_processing.ialirt.utils.grouping import (
+    _populate_instrument_header_items,
+)
+from imap_processing.ialirt.utils.time import calculate_time
+from imap_processing.spice.time import met_to_ttj2000ns
+
+logger = logging.getLogger(__name__)
+
+
+def process_status(xarray_data: xr.Dataset) -> list[dict]:
+    """
+    Create L1 data dictionary.
+
+    Parameters
+    ----------
+    xarray_data : xr.Dataset
+        Parsed data.
+
+    Returns
+    -------
+    status_data : list[dict]
+        Dictionary final data product.
+    """
+    status_data = []
+
+    # Subsecond time conversion specified in 7516-9054 GSW-FSW ICD.
+    # Value of SCLK subseconds, unsigned, (LSB = 1/256 sec)
+    met = calculate_time(
+        xarray_data["sc_sclk_sec"], xarray_data["sc_sclk_sub_sec"], 256
+    )
+
+    # Add required parameters.
+    xarray_data["met"] = met
+
+    sc_swapi_status = xarray_data["sc_swapi_status"]
+    sc_mag_status = xarray_data["sc_mag_status"]
+    sc_hit_status = xarray_data["sc_hit_status"]
+    sc_codice_status = xarray_data["sc_codice_status"]
+    sc_lo_status = xarray_data["sc_lo_status"]
+    sc_hi_45_status = xarray_data["sc_hi_45_status"]
+    sc_hi_90_status = xarray_data["sc_hi_90_status"]
+    sc_ultra_45_status = xarray_data["sc_ultra_45_status"]
+    sc_ultra_90_status = xarray_data["sc_ultra_90_status"]
+    sc_swe_status = xarray_data["sc_swe_status"]
+    sc_idex_status = xarray_data["sc_idex_status"]
+    sc_glows_status = xarray_data["sc_glows_status"]
+    sc_autonomy_status = xarray_data["sc_autonomy"]
+
+    for i in range(len(xarray_data["met"])):
+        status_data.append(
+            _populate_instrument_header_items(met)
+            | {
+                "instrument": "spacecraft_status",
+                "status_epoch": int(met_to_ttj2000ns(met[i])),
+                "sc_swapi_status": int(sc_swapi_status[i]),
+                "sc_mag_status": int(sc_mag_status[i]),
+                "sc_hit_status": int(sc_hit_status[i]),
+                "sc_codice_status": int(sc_codice_status[i]),
+                "sc_lo_status": int(sc_lo_status[i]),
+                "sc_hi_45_status": int(sc_hi_45_status[i]),
+                "sc_hi_90_status": int(sc_hi_90_status[i]),
+                "sc_ultra_45_status": int(sc_ultra_45_status[i]),
+                "sc_ultra_90_status": int(sc_ultra_90_status[i]),
+                "sc_swe_status": int(sc_swe_status[i]),
+                "sc_idex_status": int(sc_idex_status[i]),
+                "sc_glows_status": int(sc_glows_status[i]),
+                "sc_autonomy_status": int(sc_autonomy_status[i]),
+            }
+        )
+
+    return status_data

imap_processing/tests/external_test_data_config.py

@@ -130,6 +130,11 @@
     ("iois_1_packets_2025_284_05_54_39", "ialirt/data/l0/"),
     ("iois_1_packets_2025_344_05_57_56", "ialirt/data/l0/"),
     ("iois_1_packets_2025_344_05_59_58", "ialirt/data/l0/"),
+    ("iois_1_packets_2026_090_05_03_05", "ialirt/data/l0/"),
+    ("iois_1_packets_2026_090_05_04_06", "ialirt/data/l0/"),
+    ("iois_1_packets_2026_090_05_05_07", "ialirt/data/l0/"),
+    ("iois_1_packets_2026_090_05_06_08", "ialirt/data/l0/"),
+    ("iois_1_packets_2026_090_05_07_09", "ialirt/data/l0/"),
     ("imap_recon_od005_20250925_20251014_v01.bsp", "spice/test_data/"),
     ("imap_2025_283_2025_284_001.ah.bc", "spice/test_data/"),
 

imap_processing/tests/ialirt/unit/test_process_status.py

@@ -0,0 +1,52 @@
+"""Tests for the process_status module."""
+
+import pytest
+import xarray as xr
+
+from imap_processing import imap_module_directory
+from imap_processing.ialirt.l0.process_status import process_status
+from imap_processing.utils import packet_file_to_datasets
+
+
+@pytest.fixture(scope="session")
+def postlaunch_packet_path():
+    """Returns the paths to the binary packets."""
+    directory = imap_module_directory / "tests" / "ialirt" / "data" / "l0"
+    filenames = [
+        "iois_1_packets_2026_090_05_03_05",
+        "iois_1_packets_2026_090_05_04_06",
+        "iois_1_packets_2026_090_05_05_07",
+        "iois_1_packets_2026_090_05_06_08",
+        "iois_1_packets_2026_090_05_07_09",
+    ]
+    return tuple(directory / fname for fname in filenames)
+
+
+@pytest.fixture
+def postlaunch_xarray_data(postlaunch_packet_path, sc_packet_path):
+    """Create xarray data for multiple packets."""
+    apid = 478
+    _, xtce_ialirt_path = sc_packet_path
+
+    xarray_data = tuple(
+        packet_file_to_datasets(packet, xtce_ialirt_path, use_derived_value=False)[apid]
+        for packet in postlaunch_packet_path
+    )
+
+    merged_xarray_data = xr.concat(xarray_data, dim="epoch")
+    return merged_xarray_data
+
+
+@pytest.mark.external_test_data
+def test_process_status(postlaunch_xarray_data):
+    """Test the process_status function."""
+
+    status_data = process_status(postlaunch_xarray_data)
+
+    for i in range(len(status_data)):
+        assert status_data[i]["sc_swapi_status"] == 1
+        assert status_data[i]["sc_mag_status"] == 1
+        assert status_data[i]["sc_hit_status"] == 1
+        assert status_data[i]["sc_codice_status"] == 1
+        assert status_data[i]["sc_lo_status"] == 1
+        assert status_data[i]["sc_autonomy_status"] == 1