[GOES16] Sending images to Rionowcast API

pipelines/rj_cor/meteorologia/precipitacao_alertario/tasks.py

@@ -34,7 +34,7 @@
 
 @task(
     nout=2,
-    max_retries=constants.TASK_MAX_RETRIES.value,
+    max_retries=10,  # constants.TASK_MAX_RETRIES.value,
     retry_delay=timedelta(seconds=constants.TASK_RETRY_DELAY.value),
 )
 def tratar_dados(

pipelines/rj_cor/meteorologia/satelite/satellite_utils.py

@@ -47,7 +47,6 @@
 # Required Libraries
 # ====================================================================
 
-import base64
 import datetime
 import os
 import shutil
@@ -69,7 +68,12 @@
 import xarray as xr
 
 from pipelines.rj_cor.meteorologia.satelite.remap import remap
-from pipelines.utils.utils import get_credentials_from_env, list_blobs_with_prefix, log
+from pipelines.utils.utils import (
+    get_credentials_from_env,
+    get_vault_secret,
+    list_blobs_with_prefix,
+    log,
+)
 
 
 def get_blob_with_prefix(bucket_name: str, prefix: str, mode: str = "prod") -> str:
@@ -589,14 +593,37 @@
     longitudes = list(matrix_temp.columns)
     latitudes = list(matrix_temp.index)
 
-    log("Convert to xr dataarray")
+    log("Convert df to xr data array")
     data_array = xr.DataArray(
         matrix, dims=("lat", "lon"), coords={"lon": longitudes, "lat": latitudes}
     )
-    log("end")
+    log("Finished converting df to data array")
+
     return data_array
 
 
+# pylint: disable=dangerous-default-value
+def get_point_value(
+    data_array: xr.DataArray, selected_point: list = [-22.89980, -43.35546]
+) -> float:
+    """
+    Find the nearest point on data_array from the selected_point and return its value
+    """
+
+    # Find the nearest index of latitude and longitude from selected_point
+    lat_idx = (data_array["lat"] - selected_point[0]).argmin().values
+    lon_idx = (data_array["lon"] - selected_point[1]).argmin().values
+
+    # Get the correspondent value of this point
+    point_value = data_array.isel(lat=lat_idx, lon=lon_idx).values
+    log(
+        "\nThe value of the selected point is {point_value}. It will be replace by 0 if is nan.\n"
+    )
+    point_value = 0 if np.isnan(point_value) else float(point_value)
+
+    return point_value
+
+
 # pylint: disable=unused-variable
 def create_and_save_image(data: xr.DataArray, info: dict, variable) -> Path:
     """
@@ -616,13 +643,21 @@
     # colormap = "gray_r" # White to black for IR channels
 
     # Plot the image
-    img = axis.imshow(data, origin="upper", extent=img_extent, cmap=colormap, alpha=0.8)
+    img = axis.imshow(
+        data,
+        origin="upper",
+        extent=img_extent,
+        cmap=colormap,
+        alpha=0.8,
+        vmin=info["vmin"],
+        vmax=info["vmax"],
+    )
 
     # # Find shapefile file "Limite_Bairros_RJ.shp" across the entire file system
     # for root, dirs, files in os.walk(os.sep):
     #     if "Limite_Bairros_RJ.shp" in files:
     #         log(f"[DEBUG] ROOT {root}")
-    #         shapefile_dir = root
+    #         shapefile_dir = Path(root)
     #         break
     # else:
     #     print("File not found.")
@@ -684,26 +719,40 @@
         output_image_path.mkdir(parents=True, exist_ok=True)
 
     plt.savefig(save_image_path, bbox_inches="tight", pad_inches=0, dpi=300)
-    log("\n Ended saving image")
+    log(f"\n Ended saving image on {save_image_path}")
     return save_image_path
 
 
-def upload_image_to_api(info: dict, save_image_path: Path):
+# def upload_image_to_api(info: dict, save_image_path: Path):
+def upload_image_to_api(var: str, save_image_path: Path, point_value: float):
     """
-    Upload image to api
+    Upload image and point value to API.
     """
-    username = "your-username"
-    password = "your-password"
+    # We need to change this variable so it can be posted on API
+    var = "cp" if var == "cape" else var
+
+    log("Getting API url")
+    url_secret = get_vault_secret("rionowcast")["data"]
+    log(f"urlsecret1 {url_secret}")
+    url_secret = url_secret["url_api_satellite_products"]
+    log(f"urlsecret2 {url_secret}")
+    api_url = f"{url_secret}/{var.lower()}"
+    # api_url = f"http://127.0.0.1:5000/upload/{var.lower()}"
+    log(
+        f"\n Sending image {save_image_path} to API: {api_url} with value {point_value}\n"
+    )
 
-    image = base64.b64encode(open(save_image_path, "rb").read()).decode()
+    payload = {"value": point_value}
 
-    response = requests.post(
-        "https://api.example.com/upload-image",
-        data={"image": image, "timestamp": info["datetime_save"]},
-        auth=(username, password),
-    )
+    # Convert from Path to string
+    save_image_path = str(save_image_path)
+
+    with open(save_image_path, "rb") as image_file:
+        files = {"image": (save_image_path, image_file, "image/jpeg")}
+        response = requests.post(api_url, data=payload, files=files)
 
     if response.status_code == 200:
-        print("Image sent to API")
+        print("Finished the request successful!")
+        print(response.json())
     else:
-        print("Problem senting imagem to API")
+        print(f"Error: {response.status_code}, {response.text}")

pipelines/rj_cor/meteorologia/satelite/tasks.py

@@ -24,9 +24,11 @@
     get_files_from_aws,
     get_files_from_gcp,
     get_info,
+    get_point_value,
     get_variable_values,
     remap_g16,
     save_data_in_file,
+    upload_image_to_api,
 )
 from pipelines.utils.utils import log
 
@@ -225,7 +227,7 @@ def save_data(info: dict, mode_redis: str = "prod") -> Union[str, Path]:
 @task
 def create_image_and_upload_to_api(info: dict, output_filepath: Path):
     """
-    Create image from dataframe and send it to API
+    Create image from dataframe, get the value of a point on the image and send these to API.
     """
 
     dfr = pd.read_csv(output_filepath)
@@ -237,12 +239,15 @@ def create_image_and_upload_to_api(info: dict, output_filepath: Path):
 
         var = var.lower()
         data_array = get_variable_values(dfr, var)
+        point_value = get_point_value(data_array)
 
         # Get the pixel values
         data = data_array.data[:]
-        log(f"\n[DEBUG] data {data}")
+        log(f"\n[DEBUG] {var} data \n{data}")
+        log(f"\nmax value: {data.max()} min value: {data.min()}")
         save_image_path = create_and_save_image(data, info, var)
+
         log(f"\nStart uploading image for variable {var} on API\n")
-        # upload_image_to_api(info, save_image_path)
+        upload_image_to_api(var, save_image_path, point_value)
         log(save_image_path)
         log(f"\nEnd uploading image for variable {var} on API\n")