Merge pull request #49 from rodekruis/dev

Dev

Author: p-phung

retrievalpipeline/config/somalia.py

@@ -78,7 +78,7 @@ def drought_data_request(self) -> EcmwfDataRequest:
                         self.datetime_config.recent_start_year, self.datetime_config.recent_end_year + 1
                     )
                 ],
-                "month": ["01", "07"],
+                "month": [f"{m:02d}" for m in range(1, 13)],
                 "time": ["00:00"],
                 "data_format": "netcdf",
                 # "download_format": "zip",
@@ -97,7 +97,7 @@ def extreme_heat_data_request(self) -> EcmwfDataRequest:
                 "year": [
                     f"{year}"
                     for year in range(
-                        self.datetime_config.recent_start_year, self.datetime_config.baseline_end_year + 1
+                        self.datetime_config.recent_start_year, self.datetime_config.recent_end_year + 1
                     )
                 ],
                 "month": [f"{m:02d}" for m in range(1, 13)],

retrievalpipeline/extract/extractors/api_json_extractor.py

@@ -1,3 +1,4 @@
+import time
 from collections.abc import Iterable, Iterator
 from typing import Any
 
@@ -46,13 +47,26 @@ def get_data(
         Returns:
             list of pydantic models (of type specified in the initiation).
         """
-        try:
-            response = self.get_response(api_url, params, headers)
-            json_data = self.get_data_from_response(response)
-        except ApiResponseError as err:
-            logger.warning(err)
-            return []
-        return list(self.validate_and_parse(json_data))
+        max_retries = 5
+        sleep_seconds = 10
+
+        for attempt in range(1, max_retries + 1):
+            try:
+                response = self.get_response(api_url, params, headers)
+                json_data = self.get_data_from_response(response)
+                return list(self.validate_and_parse(json_data))
+            except ApiResponseError as err:
+                if attempt < max_retries:
+                    logger.warning(
+                        f"""Attempt {attempt}/{max_retries} failed for {api_url}: {err}.
+                        Retrying in {sleep_seconds} seconds""",
+                    )
+                    time.sleep(sleep_seconds)
+                    continue
+                else:
+                    logger.warning(f"All {max_retries} attempts failed for {api_url}: {err}")
+
+        return []
 
     def get_response(
         self, api_url: str, params: dict[str, str] | None = None, headers: dict[str, str] | None = None

retrievalpipeline/extract/extractors/ecmwf.py

@@ -74,7 +74,7 @@ async def extract(self) -> None:
             logger.info(f"{self.name}: unzipped {zip_file.name}")
 
     async def _wait_on_results(
-        self, remote: Remote, offset: float = 0.1, factor: float = 1.3, max_waiting_time: int = 100
+        self, remote: Remote, offset: float = 0.1, factor: float = 1.3, max_waiting_time: int = 3600
     ) -> None:
         """Wait till job is done.
 

retrievalpipeline/extract/extractors/ipc.py

@@ -188,12 +188,11 @@ def get_available_data(self, data_type: str) -> gpd.GeoDataFrame:
     def get_metadata(self, anl_id: str) -> pd.DataFrame:
         """Collect metadata from IPC based on analysis ID."""
         metadata_frames = []
-        for iso2 in self.country_codes_iso2:
-            try:
-                metadata = self.get_available_metadata_iso2(iso2, self.types, [anl_id])
-                metadata_frames.append(metadata)
-            except Exception as e:
-                logger.warning(f"Failed to get metadata for {iso2}: {e}")
+        try:
+            metadata = self.get_available_metadata_iso2(self.country_codes_iso2, self.types, [anl_id])
+            metadata_frames.append(metadata)
+        except Exception as e:
+            logger.warning(f"Failed to get metadata for {self.country_codes_iso2}: {e}")
 
         return pd.concat(metadata_frames, ignore_index=True)
 

run_pipeline.py

@@ -81,15 +81,15 @@ def construct_pipeline(run_id: str, iso3: str) -> ETLPipeline:
 
     extractors = [
         ChirpsExtractor(
-            name="Drought Extractor",
+            name="Flood Extractor",
             storage=storage,
             path_to_output=f"{path_to_bronze_global}/chirps/rainfall/",
             data_request=config.chirps_data_request,
         ),
         EcmwfExtractor(
             name="Drought Extractor",
             storage=storage,
-            path_to_output=f"{path_to_bronze}/ecmwf/rainfall/",
+            path_to_output=f"{path_to_bronze}/ecmwf/rainfall/drought.zip",
             data_request=config.drought_data_request,
         ),
         EcmwfExtractor(