Add mypy Pre-Commit For hi & hit Pt.10 (IMAP-Science-Operations-Center#674)

daralynnrhode · web-flow · commit d556849de75f · 2024-07-04T11:59:16.000-06:00
* mypy set up for hit and hi

* mypy set up for hit and hi

* mypy for hi

* mypy for hit

* fixing test failures

* g

* changes

* inline ignore comments for merge conflict

* mypyfixes for merge
diff --git a/imap_processing/decom.py b/imap_processing/decom.py
@@ -5,10 +5,15 @@
 to decommutate CCSDS packet data using a given XTCE packet definition.
 """
 
+from pathlib import Path
+from typing import Union
+
 from space_packet_parser import parser, xtcedef
 
 
-def decom_packets(packet_file: str, xtce_packet_definition: str) -> list:
+def decom_packets(
+    packet_file: Union[str, Path], xtce_packet_definition: Union[str, Path]
+) -> list:
     """
     Unpack CCSDS data packet.
 
diff --git a/imap_processing/hi/l0/decom_hi.py b/imap_processing/hi/l0/decom_hi.py
@@ -3,7 +3,7 @@
 from imap_processing import decom, imap_module_directory
 
 
-def decom_packets(packet_file_path: str):
+def decom_packets(packet_file_path: str) -> list:
     """
     Decom IMAP-Hi data using its packet definition.
 
@@ -20,4 +20,5 @@ def decom_packets(packet_file_path: str):
     packet_def_file = (
         imap_module_directory / "hi/packet_definitions/hi_packet_definition.xml"
     )
-    return decom.decom_packets(packet_file_path, packet_def_file)
+    decom_file_packets: list = decom.decom_packets(packet_file_path, packet_def_file)
+    return decom_file_packets
diff --git a/imap_processing/hi/l1a/hi_l1a.py b/imap_processing/hi/l1a/hi_l1a.py
@@ -1,8 +1,6 @@
 """IMAP-HI L1A processing module."""
 
 import logging
-from pathlib import Path
-from typing import Union
 
 from imap_processing.hi.l0 import decom_hi
 from imap_processing.hi.l1a.histogram import create_dataset as hist_create_dataset
@@ -14,7 +12,7 @@
 logger = logging.getLogger(__name__)
 
 
-def hi_l1a(packet_file_path: Union[str, Path], data_version: str):
+def hi_l1a(packet_file_path: str, data_version: str) -> list:
     """
     Will process IMAP raw data to l1a.
 
diff --git a/imap_processing/hi/l1a/science_direct_event.py b/imap_processing/hi/l1a/science_direct_event.py
@@ -102,18 +102,17 @@ def parse_direct_event(event_data: str) -> dict:
     dict
         Parsed event data.
     """
-    event_type = int(event_data[:2])
+    event_type = int(event_data[:2], 2)
     metaevent = 0
     if event_type == metaevent:
         # parse metaevent
-        event_type = event_data[:2]
         esa_step = event_data[2:6]
         subseconds = event_data[6:16]
         seconds = event_data[16:]
 
         # return parsed metaevent data
         return {
-            "start_bitmask_data": int(event_type, 2),
+            "start_bitmask_data": event_type,
             "esa_step": int(esa_step, 2),
             "subseconds": int(subseconds, 2),
             "seconds": int(seconds, 2),
@@ -176,7 +175,7 @@ def create_dataset(de_data_list: list, packet_met_time: list) -> xr.Dataset:
         Xarray dataset.
     """
     # These are the variables that we will store in the dataset
-    data_dict = {
+    data_dict: dict = {
         "epoch": list(),
         "event_met": list(),
         "ccsds_met": list(),
diff --git a/imap_processing/hi/l1b/hi_l1b.py b/imap_processing/hi/l1b/hi_l1b.py
@@ -16,7 +16,7 @@
 CDF_MANAGER.load_variable_attributes("imap_hi_variable_attrs.yaml")
 
 
-def hi_l1b(l1a_dataset: xr.Dataset, data_version: str):
+def hi_l1b(l1a_dataset: xr.Dataset, data_version: str) -> xr.Dataset:
     """
     High level IMAP-HI L1B processing function.
 
@@ -43,14 +43,16 @@ def hi_l1b(l1a_dataset: xr.Dataset, data_version: str):
     if logical_source_parts[-1].endswith("hk"):
         # if packet_enum in (HIAPID.H45_APP_NHK, HIAPID.H90_APP_NHK):
         packet_enum = HIAPID(l1a_dataset["pkt_apid"].data[0])
-        conversion_table_path = (
+        conversion_table_path = str(
             imap_module_directory / "hi" / "l1b" / "hi_eng_unit_convert_table.csv"
         )
         l1b_dataset = convert_raw_to_eu(
             l1a_dataset,
             conversion_table_path=conversion_table_path,
             packet_name=packet_enum.name,
-            comment="#",
+            comment="#",  # type: ignore[arg-type]
+            # Todo error, Argument "comment" to "convert_raw_to_eu" has incompatible
+            # type "str"; expected "dict[Any, Any]"
             converters={"mnemonic": str.lower},
         )
 
@@ -77,7 +79,7 @@ def hi_l1b(l1a_dataset: xr.Dataset, data_version: str):
     return l1b_dataset
 
 
-def annotate_direct_events(l1a_dataset):
+def annotate_direct_events(l1a_dataset: xr.Dataset) -> xr.Dataset:
     """
     Perform Hi L1B processing on direct event data.
 
diff --git a/imap_processing/hi/l1c/hi_l1c.py b/imap_processing/hi/l1c/hi_l1c.py
@@ -1,8 +1,8 @@
 """IMAP-HI l1c processing module."""
 
 import logging
-from pathlib import Path
-from typing import Union
+from collections.abc import Sized
+from typing import Optional
 
 import numpy as np
 import xarray as xr
@@ -13,7 +13,7 @@
 logger = logging.getLogger(__name__)
 
 
-def hi_l1c(dependencies: list, data_version: str):
+def hi_l1c(dependencies: list, data_version: str) -> xr.Dataset:
     """
     High level IMAP-Hi l1c processing function.
 
@@ -52,7 +52,7 @@ def hi_l1c(dependencies: list, data_version: str):
     return l1c_dataset
 
 
-def generate_pset_dataset(de_dataset: Union[Path, str]) -> xr.Dataset:
+def generate_pset_dataset(de_dataset: xr.Dataset) -> xr.Dataset:
     """
     Will process IMAP-Hi l1b product into a l1c pset xarray dataset.
 
@@ -77,7 +77,7 @@ def generate_pset_dataset(de_dataset: Union[Path, str]) -> xr.Dataset:
     return pset_dataset
 
 
-def allocate_pset_dataset(n_esa_steps: int, sensor_str: str):
+def allocate_pset_dataset(n_esa_steps: int, sensor_str: str) -> xr.Dataset:
     """
     Allocate an empty xarray.Dataset.
 
@@ -191,7 +191,9 @@ def allocate_pset_dataset(n_esa_steps: int, sensor_str: str):
     return dataset
 
 
-def full_dataarray(name, attrs, coords: dict, shape=None):
+def full_dataarray(
+    name: str, attrs: dict, coords: dict, shape: Optional[Sized] = None
+) -> xr.DataArray:
     """
     Generate an empty xarray.DataArray with appropriate attributes.
 
diff --git a/imap_processing/hi/utils.py b/imap_processing/hi/utils.py
@@ -15,7 +15,7 @@ class HIAPID(IntEnum):
     H90_SCI_DE = 834
 
     @property
-    def sensor(self):
+    def sensor(self) -> str:
         """
         Define the sensor name attribute for this class.
 
diff --git a/imap_processing/hit/hit_cdf_attrs.py b/imap_processing/hit/hit_cdf_attrs.py
@@ -863,5 +863,9 @@
 }
 
 # Dictionaries of complete L1A and L1B housekeeping attributes
-l1a_hk_attrs.update(l1a_l1b_hk_attrs)
-l1b_hk_attrs.update(l1a_l1b_hk_attrs)
+l1a_hk_attrs.update(l1a_l1b_hk_attrs)  # type: ignore[arg-type]
+l1b_hk_attrs.update(l1a_l1b_hk_attrs)  # type: ignore[arg-type]
+
+# TODO Fix mypy error above.
+# Argument 1 to "update" of "MutableMapping" has incompatible type
+# "dict[str, AttrBase]"; expected "SupportsKeysAndGetItem[str, ScienceAttrs]"
diff --git a/imap_processing/hit/l0/data_classes/housekeeping.py b/imap_processing/hit/l0/data_classes/housekeeping.py
@@ -3,6 +3,7 @@
 from dataclasses import dataclass
 
 import numpy as np
+import space_packet_parser
 
 from imap_processing.ccsds.ccsds_data import CcsdsData
 from imap_processing.hit.l0.utils.hit_base import HITBase
@@ -210,13 +211,18 @@ class Housekeeping(HITBase):
     L34B_BIAS: int
     EBOX_P2D0VD: int
 
-    def __init__(self, packet, software_version: str, packet_file_name: str):
+    def __init__(
+        self,
+        packet: space_packet_parser.parser.Packet,
+        software_version: str,
+        packet_file_name: str,
+    ):
         """Housekeeping Data class initialization method."""
         super().__init__(software_version, packet_file_name, CcsdsData(packet.header))
         self.parse_data(packet)
         self._parse_leak()
 
-    def _parse_leak(self):
+    def _parse_leak(self) -> None:
         """Parse each current leakage field and put into an array."""
         # Each Leak field is 10 bits long
         leak_bit_length = 10
diff --git a/imap_processing/hit/l0/utils/hit_base.py b/imap_processing/hit/l0/utils/hit_base.py
@@ -2,6 +2,8 @@
 
 from dataclasses import dataclass, fields
 
+import space_packet_parser
+
 from imap_processing.ccsds.ccsds_data import CcsdsData
 
 
@@ -29,13 +31,13 @@ class HITBase:
     packet_file_name: str
     ccsds_header: CcsdsData
 
-    def parse_data(self, packet):
+    def parse_data(self, packet: space_packet_parser.parser.Packet) -> None:
         """
         Parse Lo L0 packet data.
 
         Parameters
         ----------
-        packet : dict
+        packet : space_packet_parser.parser.Packet
             A single Lo L0 packet from space packet parser.
         """
         attributes = [field.name for field in fields(self)]
diff --git a/imap_processing/hit/l1a/hit_l1a.py b/imap_processing/hit/l1a/hit_l1a.py
@@ -1,17 +1,16 @@
 """Decommutate HIT CCSDS data and create L1a data products."""
 
 import logging
-import typing
 from collections import defaultdict
 from dataclasses import fields
 from enum import IntEnum
-from pathlib import Path
 
 import numpy as np
 import xarray as xr
 
 from imap_processing import decom, imap_module_directory, utils
 from imap_processing.cdf.imap_cdf_manager import ImapCdfAttributes
+from imap_processing.cdf.utils import met_to_j2000ns
 from imap_processing.hit.l0.data_classes.housekeeping import Housekeeping
 
 logger = logging.getLogger(__name__)
@@ -38,7 +37,7 @@ class HitAPID(IntEnum):
     HIT_IALRT = 1253
 
 
-def hit_l1a(packet_file: typing.Union[Path, str], data_version: str):
+def hit_l1a(packet_file: str, data_version: str) -> list:
     """
     Will process HIT L0 data into L1A data products.
 
@@ -51,7 +50,7 @@ def hit_l1a(packet_file: typing.Union[Path, str], data_version: str):
 
     Returns
     -------
-    cdf_filepaths : dict
+    cdf_filepaths : list
         List of file paths to CDF data product files.
     """
     # Decom, sort, and group packets by apid
@@ -71,7 +70,7 @@ def hit_l1a(packet_file: typing.Union[Path, str], data_version: str):
     return list(datasets.values())
 
 
-def decom_packets(packet_file: str):
+def decom_packets(packet_file: str) -> list:
     """
     Unpack and decode packets using CCSDS file and XTCE packet definitions.
 
@@ -88,12 +87,12 @@ def decom_packets(packet_file: str):
     # TODO: update path to use a combined packets xtce file
     xtce_file = imap_module_directory / "hit/packet_definitions/P_HIT_HSKP.xml"
     logger.debug(f"Unpacking {packet_file} using xtce definitions in {xtce_file}")
-    unpacked_packets = decom.decom_packets(packet_file, xtce_file)
+    unpacked_packets: list = decom.decom_packets(packet_file, xtce_file)
     logger.debug(f"{packet_file} unpacked")
     return unpacked_packets
 
 
-def group_data(unpacked_data: list):
+def group_data(unpacked_data: list) -> dict:
     """
     Group data by apid.
 
@@ -108,7 +107,7 @@ def group_data(unpacked_data: list):
         Grouped data by apid.
     """
     logger.debug("Grouping packet values for each apid")
-    grouped_data = utils.group_by_apid(unpacked_data)
+    grouped_data: dict = utils.group_by_apid(unpacked_data)
 
     # Create data classes for each packet
     for apid in grouped_data:
@@ -125,7 +124,7 @@ def group_data(unpacked_data: list):
     return grouped_data
 
 
-def create_datasets(data: dict, attr_mgr: ImapCdfAttributes):
+def create_datasets(data: dict, attr_mgr: ImapCdfAttributes) -> dict:
     """
     Create a dataset for each APID in the data.
 
@@ -178,7 +177,7 @@ def create_datasets(data: dict, attr_mgr: ImapCdfAttributes):
                 metadata_arrays[data_key].append(field_value)
 
         # Convert integers into datetime64[s]
-        epoch_converted_times = utils.met_to_j2000ns(metadata_arrays["shcoarse"])
+        epoch_converted_times = met_to_j2000ns(metadata_arrays["shcoarse"])
 
         # Create xarray data arrays for dependencies
         epoch_time = xr.DataArray(
@@ -214,27 +213,39 @@ def create_datasets(data: dict, attr_mgr: ImapCdfAttributes):
         )
 
         # Create xarray data array for each metadata field
-        for field, data in metadata_arrays.items():
-            if field not in skip_keys:
+        for field, data in metadata_arrays.items():  # type: ignore[assignment]
+            # TODO Error, Incompatible types in assignment
+            # (expression has type "str", variable has type "Field[Any]")
+            # AND
+            # Incompatible types in assignment
+            # (expression has type "list[Any]", variable has type "dict[Any, Any]")
+            if field not in skip_keys:  # type: ignore[comparison-overlap]
+                # TODO Error, Non-overlapping container check
+                # (element type: "Field[Any]", container item type: "str")
+
                 # Create a list of all the dimensions using the DEPEND_I keys in the
                 # attributes
                 dims = [
                     value
-                    for key, value in attr_mgr.get_variable_attributes(field).items()
+                    for key, value in attr_mgr.get_variable_attributes(field).items()  # type: ignore[arg-type]
                     if "DEPEND" in key
                 ]
-                if field == "leak_i":
+                if field == "leak_i":  # type: ignore[comparison-overlap]
+                    # TODO Error,  Non-overlapping equality check
+                    # (left operand type: "Field[Any]",
+                    # right operand type: "Literal['leak_i']")
+
                     # 2D array - needs two dims
                     dataset[field] = xr.DataArray(
                         data,
                         dims=dims,
-                        attrs=attr_mgr.get_variable_attributes(field),
+                        attrs=attr_mgr.get_variable_attributes(field),  # type: ignore[arg-type]
                     )
                 else:
                     dataset[field] = xr.DataArray(
                         data,
                         dims=dims,
-                        attrs=attr_mgr.get_variable_attributes(field),
+                        attrs=attr_mgr.get_variable_attributes(field),  # type: ignore[arg-type]
                     )
         processed_data[apid] = dataset
     logger.info("HIT L1A datasets created")
diff --git a/imap_processing/hit/l1b/hit_l1b.py b/imap_processing/hit/l1b/hit_l1b.py
diff --git a/imap_processing/ultra/l1c/histogram.py b/imap_processing/ultra/l1c/histogram.py
diff --git a/imap_processing/utils.py b/imap_processing/utils.py
diff --git a/pyproject.toml b/pyproject.toml
