WIP, reset me

icepyx/core/granules.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import datetime
 import requests
 import time
@@ -7,12 +9,14 @@
 import numpy as np
 import os
 import pprint
-from xml.etree import ElementTree as ET
 import zipfile
+from typing import Sequence
+from xml.etree import ElementTree as ET
 
 import icepyx.core.APIformatting as apifmt
-from icepyx.core.auth import EarthdataAuthMixin
 import icepyx.core.exceptions
+from icepyx.core.auth import EarthdataAuthMixin
+from icepyx.core.types import CMRParams
 
 
 def info(grans):
@@ -151,6 +155,8 @@ class Granules(EarthdataAuthMixin):
     Granules object
     """
 
+    avail: list
+
     def __init__(
         self,
         # avail=[],
@@ -169,14 +175,19 @@ def __init__(
     # ----------------------------------------------------------------------
     # Methods
 
-    def get_avail(self, CMRparams, reqparams, cloud=False):
+    def get_avail(
+        self,
+        CMRparams: CMRParams | None,
+        reqparams,
+        cloud=False,
+    ):
         """
         Get a list of available granules for the query object's parameters.
         Generates the `avail` attribute of the granules object.
 
         Parameters
         ----------
-        CMRparams : dictionary
+        CMRparams :
             Dictionary of properly formatted CMR search parameters.
         reqparams : dictionary
             Dictionary of properly formatted parameters required for searching, ordering,
@@ -257,18 +268,18 @@ def get_avail(self, CMRparams, reqparams, cloud=False):
             len(self.avail) > 0
         ), "Your search returned no results; try different search parameters"
 
-    # DevNote: currently, default subsetting DOES NOT include variable subsetting,
+    # NOTE: currently, default subsetting DOES NOT include variable subsetting,
     # only spatial and temporal
-    # DevGoal: add kwargs to allow subsetting and more control over request options.
+    # GOAL: add kwargs to allow subsetting and more control over request options.
     def place_order(
         self,
-        CMRparams,
+        CMRparams: CMRParams,
         reqparams,
         subsetparams,
         verbose,
         subset=True,
         geom_filepath=None,
-    ):  # , **kwargs):
+    ):
         """
         Place an order for the available granules for the query object.
         Adds the list of zipped files (orders) to the granules data object (which is
@@ -277,7 +288,7 @@ def place_order(
 
         Parameters
         ----------
-        CMRparams : dictionary
+        CMRparams :
             Dictionary of properly formatted CMR search parameters.
         reqparams : dictionary
             Dictionary of properly formatted parameters required for searching, ordering,
@@ -310,6 +321,7 @@ def place_order(
         """
 
         base_url = "https://n5eil02u.ecs.nsidc.org/egi/request"
+        breakpoint()
 
         self.get_avail(CMRparams, reqparams)
 
@@ -331,6 +343,7 @@ def place_order(
             " granules.",
         )
 
+        pagenums: Sequence
         if reqparams["page_num"] > 0:
             pagenums = [reqparams["page_num"]]
         else:
@@ -362,21 +375,34 @@ def place_order(
             request.raise_for_status()
             esir_root = ET.fromstring(request.content)
             if verbose is True:
-                print("Order request URL: ", requests.utils.unquote(request.url))
+                print("Order request URL: ", requests.compat.unquote(request.url))
                 print(
                     "Order request response XML content: ",
                     request.content.decode("utf-8"),
                 )
 
             # Look up order ID
-            orderlist = []
+            orderlist: list[str | None] = []
             for order in esir_root.findall("./order/"):
                 # if verbose is True:
                 #     print(order)
                 orderlist.append(order.text)
+
+            if not orderlist:
+                raise RuntimeError(
+                    "Expected to find at least one order in response:"
+                    f" {request.content}"
+                )
+
             orderID = orderlist[0]
             print("order ID: ", orderID)
 
+            if not orderID:
+                raise RuntimeError(
+                    "Expected to find an order ID in response, found None:"
+                    f" {request.content}"
+                )
+
             # Create status URL
             statusURL = base_url + "/" + orderID
             if verbose is True:
@@ -393,9 +419,9 @@ def place_order(
             # Raise bad request: Loop will stop for bad response code.
             request_response.raise_for_status()
             request_root = ET.fromstring(request_response.content)
-            statuslist = []
-            for status in request_root.findall("./requestStatus/"):
-                statuslist.append(status.text)
+            statuslist: list[str | None] = []
+            for status_element in request_root.findall("./requestStatus/"):
+                statuslist.append(status_element.text)
             status = statuslist[0]
             print("Initial status of your order request at NSIDC is: ", status)
 
@@ -416,8 +442,9 @@ def place_order(
 
                 # find status
                 statuslist = []
-                for status in loop_root.findall("./requestStatus/"):
-                    statuslist.append(status.text)
+                for status_element in loop_root.findall("./requestStatus/"):
+                    if status_element.text:
+                        statuslist.append(status_element.text)
                 status = statuslist[0]
                 # print('Retry request status is: ', status)
                 if status == "pending" or status == "processing":

icepyx/core/query.py

@@ -11,6 +11,7 @@
 import icepyx.core.spatial as spat
 import icepyx.core.temporal as tp
 import icepyx.core.validate_inputs as val
+from icepyx.core.types import CMRParams
 from icepyx.core.variables import Variables as Variables
 from icepyx.core.visualization import Visualize
 
@@ -531,7 +532,7 @@ def tracks(self):
             return sorted(set(self._tracks))
 
     @property
-    def CMRparams(self):
+    def CMRparams(self) -> CMRParams:
         """
         Display the CMR key:value pairs that will be submitted.
         It generates the dictionary if it does not already exist.

icepyx/core/types.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import TypedDict
+
+
+class CMRParamsBase(TypedDict):
+    temporal: str
+
+
+class CMRParamsWithBbox(CMRParamsBase):
+    bounding_box: str
+
+
+class CMRParamsWithPolygon(CMRParamsBase):
+    polygon: str
+
+
+CMRParams = CMRParamsWithBbox | CMRParamsWithPolygon

icepyx/core/visualization.py

@@ -466,7 +466,7 @@ def parallel_request_OA(self) -> da.array:
             OA_data_da = da.concatenate(requested_OA_data, axis=0)
             return OA_data_da
 
-    def viz_elevation(self) -> (hv.DynamicMap, hv.Layout):
+    def viz_elevation(self) -> tuple[hv.DynamicMap, hv.Layout]:
         """
         Visualize elevation requested from OpenAltimetry API using datashader based on cycles
         https://holoviz.org/tutorial/Large_Data.html

requirements.txt

@@ -14,3 +14,4 @@ requests
 s3fs
 shapely
 xarray
+pydantic