[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

icepyx/__init__.py

@@ -1,4 +1,5 @@
 from _icepyx_version import version as __version__
+
 from icepyx.core.base_query import BaseQuery, GenQuery
 from icepyx.core.query import Query
 from icepyx.core.read import Read

icepyx/core/harmony.py

@@ -67,7 +67,7 @@ def _place_order(
             spatial=spatial,  # type: ignore[arg-type]
             temporal=temporal,  # type: ignore[arg-type]
             # shape=shape,  # type: ignore[arg-type]
-            granule_name=granule_name
+            granule_name=granule_name,
         )
 
         if not request.is_valid():
@@ -116,7 +116,7 @@ def place_order(
             spatial=spatial,
             temporal=temporal,
             shape=shape,
-            granule_name=granule_name
+            granule_name=granule_name,
         )
 
         # Append this job to the list of job ids.

icepyx/core/query.py

@@ -30,13 +30,11 @@
 from icepyx.core.variables import Variables
 
 
-class DataOrder():
-
+class DataOrder:
     HARMONY_BASE_URL = "https://harmony.earthdata.nasa.gov/workflow-ui/"
 
     def __init__(self, job_id, type, granules, harmony_client):
-        """Initialize a DataOrder object. This object represents an order for Harmony.
-        """
+        """Initialize a DataOrder object. This object represents an order for Harmony."""
         self.job_id = job_id
         self.harmony_api = harmony_client
         self.granules = granules
@@ -85,7 +83,9 @@ def status(self):
 
     def download(self, path, overwrite=False):
         if self.type == "subset":
-            return self.harmony_api.download_granules(download_dir=path, overwrite=overwrite)
+            return self.harmony_api.download_granules(
+                download_dir=path, overwrite=overwrite
+            )
         else:
             return earthaccess.download(self.granules, local_path=path)
 
@@ -165,9 +165,10 @@ def _get_concept_id(self, product, version) -> Union[str, None]:
             return collections[0].concept_id()
         else:
             return None
+
     @property
     def order_vars(self) -> list[str]:
-        """This used to print the list of vasriables for subsetting, Harmony doesn't provide that 
+        """This used to print the list of vasriables for subsetting, Harmony doesn't provide that
         we do need to implement a class that gets the variables even if it'sm only for listing.
         """
         if self.product:
@@ -328,7 +329,6 @@ def _order_subset_granules(self) -> str:
             version=self._version,
         )
 
-
         if concept_id is None:
             raise ValueError(
                 f"Could not find concept ID for {self._prod} v{self._version}"
@@ -364,13 +364,14 @@ def _order_subset_granules(self) -> str:
                 # Polygons must be passed to `harmony-py` as a path to a valid
                 # shapefile (json, geojson, kml, shz, or zip). Create a temporary
                 # directory to store this file for the harmony order.
-            else: 
-                raise NotImplementedError("Only bounding box and polygon spatial subsetting is supported.")
+            else:
+                raise NotImplementedError(
+                    "Only bounding box and polygon spatial subsetting is supported."
+                )
         else:
             if harmony_temporal is None:
                 raise ValueError("No temporal or spatial parameters provided.")
 
-
         job_id = self.harmony_api.place_order(
             concept_id=concept_id,
             temporal=harmony_temporal,
@@ -383,13 +384,21 @@ def get_granule_links(self, cloud_hosted=False) -> list[str]:
         links = []
         for granule in self.granules.avail:
             for link in granule["links"]:
-                if cloud_hosted and link["rel"] == "http://esipfed.org/ns/fedsearch/1.1/s3#" or ((link["rel"] == "http://esipfed.org/ns/fedsearch/1.1/data#") and 
-                     ("type" in link and link["type"] in ["application/x-hdf5",
-                                                           "application/x-hdfeos"])):
+                if (
+                    cloud_hosted
+                    and link["rel"] == "http://esipfed.org/ns/fedsearch/1.1/s3#"
+                    or (
+                        (link["rel"] == "http://esipfed.org/ns/fedsearch/1.1/data#")
+                        and (
+                            "type" in link
+                            and link["type"]
+                            in ["application/x-hdf5", "application/x-hdfeos"]
+                        )
+                    )
+                ):
                     links.append(link["href"])
         return links
 
-
     def _order_whole_granules(self, cloud_hosted=False, path="./") -> list[str]:
         """
         Downloads the whole granules for the query object. This is not an asnc operation
@@ -409,7 +418,6 @@ def _order_whole_granules(self, cloud_hosted=False, path="./") -> list[str]:
         files = earthaccess.download(links, local_path=path)
         return files
 
-
     def order_granules(self, subset=True) -> DataOrder:
         """
         Place an order for the available granules for the query object.
@@ -438,8 +446,10 @@ def order_granules(self, subset=True) -> DataOrder:
         Your harmony order is:  complete
         """
         if subset:
-            job_id =self._order_subset_granules()
-            self.last_order = DataOrder(job_id, "subset", self.granules, self.harmony_api)
+            job_id = self._order_subset_granules()
+            self.last_order = DataOrder(
+                job_id, "subset", self.granules, self.harmony_api
+            )
             return self.last_order
         else:
             files = self._order_whole_granules()
@@ -465,12 +475,14 @@ def download_granules(
         status = self.last_order.status()
         if status["status"] == "running" or status["status"] == "accepted":
             print(
-            (
-                "Your harmony job status is still "
-                f"{status['status']}. Please continue waiting... this may take a few moments."
-            )
+                (
+                    "Your harmony job status is still "
+                    f"{status['status']}. Please continue waiting... this may take a few moments."
+                )
             )
-            while status["status"].startswith("running") or status["status"] == "accepted":
+            while (
+                status["status"].startswith("running") or status["status"] == "accepted"
+            ):
                 sys.stdout.write(".")
                 sys.stdout.flush()
                 # Requesting the status too often can result in a 500 error.

icepyx/tests/unit/test_orders.py

@@ -26,6 +26,7 @@ def test_status_subset():
     assert result == {"status": "processing"}
     mock_harmony_client.check_order_status.assert_called_once_with(123)
 
+
 def test_status_non_subset():
     # Mock the harmony_client (not used in this case)
     mock_harmony_client = Mock()
@@ -45,6 +46,7 @@ def test_status_non_subset():
     assert result == {"status": "complete"}
     mock_harmony_client.check_order_status.assert_not_called()
 
+
 def test_download_subset():
     # Mock the harmony_client
     mock_harmony_client = Mock()
@@ -74,7 +76,10 @@ def test_download_non_subset(mock_earthaccess_download):
     mock_harmony_client = Mock()
 
     # Mock the earthaccess.download method to return specific file paths
-    mock_earthaccess_download.return_value = ["/fake/path/granule1.nc", "/fake/path/granule2.nc"]
+    mock_earthaccess_download.return_value = [
+        "/fake/path/granule1.nc",
+        "/fake/path/granule2.nc",
+    ]
 
     # Create a DataOrder instance with type="download"
     order = DataOrder(
@@ -93,4 +98,3 @@ def test_download_non_subset(mock_earthaccess_download):
         ["granule1", "granule2"], local_path="/fake/path"
     )
     mock_harmony_client.download_granules.assert_not_called()
-    

icepyx/tests/unit/test_spatial.py

@@ -1447,6 +1447,7 @@ def poly():
     ]
     return spat.Spatial(coords)
 
+
 @pytest.mark.skip(reason="Skipping this test for now, it fails with current polygons")
 def test_polygon_fmt(poly):
     obs = poly.fmt_for_CMR()