MNT: Upgrade space_packet_parser to next release

This gets rid of some warnings in the code base and simplifies some
instantiation logic.

Author: greglucas

imap_processing/decom.py

@@ -5,9 +5,13 @@
 to decommutate CCSDS packet data using a given XTCE packet definition.
 """
 
+import logging
 from pathlib import Path
 
-from space_packet_parser import definitions
+import space_packet_parser as spp
+from space_packet_parser.exceptions import UnrecognizedPacketTypeError
+
+logger = logging.getLogger(__name__)
 
 
 def decom_packets(packet_file: str | Path, xtce_packet_definition: str | Path) -> list:
@@ -29,8 +33,15 @@ def decom_packets(packet_file: str | Path, xtce_packet_definition: str | Path) -
     list
         List of all the unpacked data.
     """
-    packet_definition = definitions.XtcePacketDefinition(xtce_packet_definition)
+    packet_definition = spp.load_xtce(xtce_packet_definition)
 
     with open(packet_file, "rb") as binary_data:
-        packet_generator = packet_definition.packet_generator(binary_data)
-        return list(packet_generator)
+        packets = []
+        for binary_packet in spp.ccsds_generator(binary_data):
+            try:
+                packets.append(packet_definition.parse_bytes(binary_packet))
+            except UnrecognizedPacketTypeError as e:
+                # Log the error and continue processing other packets
+                logger.debug(e)
+                continue
+        return packets

imap_processing/glows/l0/decom_glows.py

@@ -3,7 +3,7 @@
 from enum import Enum
 from pathlib import Path
 
-from space_packet_parser import definitions
+import space_packet_parser as spp
 
 from imap_processing import imap_module_directory
 from imap_processing.ccsds.ccsds_data import CcsdsData
@@ -49,32 +49,40 @@ def decom_packets(
         f"{imap_module_directory}/glows/packet_definitions/GLX_COMBINED.xml"
     )
 
-    packet_definition = definitions.XtcePacketDefinition(xtce_document)
+    packet_definition = spp.load_xtce(xtce_document)
 
     histdata = []
     dedata = []
 
     filename = packet_file_path.name
 
     with open(packet_file_path, "rb") as binary_data:
-        glows_packets = packet_definition.packet_generator(binary_data)
-
-        for packet in glows_packets:
+        for binary_packet in spp.ccsds_generator(binary_data):
+            packet = packet_definition.parse_bytes(binary_packet)
             apid = packet["PKT_APID"]
             # Do something with the packet data
             if apid == GlowsParams.HIST_APID.value:
-                values = [item.raw_value for item in packet.user_data.values()]
-                hist_l0 = HistogramL0(
-                    __version__, filename, CcsdsData(packet.header), *values
-                )
+                values = [
+                    item.raw_value for i, item in enumerate(packet.values()) if i > 6
+                ]
+                header = {
+                    key: value
+                    for i, (key, value) in enumerate(packet.items())
+                    if i <= 6
+                }
+                hist_l0 = HistogramL0(__version__, filename, CcsdsData(header), *values)
                 histdata.append(hist_l0)
 
             if apid == GlowsParams.DE_APID.value:
-                values = [item.raw_value for item in packet.user_data.values()]
-
-                de_l0 = DirectEventL0(
-                    __version__, filename, CcsdsData(packet.header), *values
-                )
+                values = [
+                    item.raw_value for i, item in enumerate(packet.values()) if i > 6
+                ]
+                header = {
+                    key: value
+                    for i, (key, value) in enumerate(packet.items())
+                    if i <= 6
+                }
+                de_l0 = DirectEventL0(__version__, filename, CcsdsData(header), *values)
                 dedata.append(de_l0)
 
         return histdata, dedata

imap_processing/idex/idex_l1a.py

@@ -288,7 +288,7 @@ class RawDustEvent:
 
     Parameters
     ----------
-    header_packet : space_packet_parser.packets.CCSDSPacket
+    header_packet : space_packet_parser.SpacePacket
         The FPGA metadata event header.
 
     Attributes
@@ -340,7 +340,7 @@ class RawDustEvent:
     MAX_HIGH_BLOCKS = 16
     MAX_LOW_BLOCKS = 64
 
-    def __init__(self, header_packet: space_packet_parser.packets.CCSDSPacket) -> None:
+    def __init__(self, header_packet: space_packet_parser.SpacePacket) -> None:
         """
         Initialize a raw dust event, with an FPGA Header Packet from IDEX.
 
@@ -352,7 +352,7 @@ def __init__(self, header_packet: space_packet_parser.packets.CCSDSPacket) -> No
 
         Parameters
         ----------
-        header_packet : space_packet_parser.packets.CCSDSPacket
+        header_packet : space_packet_parser.SpacePacket
             The FPGA metadata event header.
         """
         # Calculate the impact time in seconds since epoch
@@ -371,8 +371,8 @@ def __init__(self, header_packet: space_packet_parser.packets.CCSDSPacket) -> No
         # Iterate through every telemetry item not in the header and pull out the values
         self.telemetry_items = {
             key.lower(): val
-            for key, val in header_packet.items()
-            if key not in header_packet.header.keys()
+            for i, (key, val) in enumerate(header_packet.items())
+            if i > 6  # Skip first 7 header items
         }
 
         logger.debug(
@@ -420,7 +420,7 @@ def _append_raw_data(self, scitype: Scitype, bits: str) -> None:
             logger.warning("Unknown science type received: [%s]", scitype)
 
     def _set_sample_trigger_times(
-        self, packet: space_packet_parser.packets.CCSDSPacket
+        self, packet: space_packet_parser.SpacePacket
     ) -> None:
         """
         Calculate the actual sample trigger time.
@@ -430,7 +430,7 @@ def _set_sample_trigger_times(
 
         Parameters
         ----------
-        packet : space_packet_parser.packets.CCSDSPacket
+        packet : space_packet_parser.SpacePacket
             The IDEX FPGA header packet info.
 
         Notes
@@ -590,15 +590,13 @@ def _calc_high_sample_resolution(self, num_samples: int) -> npt.NDArray:
         )
         return time_high_sample_rate_data
 
-    def _populate_bit_strings(
-        self, packet: space_packet_parser.packets.CCSDSPacket
-    ) -> None:
+    def _populate_bit_strings(self, packet: space_packet_parser.SpacePacket) -> None:
         """
         Parse IDEX data packets to populate bit strings.
 
         Parameters
         ----------
-        packet : space_packet_parser.packets.CCSDSPacket
+        packet : space_packet_parser.SpacePacket
             A single science data packet for one of the 6.
             IDEX observables.
         """

imap_processing/mag/l0/decom_mag.py

@@ -8,8 +8,8 @@
 from pathlib import Path
 
 import numpy as np
+import space_packet_parser as spp
 import xarray as xr
-from space_packet_parser import definitions
 
 from imap_processing import imap_module_directory
 from imap_processing.ccsds.ccsds_data import CcsdsData
@@ -41,20 +41,26 @@ def decom_packets(packet_file_path: str | Path) -> dict[str, list[MagL0]]:
         f"{imap_module_directory}/mag/packet_definitions/MAG_SCI_COMBINED.xml"
     )
 
-    packet_definition = definitions.XtcePacketDefinition(xtce_document)
+    packet_definition = spp.load_xtce(xtce_document)
 
     # Store in a dict for de-duplication. Only the keys are returned as a list.
     norm_dict: dict[MagL0, None] = {}
     burst_dict: dict[MagL0, None] = {}
 
     with open(packet_file_path, "rb") as binary_data:
-        mag_packets = packet_definition.packet_generator(binary_data)
-
-        for packet in mag_packets:
+        for binary_packet in spp.ccsds_generator(binary_data):
+            packet = packet_definition.parse_bytes(binary_packet)
             apid = packet["PKT_APID"]
             if apid in (Mode.BURST, Mode.NORMAL):
-                values = [item.raw_value for item in packet.user_data.values()]
-                mag_l0 = MagL0(CcsdsData(packet.header), *values)
+                values = [
+                    item.raw_value for i, item in enumerate(packet.values()) if i > 6
+                ]
+                header = {
+                    key: value
+                    for i, (key, value) in enumerate(packet.items())
+                    if i <= 6
+                }
+                mag_l0 = MagL0(CcsdsData(header), *values)
                 if apid == Mode.NORMAL:
                     if mag_l0 not in norm_dict:
                         norm_dict[mag_l0] = None

imap_processing/utils.py

@@ -6,8 +6,10 @@
 
 import numpy as np
 import pandas as pd
+import space_packet_parser as spp
 import xarray as xr
-from space_packet_parser import definitions, encodings, parameters
+from space_packet_parser.exceptions import UnrecognizedPacketTypeError
+from space_packet_parser.xtce import definitions, encodings, parameter_types
 
 from imap_processing.spice.time import met_to_ttj2000ns
 
@@ -158,11 +160,11 @@ def _get_minimum_numpy_datatype(  # noqa: PLR0912 - Too many branches
     datatype : str
         The minimum datatype.
     """
-    data_encoding = definition.named_parameters[name].parameter_type.encoding
+    data_encoding = definition.parameters[name].parameter_type.encoding
 
     if use_derived_value and isinstance(
-        definition.named_parameters[name].parameter_type,
-        parameters.EnumeratedParameterType,
+        definition.parameters[name].parameter_type,
+        parameter_types.EnumeratedParameterType,
     ):
         # We don't have a way of knowing what is enumerated,
         # let numpy infer the datatype
@@ -254,11 +256,18 @@ def packet_file_to_datasets(
     variable_mapping: dict[int, set] = dict()
 
     # Set up the parser from the input packet definition
-    packet_definition = definitions.XtcePacketDefinition(xtce_packet_definition)
+    packet_definition = spp.load_xtce(xtce_packet_definition)
 
     with open(packet_file, "rb") as binary_data:
-        packet_generator = packet_definition.packet_generator(binary_data)
-        for packet in packet_generator:
+        for binary_packet in spp.ccsds_generator(binary_data):
+            try:
+                packet = packet_definition.parse_bytes(binary_packet)
+            except UnrecognizedPacketTypeError as e:
+                # NOTE: Not all of our definitions have all of the APIDs
+                #       we may encounter, so we only want to process ones
+                #       we can actual parse.
+                logger.debug(e)
+                continue
             apid = packet["PKT_APID"]
             if apid not in data_dict:
                 # This is the first packet for this APID
@@ -274,10 +283,7 @@ def packet_file_to_datasets(
                     f"got: {packet.keys()}"
                 )
 
-            # TODO: Do we want to give an option to remove the header content?
-            packet_content = packet.user_data | packet.header
-
-            for key, value in packet_content.items():
+            for key, value in packet.items():
                 val = value if use_derived_value else value.raw_value
                 data_dict[apid][key].append(val)
                 if key not in datatype_mapping[apid]:
@@ -289,8 +295,15 @@ def packet_file_to_datasets(
     dataset_by_apid = {}
 
     for apid, data in data_dict.items():
-        # The time key is always the first key in the data dictionary on IMAP
-        time_key = next(iter(data.keys()))
+        # The time key is the secondary header, right after the primary header
+        # in the data dictionary on IMAP (8th key overall)
+        try:
+            time_key = list(data.keys())[7]
+        except IndexError:
+            logger.debug(
+                f"Could not determine time key for APID {apid}, skipping dataset."
+            )
+            continue
         # Convert to J2000 time and use that as our primary dimension
         time_data = met_to_ttj2000ns(data[time_key])
         ds = xr.Dataset(