L0 Decom Ultra

imap_processing/tests/ultra/unit/conftest.py

@@ -19,40 +19,44 @@ def ccsds_path():
 
 
 @pytest.fixture()
-def xtce_image_rates_path():
-    """Returns the xtce image rates directory."""
+def ccsds_path_events():
+    """Returns the ccsds directory."""
     return (
         Path(sys.modules[__name__.split(".")[0]].__file__).parent
+        / "tests"
         / "ultra"
-        / "packet_definitions"
-        / "P_U45_IMAGE_RATES.xml"
+        / "test_data"
+        / "l0"
+        / "FM45_7P_Phi0.0_BeamCal_LinearScan_phi0.04_theta-0.01_20230821T121304.CCSDS"
     )
 
 
 @pytest.fixture()
-def xtce_aux_path():
-    """Returns the xtce auxilliary directory."""
+def ccsds_path_tof():
+    """Returns the ccsds directory."""
     return (
         Path(sys.modules[__name__.split(".")[0]].__file__).parent
+        / "tests"
         / "ultra"
-        / "packet_definitions"
-        / "P_U45_AUXILIARY.xml"
+        / "test_data"
+        / "l0"
+        / "FM45_TV_Cycle6_Hot_Ops_Front212_20240124T063837.CCSDS"
     )
 
 
 @pytest.fixture()
-def xtce_image_raw_events_path():
-    """Returns the xtce image raw events directory."""
+def xtce_path():
+    """Returns the xtce image rates directory."""
     return (
         Path(sys.modules[__name__.split(".")[0]].__file__).parent
         / "ultra"
         / "packet_definitions"
-        / "P_U45_IMG_RAW_EVENTS.xml"
+        / "ULTRA_SCI_COMBINED.xml"
     )
 
 
 @pytest.fixture()
-def xtce_image_rates_test_path():
+def rates_test_path():
     """Returns the xtce image rates test data directory."""
     filename = (
         "ultra45_raw_sc_ultraimgrates_Ultra45_EM_SwRI_Cal_Run7_ThetaScan_"
@@ -69,7 +73,7 @@ def xtce_image_rates_test_path():
 
 
 @pytest.fixture()
-def xtce_aux_test_path():
+def aux_test_path():
     """Returns the xtce auxiliary test data directory."""
     filename = (
         "ultra45_raw_sc_auxdata_Ultra45_EM_SwRI_Cal_Run7_ThetaScan_"
@@ -83,3 +87,37 @@ def xtce_aux_test_path():
         / "l0"
         / filename
     )
+
+
+@pytest.fixture()
+def events_test_path():
+    """Returns the xtce auxiliary test data directory."""
+    filename = (
+        "ultra45_raw_sc_ultrarawimgevent_FM45_7P_Phi00_BeamCal_"
+        "LinearScan_phi004_theta-001_20230821T121304.csv"
+    )
+    return (
+        Path(sys.modules[__name__.split(".")[0]].__file__).parent
+        / "tests"
+        / "ultra"
+        / "test_data"
+        / "l0"
+        / filename
+    )
+
+
+@pytest.fixture()
+def tof_test_path():
+    """Returns the xtce auxiliary test data directory."""
+    filename = (
+        "ultra45_raw_sc_enaphxtofhangimg_FM45_TV_Cycle6_Hot_Ops_"
+        "Front212_20240124T063837.csv"
+    )
+    return (
+        Path(sys.modules[__name__.split(".")[0]].__file__).parent
+        / "tests"
+        / "ultra"
+        / "test_data"
+        / "l0"
+        / filename
+    )

imap_processing/tests/ultra/unit/test_decom_apid_880.py

@@ -1,58 +1,63 @@
+import numpy as np
 import pandas as pd
 import pytest
 
 from imap_processing import decom
+from imap_processing.ultra.l0.decom_ultra import decom_ultra_apids
+from imap_processing.ultra.l0.ultra_utils import ULTRA_AUX
+from imap_processing.utils import group_by_apid
 
 
 @pytest.fixture()
-def decom_test_data(ccsds_path, xtce_aux_path):
+def decom_test_data(ccsds_path, xtce_path):
     """Read test data from file"""
-    data_packet_list = decom.decom_packets(ccsds_path, xtce_aux_path)
+    data_packet_list = decom.decom_packets(ccsds_path, xtce_path)
     return data_packet_list
 
 
-def test_aux_length(decom_test_data):
-    """Test if total packets in data file is correct"""
-    total_packets = 23
-    assert len(decom_test_data) == total_packets
-
-
 def test_aux_enumerated(decom_test_data):
     """Test if enumerated values derived correctly"""
 
-    for packet in decom_test_data:
+    count = 0  # count number of packets with APID 880
+    total_packets = 23
+
+    grouped_data = group_by_apid(decom_test_data)
+    apid_data = grouped_data[880]
+
+    for packet in apid_data:
         assert packet.data["SPINPERIODVALID"].derived_value == "INVALID"
         assert packet.data["SPINPHASEVALID"].derived_value == "VALID"
         assert packet.data["SPINPERIODSOURCE"].derived_value == "NOMINAL"
         assert packet.data["CATBEDHEATERFLAG"].derived_value == "UNFLAGGED"
+        count += 1
+
+    assert count == total_packets
 
 
 def test_aux_mode(decom_test_data):
     """Test if enumerated values derived correctly"""
 
     for packet in decom_test_data:
-        assert packet.data["HWMODE"].derived_value == "MODE0"
-        assert packet.data["IMCENB"].derived_value == "MODE0"
-        assert packet.data["LEFTDEFLECTIONCHARGE"].derived_value == "MODE0"
-        assert packet.data["RIGHTDEFLECTIONCHARGE"].derived_value == "MODE0"
+        if packet.header["PKT_APID"].derived_value == 880:
+            assert packet.data["HWMODE"].derived_value == "MODE0"
+            assert packet.data["IMCENB"].derived_value == "MODE0"
+            assert packet.data["LEFTDEFLECTIONCHARGE"].derived_value == "MODE0"
+            assert packet.data["RIGHTDEFLECTIONCHARGE"].derived_value == "MODE0"
 
 
-def test_aux_decom(decom_test_data, xtce_aux_test_path):
+def test_aux_decom(ccsds_path, xtce_path, aux_test_path):
     """This function reads validation data and checks that
     decom data matches validation data for auxiliary packet"""
 
-    df = pd.read_csv(xtce_aux_test_path, index_col="MET")
-
-    for packet in decom_test_data:
-        time = packet.data["SHCOARSE"].derived_value
-        subdf = df.loc[time]
-
-        assert subdf.SpinStartSeconds == packet.data["TIMESPINSTART"].derived_value
-        assert (
-            subdf.SpinStartSubseconds == packet.data["TIMESPINSTARTSUB"].derived_value
-        )
-        assert subdf.SpinDuration == packet.data["DURATION"].derived_value
-        assert subdf.SpinNumber == packet.data["SPINNUMBER"].derived_value
-        assert subdf.SpinDataTime == packet.data["TIMESPINDATA"].derived_value
-        assert subdf.SpinPeriod == packet.data["SPINPERIOD"].derived_value
-        assert subdf.SpinPhase == packet.data["SPINPHASE"].derived_value
+    decom_ultra = decom_ultra_apids(ccsds_path, xtce_path, ULTRA_AUX.apid[0])
+    df = pd.read_csv(aux_test_path, index_col="MET")
+
+    np.testing.assert_array_equal(df.SpinStartSeconds, decom_ultra["TIMESPINSTART"])
+    np.testing.assert_array_equal(
+        df.SpinStartSubseconds, decom_ultra["TIMESPINSTARTSUB"]
+    )
+    np.testing.assert_array_equal(df.SpinDuration, decom_ultra["DURATION"])
+    np.testing.assert_array_equal(df.SpinNumber, decom_ultra["SPINNUMBER"])
+    np.testing.assert_array_equal(df.SpinDataTime, decom_ultra["TIMESPINDATA"])
+    np.testing.assert_array_equal(df.SpinPeriod, decom_ultra["SPINPERIOD"])
+    np.testing.assert_array_equal(df.SpinPhase, decom_ultra["SPINPHASE"])

imap_processing/tests/ultra/unit/test_decom_apid_881.py

@@ -5,41 +5,42 @@
 import pytest
 
 from imap_processing import decom
-from imap_processing.ultra.l0.decom_ultra import decom_ultra_packets
+from imap_processing.ultra.l0.decom_ultra import decom_ultra_apids
+from imap_processing.ultra.l0.ultra_utils import ULTRA_RATES
 
 
 @pytest.fixture()
-def decom_test_data(ccsds_path, xtce_image_rates_path):
+def decom_test_data(ccsds_path, xtce_path):
     """Data for decom"""
-    data_packet_list = decom.decom_packets(ccsds_path, xtce_image_rates_path)
+    data_packet_list = decom.decom_packets(ccsds_path, xtce_path)
     return data_packet_list
 
 
 @pytest.fixture()
-def decom_ultra(ccsds_path, xtce_image_rates_path):
+def decom_ultra(ccsds_path, xtce_path):
     """Data for decom_ultra"""
-    data_packet_list = decom_ultra_packets(ccsds_path, xtce_image_rates_path)
-    return data_packet_list
-
-
-def test_image_rate_length(decom_test_data):
-    """Test if total packets in data file is correct"""
-    total_packets = 23
-    assert len(decom_test_data) == total_packets
+    data_packets = decom_ultra_apids(ccsds_path, xtce_path, ULTRA_RATES.apid[0])
+    return data_packets
 
 
-def test_image_rate_decom(decom_ultra, xtce_image_rates_test_path):
+def test_image_rate_decom(decom_ultra, rates_test_path):
     """This function reads validation data and checks that decom data
     matches validation data for image rate packet"""
 
-    df = pd.read_csv(xtce_image_rates_test_path, index_col="MET")
+    df = pd.read_csv(rates_test_path, index_col="MET")
+    count = 0
+    total_packets = 23
 
-    assert (df.SID == decom_ultra["science_id"]).all()
-    assert (df.Spin == decom_ultra["spin_data"]).all()
-    assert (df.AbortFlag == decom_ultra["abortflag_data"]).all()
-    assert (df.StartDelay == decom_ultra["startdelay_data"]).all()
+    np.testing.assert_array_equal(df.SID, decom_ultra["SID"])
+    np.testing.assert_array_equal(df.Spin, decom_ultra["SPIN"])
+    np.testing.assert_array_equal(df.AbortFlag, decom_ultra["ABORTFLAG"])
+    np.testing.assert_array_equal(df.StartDelay, decom_ultra["STARTDELAY"])
 
-    for time in decom_ultra.epoch.values:
+    for time in decom_ultra["SHCOARSE"]:
         arr1 = json.loads(df.loc[time].Counts)
-        arr2 = decom_ultra["fastdata_00"].sel(epoch=time).data
+        index = decom_ultra["SHCOARSE"].index(time)
+        arr2 = decom_ultra["FASTDATA_00"][index]
         np.testing.assert_array_equal(arr1, arr2)
+        count += 1
+
+    assert count == total_packets

imap_processing/tests/ultra/unit/test_decom_apid_883.py

@@ -0,0 +1,37 @@
+import json
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from imap_processing.ultra.l0.decom_ultra import decom_ultra_apids
+from imap_processing.ultra.l0.ultra_utils import ULTRA_TOF
+
+
+@pytest.fixture()
+def decom_ultra(ccsds_path_tof, xtce_path):
+    """Data for decom_ultra"""
+    data_packet_list = decom_ultra_apids(ccsds_path_tof, xtce_path, ULTRA_TOF.apid[0])
+    return data_packet_list
+
+
+def test_tof_decom(decom_ultra, tof_test_path):
+    """This function reads validation data and checks that decom data
+    matches validation data for image rate packet"""
+
+    df = pd.read_csv(tof_test_path, index_col="SequenceCount")
+
+    np.testing.assert_array_equal(df.Spin, decom_ultra["SPIN"])
+    np.testing.assert_array_equal(df.AbortFlag, decom_ultra["ABORTFLAG"])
+    np.testing.assert_array_equal(df.StartDelay, decom_ultra["STARTDELAY"])
+    assert json.loads(df["P00s"].values[0])[0] == decom_ultra["P00"][0]
+
+    for count in df.index.get_level_values("SequenceCount").values:
+        df_data = df[df.index.get_level_values("SequenceCount") == count].Images.values[
+            0
+        ]
+        index = decom_ultra["SRC_SEQ_CTR"].index(count)
+        decom_data = decom_ultra["PACKETDATA"][index]
+        df_data_array = np.array(json.loads(df_data)[0])
+
+        np.testing.assert_array_equal(df_data_array, decom_data)

imap_processing/tests/ultra/unit/test_decom_apid_896.py

@@ -0,0 +1,85 @@
+import numpy as np
+import pandas as pd
+import pytest
+
+from imap_processing.cdf.defaults import GlobalConstants
+from imap_processing.ultra.l0.decom_ultra import decom_ultra_apids
+from imap_processing.ultra.l0.ultra_utils import ULTRA_EVENTS
+
+
+@pytest.fixture()
+def decom_ultra(ccsds_path_events, xtce_path):
+    """Data for decom_ultra"""
+    data_packet_list = decom_ultra_apids(
+        ccsds_path_events, xtce_path, ULTRA_EVENTS.apid[0]
+    )
+    return data_packet_list
+
+
+def test_image_raw_events_decom(decom_ultra, events_test_path):
+    """This function reads validation data and checks that decom data
+    matches validation data for image rate packet"""
+
+    df = pd.read_csv(events_test_path, index_col="MET")
+    df.replace(-1, GlobalConstants.INT_FILLVAL, inplace=True)
+
+    np.testing.assert_array_equal(df.SID, decom_ultra["SID"])
+    np.testing.assert_array_equal(df["Spin"], decom_ultra["SPIN"])
+    np.testing.assert_array_equal(df["AbortFlag"], decom_ultra["ABORTFLAG"])
+    np.testing.assert_array_equal(df["StartDelay"], decom_ultra["STARTDELAY"])
+    np.testing.assert_array_equal(df["Count"], decom_ultra["COUNT"])
+    np.testing.assert_array_equal(df["CoinType"], decom_ultra["coin_type"])
+    np.testing.assert_array_equal(df["StartType"], decom_ultra["start_type"])
+    np.testing.assert_array_equal(df["StopType"], decom_ultra["stop_type"])
+    np.testing.assert_array_equal(df["StartPosTDC"], decom_ultra["start_pos_tdc"])
+    np.testing.assert_array_equal(df["StopNorthTDC"], decom_ultra["stop_north_tdc"])
+    np.testing.assert_array_equal(df["StopEastTDC"], decom_ultra["stop_east_tdc"])
+    np.testing.assert_array_equal(df["StopSouthTDC"], decom_ultra["stop_south_tdc"])
+    np.testing.assert_array_equal(df["StopWestTDC"], decom_ultra["stop_west_tdc"])
+    np.testing.assert_array_equal(df["CoinNorthTDC"], decom_ultra["coin_north_tdc"])
+    np.testing.assert_array_equal(df["CoinSouthTDC"], decom_ultra["coin_south_tdc"])
+    np.testing.assert_array_equal(
+        df["CoinDiscreteTDC"], decom_ultra["coin_discrete_tdc"]
+    )
+    np.testing.assert_array_equal(df["EnergyOrPH"], decom_ultra["energy_ph"])
+    np.testing.assert_array_equal(df["PulseWidth"], decom_ultra["pulse_width"])
+    np.testing.assert_array_equal(df["PhaseAngle"], decom_ultra["phase_angle"])
+    np.testing.assert_array_equal(df["Bin"], decom_ultra["bin"])
+
+
+def test_image_raw_events_decom_flags(decom_ultra, events_test_path):
+    """This function reads validation data and checks that decom data
+    matches validation data for image rate packet"""
+
+    df = pd.read_csv(events_test_path, index_col="MET")
+    df.replace(-1, GlobalConstants.INT_FILLVAL, inplace=True)
+
+    np.testing.assert_array_equal(df["CnT"], decom_ultra["event_flag_cnt"])
+    np.testing.assert_array_equal(df["PHCmpSL"], decom_ultra["event_flag_phcmpsl"])
+    np.testing.assert_array_equal(df["PHCmpSR"], decom_ultra["event_flag_phcmpsr"])
+    np.testing.assert_array_equal(df["PHCmpCD"], decom_ultra["event_flag_phcmpcd"])
+    np.testing.assert_array_equal(df["SSDS7"], decom_ultra["ssd_flag_7"])
+    np.testing.assert_array_equal(df["SSDS6"], decom_ultra["ssd_flag_6"])
+    np.testing.assert_array_equal(df["SSDS5"], decom_ultra["ssd_flag_5"])
+    np.testing.assert_array_equal(df["SSDS4"], decom_ultra["ssd_flag_4"])
+    np.testing.assert_array_equal(df["SSDS3"], decom_ultra["ssd_flag_3"])
+    np.testing.assert_array_equal(df["SSDS2"], decom_ultra["ssd_flag_2"])
+    np.testing.assert_array_equal(df["SSDS1"], decom_ultra["ssd_flag_1"])
+    np.testing.assert_array_equal(df["SSDS0"], decom_ultra["ssd_flag_0"])
+    np.testing.assert_array_equal(df["CFDCoinTN"], decom_ultra["cfd_flag_cointn"])
+    np.testing.assert_array_equal(df["CFDCoinBN"], decom_ultra["cfd_flag_coinbn"])
+    np.testing.assert_array_equal(df["CFDCoinTS"], decom_ultra["cfd_flag_coints"])
+    np.testing.assert_array_equal(df["CFDCoinBS"], decom_ultra["cfd_flag_coinbs"])
+    np.testing.assert_array_equal(df["CFDCoinD"], decom_ultra["cfd_flag_coind"])
+    np.testing.assert_array_equal(df["CFDStartRF"], decom_ultra["cfd_flag_startrf"])
+    np.testing.assert_array_equal(df["CFDStartLF"], decom_ultra["cfd_flag_startlf"])
+    np.testing.assert_array_equal(df["CFDStartRP"], decom_ultra["cfd_flag_startrp"])
+    np.testing.assert_array_equal(df["CFDStartLP"], decom_ultra["cfd_flag_startlp"])
+    np.testing.assert_array_equal(df["CFDStopTN"], decom_ultra["cfd_flag_stoptn"])
+    np.testing.assert_array_equal(df["CFDStopBN"], decom_ultra["cfd_flag_stopbn"])
+    np.testing.assert_array_equal(df["CFDStopTE"], decom_ultra["cfd_flag_stopte"])
+    np.testing.assert_array_equal(df["CFDStopBE"], decom_ultra["cfd_flag_stopbe"])
+    np.testing.assert_array_equal(df["CFDStopTS"], decom_ultra["cfd_flag_stopts"])
+    np.testing.assert_array_equal(df["CFDStopBS"], decom_ultra["cfd_flag_stopbs"])
+    np.testing.assert_array_equal(df["CFDStopTW"], decom_ultra["cfd_flag_stoptw"])
+    np.testing.assert_array_equal(df["CFDStopBW"], decom_ultra["cfd_flag_stopbw"])