changes to decom

imap_processing/ialirt/l0/decom_ialirt.py

@@ -1,19 +1,16 @@
-"""Decommutates i-alert packets and creates L1 data products."""
+"""Decommutates i-alirt packets and creates L1 data products."""
 
-import collections
 import logging
-from typing import Optional
+from typing import cast
 
 import xarray as xr
 
-from imap_processing.decom import decom_packets
+from imap_processing.utils import packet_file_to_datasets
 
 logger = logging.getLogger(__name__)
 
 
-def generate_xarray(
-    packet_file: str, xtce: str, time_keys: Optional[dict] = None
-) -> xr.Dataset:
+def generate_xarray(packet_file: str, xtce: str) -> dict[int, xr.Dataset]:
     """
     Generate xarray from unpacked data.
 
@@ -23,77 +20,15 @@ def generate_xarray(
         Path to the CCSDS data packet file.
     xtce : str
         Path to the XTCE packet definition file.
-    time_keys : dict
-        Keys used for creating xarray dataset.
 
     Returns
     -------
-    dataset : xarray.Dataset
-        A dataset containing the decoded data fields with 'time' as the coordinating
-        dimension.
-
-    Examples
-    --------
-    # This is an example of what the xarray dataset might look like
-    # after being processed by this function.
-
-    <xarray.Dataset>
-    Dimensions:       (SC_SCLK_SEC: 5)
-    Coordinates:
-      * SC_SCLK_SEC   (SC_SCLK_SEC) int64 322168 322169 322170 322171 322172
-    Data variables:
-        SC_MAG_STATUS (SC_SCLK_SEC) int64 0 1 0 1 0
-        SC_HIT_STATUS (SC_SCLK_SEC) int64 1 0 1 0 1
-
-    This example shows a dataset with 'SC_SCLK_SEC' as the coordinate
-    and two data variables 'SC_MAG_STATUS' and 'SC_HIT_STATUS'.
+    alirt_dict : dict
+        A dictionary of the dataset containing the decoded data fields.
     """
-    packets = decom_packets(packet_file, xtce)
-
-    logger.info(f"Decommutated {len(packets)} packets from {packet_file}.")
-
-    if time_keys is None:
-        time_keys = {
-            "SC": "SC_SCLK_SEC",
-            "HIT": "HIT_MET",
-            "MAG": "MAG_ACQ",
-            "COD_LO": "COD_LO_ACQ",
-            "COD_HI": "COD_HI_ACQ",
-            "SWE": "SWE_ACQ_SEC",
-            "SWAPI": "SWAPI_ACQ",
-        }
-
-    instruments = list(time_keys.keys())
-
-    # Initialize storage dictionary using defaultdict
-    data_storage: dict = {inst: collections.defaultdict(list) for inst in instruments}
-
-    for packet in packets:
-        for key, value in packet.data.items():
-            key_matched = False
-            for inst in instruments:
-                if key.startswith(inst):
-                    # Directly append to the list
-                    data_storage[inst][key].append(value.derived_value)
-                    key_matched = True
-                    break
-
-            if not key_matched:
-                # If after checking all instruments, none match, raise an error.
-                raise ValueError(f"Unexpected key '{key}' found in packet data.")
-
-    logger.info("Generating datasets for each instrument.")
-
-    # Generate xarray dataset for each instrument and spacecraft
-    datasets = {}
-    for inst in instruments:
-        dataset_dict = {
-            key: (time_keys[inst], data_storage[inst][key])
-            for key in data_storage[inst]
-            if key != time_keys[inst]
-        }
-        datasets[inst] = xr.Dataset(
-            dataset_dict, coords={time_keys[inst]: data_storage[inst][time_keys[inst]]}
-        )
+    alirt_dict = cast(
+        dict[int, xr.Dataset],
+        packet_file_to_datasets(packet_file, xtce, use_derived_value=False),
+    )
 
-    return datasets
+    return alirt_dict

imap_processing/tests/ialirt/unit/test_decom_ialirt.py

@@ -85,57 +85,10 @@ def test_enumerated(decom_packets_data):
 
 
 def test_generate_xarray(binary_packet_path, xtce_ialirt_path, decom_packets_data):
-    """This function checks that all instrument parameters are accounted for."""
+    """This function checks that all instrument parameters are correct length."""
 
-    instrument_lengths = {}
-    instruments = ["SC", "HIT", "MAG", "COD_LO", "COD_HI", "SWE", "SWAPI"]
+    apid = 478
+    xarray_data = generate_xarray(binary_packet_path, xtce_ialirt_path)[apid]
 
-    xarray_data = generate_xarray(binary_packet_path, xtce_ialirt_path)
-
-    # Assert that all parameters are the same
-    # between packet data and generated xarray
-    for instrument in instruments:
-        instrument_list = list(xarray_data[instrument].coords)
-        instrument_list.extend(list(xarray_data[instrument].data_vars))
-
-        # Create a dictionary of the number of parameters for each instrument
-        instrument_lengths[instrument] = len(instrument_list)
-
-        packet_data_keys_set = set(decom_packets_data[0].data.keys())
-        instrument_list_set = set(instrument_list)
-
-        # Assert that the instrument parameters from the xarray are
-        # a subset of the packet data
-        assert instrument_list_set.issubset(packet_data_keys_set) is True
-
-    # Assert that the length of parameters is the same
-    # between packet data and generated xarray
-    assert sum(instrument_lengths.values()) == len(
-        list(decom_packets_data[0].data.keys())
-    )
-
-    # Check that all the dimensions are the correct length
-    # example:len(xarray_data['HIT'].coords['HIT_SC_TICK'].values) == 32
-    for instrument in instruments:
-        for dimension_name in xarray_data[instrument].dims:
-            assert len(xarray_data[instrument].coords[dimension_name].values) == 32
-
-
-def test_unexpected_key(binary_packet_path, xtce_ialirt_path, decom_packets_data):
-    """
-    This function tests that a ValueError is raised when an unexpected
-    key is encountered.
-    """
-
-    time_keys = {
-        "UNEXPECTED": "SC_SCLK_SEC",
-        "HIT": "HIT_SC_TICK",
-        "MAG": "MAG_ACQ",
-        "COD_LO": "COD_LO_ACQ",
-        "COD_HI": "COD_HI_ACQ",
-        "SWE": "SWE_ACQ_SEC",
-        "SWAPI": "SWAPI_ACQ",
-    }
-
-    with pytest.raises(ValueError, match=r"Unexpected key '.*' found in packet data."):
-        generate_xarray(binary_packet_path, xtce_ialirt_path, time_keys=time_keys)
+    for key in xarray_data.keys():
+        assert len(xarray_data[key]) == 32

imap_processing/tests/ialirt/unit/test_process_hit.py

@@ -34,9 +34,7 @@ def xarray_data(binary_packet_path, xtce_hit_path):
     """Create xarray data"""
     apid = 1253
 
-    xarray_data = generate_xarray(
-        binary_packet_path, xtce_hit_path, time_keys={"HIT": "HIT_SC_TICK"}
-    )[apid]
+    xarray_data = generate_xarray(binary_packet_path, xtce_hit_path)[apid]
     return xarray_data