[WIP] satelite backfill

pipelines/rj_cor/meteorologia/satelite/flows.py

@@ -93,6 +93,7 @@
         table_id_rr,
         "prod",
         redis_files_rr_updated,
+        keep_last=200,
         wait=path_rr,
     )
 
@@ -130,6 +131,7 @@
         table_id_tpw,
         "prod",
         redis_files_tpw_updated,
+        keep_last=200,
         wait=path_tpw,
     )
 
@@ -169,6 +171,7 @@
         table_id_cmip,
         "prod",
         redis_files_cmip_updated,
+        keep_last=200,
         wait=path_cmip,
     )
 

pipelines/rj_cor/meteorologia/satelite/remap.py

@@ -3,7 +3,6 @@
 """
 Converte coordenada X,Y para latlon
 """
-import time as t
 
 import netCDF4 as nc
 import numpy as np
@@ -106,9 +105,9 @@ def remap(
 
     # Perform the projection/resampling
 
-    print("Remapping", path)
+    # print("Remapping", path)
 
-    start = t.time()
+    # start = t.time()
 
     gdal.ReprojectImage(
         raw,
@@ -119,7 +118,7 @@ def remap(
         options=["NUM_THREADS=ALL_CPUS"],
     )
 
-    print("- finished! Time:", t.time() - start, "seconds")
+    # print("- finished! Time:", t.time() - start, "seconds")
 
     # Close file
     raw = None

pipelines/rj_cor/meteorologia/satelite/satellite_utils.py

@@ -1,5 +1,6 @@
 # -*- coding: utf-8 -*-
-# pylint: disable=too-many-locals
+# pylint: disable=too-many-locals,R0913
+# flake8: noqa
 """
 Funções úteis no tratamento de dados de satélite
 """
@@ -64,6 +65,7 @@
 import datetime
 import os
 from pathlib import Path
+import re
 from typing import Tuple, Union
 
 from google.cloud import storage
@@ -111,6 +113,7 @@ def download_blob(
 
     credentials = get_credentials_from_env(mode=mode)
     storage_client = storage.Client(credentials=credentials)
+    # storage_client = storage.Client()
 
     bucket = storage_client.bucket(bucket_name)
 
@@ -128,10 +131,11 @@ def converte_timezone(datetime_save: str) -> str:
     Recebe o formato de data hora em 'YYYYMMDD HHmm' no UTC e
     retorna no mesmo formato no horário São Paulo
     """
-    log(f">>>>>>> {datetime_save}")
-    datahora = pendulum.from_format(datetime_save, "YYYYMMDD HHmm")
+    log(f">>>>>>> datetime_save {datetime_save}")
+    datahora = pendulum.from_format(datetime_save, "YYYYMMDD HHmmss")
+    log(f">>>>>>> datahora {datahora}")
     datahora = datahora.in_tz("America/Sao_Paulo")
-    return datahora.format("YYYYMMDD HHmm")
+    return datahora.format("YYYYMMDD HHmmss")
 
 
 def extract_julian_day_and_hour_from_filename(filename: str):
@@ -154,7 +158,7 @@ def extract_julian_day_and_hour_from_filename(filename: str):
     julian_day = int(start[4:7])
 
     # Time (UTC) as string
-    hour_utc = start[7:11]
+    hour_utc = start[7:13]
 
     # Time of the start of the Scan
     # time = start[7:9] + ":" + start[9:11] + ":" + start[11:13] + " UTC"
@@ -207,7 +211,7 @@ def get_info(path: str) -> Tuple[dict, str]:
     if procura_m == -1:
         procura_m = path.find("-M4")
     product = path[path.find("L2-") + 3 : procura_m]
-    print(product)
+    # print(product)
 
     # Nem todos os produtos foram adicionados no dicionário de características
     # dos produtos. Olhar arquivo original caso o produto não estaja aqui
@@ -367,8 +371,10 @@ def get_info(path: str) -> Tuple[dict, str]:
 
     if variable == "CMI":
         # Search for the GOES-16 channel in the file name
+        regex = "-M\\dC\\d"  # noqa: W605
+        find_expression = re.findall(regex, path)[0]
         product_caracteristics["band"] = int(
-            (path[path.find("M6C") + 3 : path.find("_G16")])
+            (path[path.find(find_expression) + 4 : path.find("_G16")])
         )
     else:
         product_caracteristics["band"] = np.nan
@@ -402,6 +408,7 @@ def remap_g16(
     resolution: int,
     variable: str,
     datetime_save: str,
+    mode_redis: str = "prod",
 ):
     """
     the GOES-16 image is reprojected to the rectangular projection in the extent region
@@ -436,7 +443,7 @@ def remap_g16(
     )
 
     tif_path = os.path.join(
-        os.getcwd(), "data", "satelite", variable, "temp", partitions
+        os.getcwd(), mode_redis, "data", "satelite", variable, "temp", partitions
     )
 
     if not os.path.exists(tif_path):
@@ -519,7 +526,7 @@ def treat_data(
 
 
 def save_data_in_file(
-    variable: str, datetime_save: str, file_path: str
+    variable: str, datetime_save: str, file_path: str, mode_redis: str = "prod"
 ) -> Union[str, Path]:
     """
     Save data in parquet
@@ -539,7 +546,14 @@ def save_data_in_file(
     )
 
     tif_data = os.path.join(
-        os.getcwd(), "data", "satelite", variable, "temp", partitions, "dados.tif"
+        os.getcwd(),
+        mode_redis,
+        "data",
+        "satelite",
+        variable,
+        "temp",
+        partitions,
+        "dados.tif",
     )
 
     data = xr.open_dataset(tif_data, engine="rasterio")
@@ -561,14 +575,20 @@ def save_data_in_file(
     )
 
     # cria pasta de partições se elas não existem
-    output_path = os.path.join(os.getcwd(), "data", "satelite", variable, "output")
+    output_path = os.path.join(
+        os.getcwd(), mode_redis, "data", "satelite", variable, "output"
+    )
     parquet_path = os.path.join(output_path, partitions)
 
     if not os.path.exists(parquet_path):
         os.makedirs(parquet_path)
 
+    data["horario"] = pendulum.from_format(
+        datetime_save, "YYYYMMDD HHmmss"
+    ).to_time_string()
+    # log(f">>>>> data head {data.head()}")
     # Fixa ordem das colunas
-    data = data[["longitude", "latitude", variable.lower()]]
+    data = data[["longitude", "latitude", "horario", variable.lower()]]
 
     # salva em csv
     filename = file_path.split("/")[-1].replace(".nc", "")
@@ -579,7 +599,7 @@ def save_data_in_file(
     return output_path
 
 
-def main(path: Union[str, Path]):
+def main(path: Union[str, Path], mode_redis: str = "prod"):
     """
     Função principal para converter dados x,y em lon,lat
     """
@@ -620,6 +640,7 @@ def main(path: Union[str, Path]):
         resolution,
         product_caracteristics["variable"],
         datetime_save,
+        mode_redis,
     )
 
     info = {

pipelines/rj_cor/meteorologia/satelite/tasks.py

@@ -80,6 +80,7 @@ def download(
     ref_filename: str = None,
     redis_files: list = [],
     wait=None,
+    mode_redis: str = "prod",
 ) -> Union[str, Path]:
     """
     Acessa o S3 e faz o download do primeiro arquivo da data-hora especificada
@@ -121,14 +122,16 @@ def download(
         skip = Skipped("No available files on API")
         raise ENDRUN(state=skip)
 
-    base_path = os.path.join(os.getcwd(), "data", "satelite", variavel[:-1], "input")
+    base_path = os.path.join(
+        os.getcwd(), mode_redis, "data", "satelite", variavel[:-1], "input"
+    )
 
     if not os.path.exists(base_path):
         os.makedirs(base_path)
 
     # Seleciona primeiro arquivo que não tem o nome salvo no redis
-    log(f"\n\n[DEBUG]: available files on API: {path_files}")
-    log(f"\n\n[DEBUG]: filenames already saved on redis_files: {redis_files}")
+    # log(f"\n\n[DEBUG]: available files on API: {path_files}")
+    # log(f"\n\n[DEBUG]: filenames already saved on redis_files: {redis_files}")
     log(f"\n\n[DEBUG]: ref_filename: {ref_filename}")
 
     # keep only ref_filename if it exists
@@ -145,12 +148,13 @@ def download(
     download_file = None
     for path_file in path_files:
         filename = path_file.split("/")[-1]
+        log(f"\n\n[DEBUG]: {filename} check if is in redis")
         if filename not in redis_files:
             log(f"\n\n[DEBUG]: {filename} not in redis")
             redis_files.append(filename)
             path_filename = os.path.join(base_path, filename)
             download_file = path_file
-            log(f"[DEBUG]: filename to be append on redis_files: {redis_files}")
+            # log(f"[DEBUG]: filename to be append on redis_files: {redis_files}")
             break
 
     # Skip task if there is no new file
@@ -174,24 +178,24 @@ def download(
 
 
 @task
-def tratar_dados(filename: str) -> dict:
+def tratar_dados(filename: str, mode_redis: str = "prod") -> dict:
     """
     Converte coordenadas X, Y para latlon e recorta área
     """
     log(f"\n>>>> Started treating file: {filename}")
-    grid, goes16_extent, info = main(filename)
+    grid, goes16_extent, info = main(filename, mode_redis)
     del grid, goes16_extent
     return info
 
 
 @task
-def save_data(info: dict, file_path: str) -> Union[str, Path]:
+def save_data(info: dict, file_path: str, mode_redis: str = "prod") -> Union[str, Path]:
     """
     Convert tif data to csv
     """
 
     variable = info["variable"]
     datetime_save = info["datetime_save"]
     print(f"Saving {variable} in parquet")
-    output_path = save_data_in_file(variable, datetime_save, file_path)
+    output_path = save_data_in_file(variable, datetime_save, file_path, mode_redis)
     return output_path

pipelines/rj_cor/meteorologia/satelite_backfill/__init__.py

@@ -0,0 +1,8 @@
+# -*- coding: utf-8 -*-
+"""
+Prefect flows for workshop project
+"""
+###############################################################################
+# Automatically managed, please do not touch
+###############################################################################
+from .flows import *