Added retry to S3 steaming in hand_fim_processing, commented out coas…

…tal services in initialize_pipeline
NOAA-OWP · Apr 19, 2024 · 3442e44 · 3442e44
1 parent 81468bb
commit 3442e44
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 23 deletions.
diff --git a/Core/LAMBDA/viz_functions/image_based/viz_hand_fim_processing/lambda_function.py b/Core/LAMBDA/viz_functions/image_based/viz_hand_fim_processing/lambda_function.py
@@ -1,4 +1,5 @@
-# import boto3
+import boto3
+from botocore.exceptions import ResponseStreamingError
 import rasterio
 import numpy as np
 import pandas as pd
@@ -15,8 +16,6 @@
 FIM_PREFIX = os.environ['FIM_PREFIX']
 FIM_VERSION = re.findall("[/_]?(\d*_\d*_\d*_\d*)/?", FIM_PREFIX)[0]
 
-# s3 = boto3.client("s3")
-
 class HANDDatasetReadError(Exception):
     """ my custom exception class """
 
@@ -437,18 +436,18 @@ def process(window):
 
     return df_final
 
-def s3_csv_to_df(bucket, key):
-    # basename = os.path.basename(key)
-    # local_file = f"/tmp/{basename}"
-
-    # print(f"Downloading {key} from {bucket}")
-    # s3.download_file(bucket, key, local_file)
-    # df = pd.read_csv(local_file)
-    # os.remove(local_file)
-
-    print(f"Reading {key} from {bucket} into DataFrame")
-    df = wr.s3.read_csv(path=f"s3://{bucket}/{key}")
-    print("DataFrame creation Successful")
+def s3_csv_to_df(bucket, key):    
+    # Allow retrying a few times before failing
+    for i in range(5):
+        try:
+            # Read S3 csv file into Pandas DataFrame
+            print(f"Reading {key} from {bucket} into DataFrame")
+            df = wr.s3.read_csv(path=f"s3://{bucket}/{key}")
+            print("DataFrame creation Successful")
+        except ResponseStreamingError:
+            if i == 4: print("Failed to read from S3")
+            continue
+        break
 
     return df
 

diff --git a/Core/LAMBDA/viz_functions/viz_initialize_pipeline/lambda_function.py b/Core/LAMBDA/viz_functions/viz_initialize_pipeline/lambda_function.py
@@ -73,14 +73,14 @@ def lambda_handler(event, context):
                                         "medium_range_blend.forcing.f240.alaska.nc",
 
                                         ## Coastal ##
-                                        "analysis_assim_coastal.total_water.tm00.atlgulf.nc",
-                                        "analysis_assim_coastal.total_water.tm00.hawaii.nc",
-                                        "analysis_assim_coastal.total_water.tm00.puertorico.nc",
-                                        "medium_range_coastal.total_water.f240.atlgulf.nc",
-                                        "medium_range_blend_coastal.total_water.f240.atlgulf.nc",
-                                        "short_range_coastal.total_water.f018.atlgulf.nc",
-                                        "short_range_coastal.total_water.f048.puertorico.nc",
-                                        "short_range_coastal.total_water.f048.hawaii.nc"
+                                        # "analysis_assim_coastal.total_water.tm00.atlgulf.nc",
+                                        # "analysis_assim_coastal.total_water.tm00.hawaii.nc",
+                                        # "analysis_assim_coastal.total_water.tm00.puertorico.nc",
+                                        # "medium_range_coastal.total_water.f240.atlgulf.nc",
+                                        # "medium_range_blend_coastal.total_water.f240.atlgulf.nc",
+                                        # "short_range_coastal.total_water.f018.atlgulf.nc",
+                                        # "short_range_coastal.total_water.f048.puertorico.nc",
+                                        # "short_range_coastal.total_water.f048.hawaii.nc"
                                         ]
         s3_event = json.loads(event.get('Records')[0].get('Sns').get('Message'))
         if s3_event.get('Records')[0].get('s3').get('object').get('key'):