I-ALiRT - Test decom for HIT CCSDS

imap_processing/ialirt/l0/decom_ialirt.py

@@ -1,19 +1,15 @@
-"""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
 
 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 +19,12 @@ 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_SC_TICK",
-            "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 = packet_file_to_datasets(packet_file, xtce, use_derived_value=False)
 
-    return datasets
+    return alirt_dict

imap_processing/ialirt/l0/process_hit.py

@@ -0,0 +1,210 @@
+"""Functions to support HIT processing."""
+
+import logging
+from enum import Enum
+from typing import ClassVar
+
+import numpy as np
+import xarray as xr
+
+logger = logging.getLogger(__name__)
+
+
+class HITPrefixes(Enum):
+    """Create ENUM for each rate type."""
+
+    FAST_RATE_1: ClassVar[list[str]] = [
+        f"{prefix}_{i:02d}"
+        for i in range(15)
+        for prefix in ["L1A_TRIG", "IA_EVNT_TRIG", "A_EVNT_TRIG", "L3A_TRIG"]
+    ]
+    FAST_RATE_2: ClassVar[list[str]] = [
+        f"{prefix}_{i:02d}"
+        for i in range(15)
+        for prefix in ["L1B_TRIG", "IB_EVNT_TRIG", "B_EVNT_TRIG", "L3B_TRIG"]
+    ]
+    SLOW_RATE: ClassVar[list[str]] = [
+        "L1A",
+        "L2A",
+        "L3A",
+        "L1A0AHG",
+        "L1B0AHG",
+        "L1C0AHG",
+        "L4IAHG",
+        "L4OAHG",
+        "SLOW_RATE_08",
+        "SLOW_RATE_09",
+        "SLOW_RATE_10",
+        "L1A0BHG",
+        "L1B0BHG",
+        "L1C0BHG",
+        "L4IBHG",
+        "L4OBHG",
+        *[f"IALRT_RATE_{i}" for i in range(1, 21)],
+        "TRIG_IA_EVNT",
+        "TRIG_IB_EVNT",
+        "NASIDE_IALRT",
+        "NBSIDE_IALRT",
+        *[f"ERATE_{i}" for i in range(1, 6)],
+        "L12A",
+        "L123A",
+        "PENA",
+        "L12B",
+        "L123B",
+        "PENB",
+        "SLOW_RATE_51",
+        "SLOW_RATE_52",
+        "SLOW_RATE_53",
+        "SLOW_RATE_54",
+        "H_06_08",
+        "H_12_15",
+        "H_15_70",
+        "HE4_06_08",
+        "HE4_15_70",
+    ]
+
+
+def find_groups(data: xr.Dataset) -> xr.Dataset:
+    """
+    Find groups in the data.
+
+    Parameters
+    ----------
+    data : xr.Dataset
+        HIT Dataset.
+
+    Returns
+    -------
+    grouped_data : xr.Dataset
+        Grouped data.
+    """
+    subcom_range = (0, 59)
+
+    data = data.sortby("hit_sc_tick", ascending=True)
+
+    # Filter out data before the first subcom=0 and after the last subcom=59
+    start_sc_ticks = data["hit_sc_tick"][(data["hit_subcom"] == subcom_range[0]).values]
+    start_sc_tick = start_sc_ticks.data.min()
+    last_sc_ticks = data["hit_sc_tick"][([data["hit_subcom"] == subcom_range[-1]][-1])]
+    last_sc_tick = last_sc_ticks.data.max()
+
+    grouped_data = data.where(
+        (data["hit_sc_tick"] >= start_sc_tick) & (data["hit_sc_tick"] <= last_sc_tick),
+        drop=True,
+    )
+
+    # Create group labels
+    group_labels = np.searchsorted(
+        start_sc_ticks, grouped_data["hit_sc_tick"], side="right"
+    )
+    grouped_data["group"] = ("group", group_labels)
+
+    return grouped_data
+
+
+def create_l1(
+    fast_rate_1: xr.DataArray,
+    fast_rate_2: xr.DataArray,
+    slow_rate: xr.DataArray,
+) -> dict[str, float]:
+    """
+    Create L1 data dictionary.
+
+    Parameters
+    ----------
+    fast_rate_1 : xr.DataArray
+        Fast rate 1 DataArray.
+    fast_rate_2 : xr.DataArray
+        Fast rate 2 DataArray.
+    slow_rate : xr.DataArray
+        Slow rate DataArray.
+
+    Returns
+    -------
+    l1 : dict
+        Dictionary containing parsed L0 packet data.
+    """
+    fast_rate_1_dict = {
+        prefix: value
+        for prefix, value in zip(HITPrefixes.FAST_RATE_1.value, fast_rate_1.data)
+    }
+    fast_rate_2_dict = {
+        prefix: value
+        for prefix, value in zip(HITPrefixes.FAST_RATE_2.value, fast_rate_2.data)
+    }
+    slow_rate_dict = {
+        prefix: value
+        for prefix, value in zip(HITPrefixes.SLOW_RATE.value, slow_rate.data)
+    }
+
+    l1 = {**fast_rate_1_dict, **fast_rate_2_dict, **slow_rate_dict}
+
+    return l1
+
+
+def process_hit(xarray_data: xr.Dataset) -> list[dict]:
+    """
+    Create L1 data dictionary.
+
+    Parameters
+    ----------
+    xarray_data : dict(xr.Dataset)
+        Dictionary of xarray data including a single
+        set for processing.
+
+    Returns
+    -------
+    hit_data : dict
+        Dictionary final data product.
+    """
+    hit_data = []
+    grouped_data = find_groups(xarray_data)
+
+    for group in np.unique(grouped_data["group"]):
+        fast_rate_1 = grouped_data["hit_fast_rate_1"][
+            (grouped_data["group"] == group).values
+        ]
+        fast_rate_2 = grouped_data["hit_fast_rate_2"][
+            (grouped_data["group"] == group).values
+        ]
+        slow_rate = grouped_data["hit_slow_rate"][
+            (grouped_data["group"] == group).values
+        ]
+        met = int(grouped_data["hit_met"][(grouped_data["group"] == group).values][0])
+
+        # Verify that each group has 60 datapoints
+        if len(slow_rate.data) != 60:
+            logger.info(
+                f"Incorrect number of packets: {len(slow_rate.data)} "
+                f"for met {met}. Skipping."
+            )
+            continue
+
+        l1 = create_l1(fast_rate_1, fast_rate_2, slow_rate)
+
+        hit_data.append(
+            {
+                "met": met,
+                "hit_lo_energy_e_A_side": float(
+                    l1["IALRT_RATE_1"] + l1["IALRT_RATE_2"]
+                ),
+                "hit_medium_energy_e_A_side": float(
+                    l1["IALRT_RATE_5"] + l1["IALRT_RATE_6"]
+                ),
+                "hit_high_energy_e_A_side": float(l1["IALRT_RATE_7"]),
+                "hit_low_energy_e_B_side": float(
+                    l1["IALRT_RATE_11"] + l1["IALRT_RATE_12"]
+                ),
+                "hit_medium_energy_e_B_side": float(
+                    l1["IALRT_RATE_15"] + l1["IALRT_RATE_16"]
+                ),
+                "hit_high_energy_e_B_side": float(l1["IALRT_RATE_17"]),
+                "hit_medium_energy_H_omni": float(l1["H_12_15"] + l1["H_15_70"]),
+                "hit_high_energy_H_A_side": float(l1["IALRT_RATE_8"]),
+                "hit_high_energy_H_B_side": float(l1["IALRT_RATE_18"]),
+                "hit_low_energy_He_omni": float(l1["HE4_06_08"]),
+                "hit_high_energy_He_omni": float(l1["HE4_15_70"]),
+            }
+        )
+
+    return hit_data

imap_processing/ialirt/packet_definitions/ialirt.xml

@@ -557,16 +557,16 @@
 			</xtce:Parameter>
 			<!-- CoDICE LO -->
 			<!-- HIT -->
-			<xtce:Parameter name="HIT_SC_TICK" parameterTypeRef="uint32">
-				<xtce:LongDescription>Spacecraft tick</xtce:LongDescription>
+			<xtce:Parameter name="HIT_MET" parameterTypeRef="uint32">
+				<xtce:LongDescription>Mission elapsed time</xtce:LongDescription>
 			</xtce:Parameter>
 			<xtce:Parameter name="HIT_STATUS" parameterTypeRef="instrument_status">
 				<xtce:LongDescription>1 = nominal, 0 = off-nominal</xtce:LongDescription>
 			</xtce:Parameter>
-			<xtce:Parameter name="HIT_RESERVED" parameterTypeRef="uint1">
-				<xtce:LongDescription>Reserved</xtce:LongDescription>
+			<xtce:Parameter name="HIT_SPARE" parameterTypeRef="uint1">
+				<xtce:LongDescription>Spare</xtce:LongDescription>
 			</xtce:Parameter>
-			<xtce:Parameter name="HIT_COUNTER" parameterTypeRef="uint6">
+			<xtce:Parameter name="HIT_SUBCOM" parameterTypeRef="uint6">
 				<xtce:LongDescription>Seconds counter</xtce:LongDescription>
 			</xtce:Parameter>
 			<xtce:Parameter name="HIT_FAST_RATE_1" parameterTypeRef="uint16">
@@ -717,10 +717,10 @@
 					<xtce:ParameterRefEntry parameterRef="COD_LO_SPARE_0"/>
 					<!-- CoDICE LO -->
 					<!-- HIT -->
-					<xtce:ParameterRefEntry parameterRef="HIT_SC_TICK"/>
+					<xtce:ParameterRefEntry parameterRef="HIT_MET"/>
 					<xtce:ParameterRefEntry parameterRef="HIT_STATUS"/>
-					<xtce:ParameterRefEntry parameterRef="HIT_RESERVED"/>
-					<xtce:ParameterRefEntry parameterRef="HIT_COUNTER"/>
+					<xtce:ParameterRefEntry parameterRef="HIT_SPARE"/>
+					<xtce:ParameterRefEntry parameterRef="HIT_SUBCOM"/>
 					<xtce:ParameterRefEntry parameterRef="HIT_FAST_RATE_1"/>
 					<xtce:ParameterRefEntry parameterRef="HIT_FAST_RATE_2"/>
 					<xtce:ParameterRefEntry parameterRef="HIT_SLOW_RATE"/>