A collection of "low-hanging fruit" refactors

icepyx/core/cmr.py

@@ -0,0 +1,3 @@
+from typing import Final
+
+CMR_PROVIDER: Final = "NSIDC_CPRD"

icepyx/core/granules.py

@@ -7,6 +7,7 @@
 import pprint
 import re
 import time
+from typing import Union
 from xml.etree import ElementTree as ET
 import zipfile
 
@@ -16,6 +17,7 @@
 
 import icepyx.core.APIformatting as apifmt
 from icepyx.core.auth import EarthdataAuthMixin
+from icepyx.core.cmr import CMR_PROVIDER
 import icepyx.core.exceptions
 from icepyx.core.types import (
     CMRParams,
@@ -25,7 +27,7 @@
 from icepyx.core.urls import DOWNLOAD_BASE_URL, GRANULE_SEARCH_BASE_URL, ORDER_BASE_URL
 
 
-def info(grans):
+def info(grans: list[dict]) -> dict[str, Union[int, float]]:
     """
     Return some basic summary information about a set of granules for an
     query object. Granule info may be from a list of those available
@@ -45,7 +47,14 @@ def info(grans):
 
 # DevNote: currently this fn is not tested
 # DevNote: could add flag to separate ascending and descending orbits based on ATL03 granule region
-def gran_IDs(grans, ids=False, cycles=False, tracks=False, dates=False, cloud=False):
+def gran_IDs(
+    grans: list[dict],
+    ids: bool = False,
+    cycles: bool = False,
+    tracks: bool = False,
+    dates: bool = False,
+    cloud: bool = False,
+):
     """
     Returns a list of granule information for each granule dictionary
     in the input list of granule dictionaries.
@@ -54,17 +63,17 @@ def gran_IDs(grans, ids=False, cycles=False, tracks=False, dates=False, cloud=Fa
 
     Parameters
     ----------
-    grans : list of dictionaries
+    grans :
         List of input granule json dictionaries. Must have key "producer_granule_id"
-    ids: boolean, default True
+    ids :
         Return a list of the available granule IDs for the granule dictionary
-    cycles : boolean, default False
+    cycles :
         Return a list of the available orbital cycles for the granule dictionary
-    tracks : boolean, default False
+    tracks :
         Return a list of the available Reference Ground Tracks (RGTs) for the granule dictionary
-    dates : boolean, default False
+    dates :
         Return a list of the available dates for the granule dictionary.
-    cloud : boolean, default False
+    cloud :
         Return a a list of AWS s3 urls for the available granules in the granule dictionary.
     """
     assert len(grans) > 0, "Your data object has no granules associated with it"
@@ -226,7 +235,7 @@ def get_avail(
         params = apifmt.combine_params(
             CMRparams,
             {k: reqparams[k] for k in ["short_name", "version", "page_size"]},
-            {"provider": "NSIDC_CPRD"},
+            {"provider": CMR_PROVIDER},
         )
 
         cmr_search_after = None

icepyx/core/is2ref.py

@@ -1,4 +1,5 @@
 import json
+from typing import Optional
 import warnings
 from xml.etree import ElementTree as ET
 
@@ -20,38 +21,38 @@
         "A valid product string was not provided. "
         "Check user input, if given, or file metadata."
     )
-    if isinstance(product, str):
-        product = str.upper(product)
-        assert product in [
-            "ATL01",
-            "ATL02",
-            "ATL03",
-            "ATL04",
-            "ATL06",
-            "ATL07",
-            "ATL07QL",
-            "ATL08",
-            "ATL09",
-            "ATL09QL",
-            "ATL10",
-            "ATL11",
-            "ATL12",
-            "ATL13",
-            "ATL14",
-            "ATL15",
-            "ATL16",
-            "ATL17",
-            "ATL19",
-            "ATL20",
-            "ATL21",
-            "ATL23",
-        ], error_msg
-    else:
+    if not isinstance(product, str):
         raise TypeError(error_msg)
+
+    product = str.upper(product)
+    assert product in [
+        "ATL01",
+        "ATL02",
+        "ATL03",
+        "ATL04",
+        "ATL06",
+        "ATL07",
+        "ATL07QL",
+        "ATL08",
+        "ATL09",
+        "ATL09QL",
+        "ATL10",
+        "ATL11",
+        "ATL12",
+        "ATL13",
+        "ATL14",
+        "ATL15",
+        "ATL16",
+        "ATL17",
+        "ATL19",
+        "ATL20",
+        "ATL21",
+        "ATL23",
+    ], error_msg
     return product
 
 
-def _validate_OA_product(product):
+def _validate_OA_product(product) -> str:
     """
     Confirm a valid ICESat-2 product was specified
     """
@@ -74,7 +75,7 @@
 
 
 # DevNote: test for this function is commented out; dates in some of the values were causing the test to fail...
-def about_product(prod):
+def about_product(prod: str) -> dict:
     """
     Ping Earthdata to get metadata about the product of interest (the collection).
 
@@ -170,7 +171,7 @@
 
 # DevGoal: populate this with default variable lists for all of the products!
 # DevGoal: add a test for this function (to make sure it returns the right list, but also to deal with product not being in the list, though it should since it was checked as valid earlier...)
-def _default_varlists(product):
+def _default_varlists(product) -> list[str]:
     """
     Return a list of default variables to select and send to the NSIDC subsetter.
     """
@@ -276,7 +277,7 @@
 # a faster version using pandas map (instead of apply) is available in SlideRule:
 # https://github.com/SlideRuleEarth/sliderule/issues/388
 # https://github.com/SlideRuleEarth/sliderule/commit/46cceac0e5f6d0a580933d399a6239bc911757f3
-def gt2spot(gt, sc_orient):
+def gt2spot(gt, sc_orient) -> np.uint8:
     warnings.warn(
         "icepyx versions 0.8.0 and earlier used an incorrect spot number calculation."
         "As a result, computations depending on spot number may be incorrect and should be redone."
@@ -295,6 +296,7 @@
     gr_lr = gt[3]
 
     # spacecraft oriented forward
+    spot: Optional[int] = None
     if sc_orient == 1:
         if gr_num == 1:
             if gr_lr == "l":
@@ -330,13 +332,13 @@
             elif gr_lr == "r":
                 spot = 6
 
-    if "spot" not in locals():
+    if spot is None:
         raise ValueError("Could not compute the spot number.")
 
     return np.uint8(spot)
 
 
-def latest_version(product):
+def latest_version(product) -> str:
     """
     Determine the most recent version available for the given product.
 
@@ -350,7 +352,7 @@
     return max([entry["version_id"] for entry in _about_product["feed"]["entry"]])
 
 
-def extract_product(filepath, auth=None):
+def extract_product(filepath, auth=None) -> str:
     """
     Read the product type from the metadata of the file. Valid for local or s3 files, but must
     provide an auth object if reading from s3. Return the product as a string.
@@ -396,7 +398,7 @@
     return product
 
 
-def extract_version(filepath, auth=None):
+def extract_version(filepath, auth=None) -> str:
     """
     Read the version from the metadata of the file. Valid for local or s3 files, but must
     provide an auth object if reading from s3. Return the version as a string.
@@ -423,30 +425,33 @@
         f = h5py.File(filepath, "r")
 
     # Read the version information
+    version_str: str
     try:
-        version = f["METADATA"]["DatasetIdentification"].attrs["VersionID"]
+        version = f["METADATA"]["DatasetIdentification"].attrs["VersionID"]  # pyright: ignore[reportIndexIssue]
         if isinstance(version, np.ndarray):
             # ATL14 stores the version as an array ['00x']
-            version = version[0]
-        if isinstance(version, bytes):
-            version = version.decode()
-
-    except KeyError as e:
+            version_str = version[0]
+        elif isinstance(version, bytes):
+            version_str = version.decode()
+        else:
+            raise TypeError(f"Unexpected type {version=}")
+    except (KeyError, TypeError) as e:
         raise Exception(
             "Unable to parse the version from file metadata"
         ).with_traceback(e.__traceback__)
+    finally:
+        # Close the file reader
+        f.close()
 
     # catch cases where the version number is an invalid string
     # e.g. a VersionID of "SET_BY_PGE", causing issues where version needs to be a valid number
     try:
-        float(version)
+        float(version)  # pyright: ignore[reportArgumentType]
     except ValueError:
         raise Exception(
             "There is an underlying issue with the version information"
             "provided in the metadata of this file."
             "Consider setting the version manually for further processing."
         )
 
-    # Close the file reader
-    f.close()
-    return version
+    return version_str